[FEAT#510] normal/low crawl 토픽 Redis 버퍼링 + 주기적 Kafka backlog drain

[juhy0987]

연관 이슈

• Closes [FEATURE] normal/low crawl 토픽 Redis 버퍼링 + 주기적 Kafka backlog drain #510

구현 내용

PR #509 머지 후 15분 라이브에서 관측된 kafka backlog exceeds threshold, throttling publish 33건 drop 해소를 위한 Redis 버퍼 + 주기적 drainer 도입. high priority 는 영향 없음.

구조

[Publisher.Publish(job)]
   ↓ producer.Publish(msg)
[BufferingProducer]  (queue.Producer 데코레이터)
   ├── topic ∈ {normal, low} → JobBuffer.EnqueueJob (Redis LIST)
   └── 그 외 (high / 기타) → underlying.Publish

[BufferDrainer goroutine]
   매 tick 마다:
   ├── BacklogChecker.Backlog(topic, group)
   ├── available = TargetBacklog - lag
   └── JobBuffer.DrainJobs(label, n) → underlying.PublishBatch

신규 컴포넌트 (5 commits)

1. pkg/redis/job_buffer.go — Redis LIST 기반 EnqueueJob/DrainJobs/JobBufferLen
   • LPUSH (head enqueue) + RPOP COUNT (tail FIFO drain, Redis 6.2+) + LTRIM (MaxLen 보장)
   • 기존 retry_queue.go (ZSET) 와 같은 pkg/redis/Client 메소드 패턴
2. pkg/queue/buffering_producer.go + buffered_message.go — 데코레이터 + 직렬화
   • BufferingProducer: topic 기반 routing, fallback 안전망, batch 지원
   • bufferedMessage JSON 직렬화 (BufferedAt 포함 — drainer 의 체류 시간 metric 용)
   • NoopJobBuffer / Underlying() 노출 — wiring/drainer 안전망
3. internal/scheduler/buffer_drainer.go — goroutine
   • 매 tick: BacklogChecker → available 계산 → DrainJobs → PublishBatch
   • publish 실패 시 재적재 (best-effort)
   • 부팅 직후 1회 즉시 drain (이전 세션 잔존물 회복)
   • Start/Stop graceful shutdown
4. pkg/config/processor/job_buffer.go — 6개 env + parse.PositiveInt64 신규 helper
5. cmd/issuetracker/main.go — Redis 클라이언트 위쪽으로 이동 + wiring + shutdown 체인

환경 변수 (.env.example 갱신)

Key	Default	Description
PUBLISHER_REDIS_BUFFER_ENABLED	false	기능 활성화 (opt-in)
PUBLISHER_REDIS_BUFFER_DRAIN_INTERVAL	30s	drainer tick 주기
PUBLISHER_REDIS_BUFFER_TARGET_BACKLOG	3000	유지하려는 Kafka lag 상한 (SCHEDULER_MAX_BACKLOG 5000 의 60%)
PUBLISHER_REDIS_BUFFER_DRAIN_BATCH	100	한 tick 당 priority 별 최대 drain 수
PUBLISHER_REDIS_BUFFER_MAX_LEN	100000	LIST LTRIM cap (0=무제한)
PUBLISHER_REDIS_BUFFER_CHECK_TIMEOUT	5s	Backlog() 호출 deadline

단위 테스트 (19개)

• test/pkg/redis/job_buffer_test.go (5): FIFO / MaxLen LTRIM / empty / validation / Len — Redis 환경에서 통합 검증 (NOAUTH 시 skip)
• test/pkg/queue/buffering_producer_test.go (9): normal/low/high routing / non-crawl topic 우회 / buffer fail fallback / batch mixed / nil buffer Noop / Underlying / Encode-Decode round-trip
• test/internal/scheduler/buffer_drainer_test.go (5): drain available / skip on threshold / re-enqueue on publish fail / idle empty / nil deps validation

Backward compatibility

• PUBLISHER_REDIS_BUFFER_ENABLED=false (default) → BufferingProducer wiring 자체 skip, raw KafkaProducer 사용 — 기존 동작 100% 보존
• Redis 미연결 + ENABLED=true → WARN 로그 후 직접 publish fallback (graceful degrade)
• 기존 scheduler.BacklogThrottler 는 그대로 유지 — drainer 가 죽거나 buffer 비정상 비활성화 시 fail-safe

IngestionLock 와의 관계

기존 internal/locks/ingestion_lock.go 의 24h IngestionLock 은 publisher 가 SETNX 로 잡음. buffer 진입 시점도 동일 → buffer 가 dedup 의미 보존, 단지 staging 영역만 추가.

Graceful shutdown 동작

• sched.Stop() 후 bufferDrainer.Stop() 호출
• 진행 중인 drain cycle 은 완료까지 대기
• 잔존 buffer URL 들은 Redis 에 그대로 보존 → 다음 부팅 시 drainAll (부팅 직후 1회) 가 자연 회복 (이슈 [FEATURE] normal/low crawl 토픽 Redis 버퍼링 + 주기적 Kafka backlog drain #510 본문 옵션 A 채택)

CI / 머지 게이트 점검

변경 영향 범위

• 영향 패키지/모듈:
  • pkg/redis/ (Client 메소드 추가)
  • pkg/queue/ (BufferingProducer 데코레이터 + JobBuffer interface)
  • internal/scheduler/ (BufferDrainer goroutine)
  • pkg/config/processor/ (JobBufferConfig + 6 env)
  • pkg/config/internal/parse/ (PositiveInt64 helper)
  • cmd/issuetracker/main.go (wiring + Redis 초기화 위치 이동)
• 위험도: Medium — opt-in 기본값이라 비활성 시 회귀 위험 0. 활성화 시 publish 경로에 Redis 매개됨 → Redis 장애 = fallback 으로 직접 publish (정상 동작 유지). main.go 의 Redis 초기화 위치 이동이 가장 큰 변경이나 동일 변수 (redisClientShared) 와 동일 fallback 로직 사용.

Required Status Checks

• 통과 확인 대상:
  • Commit Lint
  • PR Title Lint
  • Linked Issue Check
  • Format Check
  • Build
  • Test
  • Lint

롤백 계획

• 기능 토글 only: PUBLISHER_REDIS_BUFFER_ENABLED=false 설정만으로 즉시 비활성화 — PR revert 불필요
• 영구 원복 필요 시 본 PR revert — main.go Redis 초기화 위치 변경도 함께 원복되나 동일 동작 유지

🤖 Generated with Claude Code

Summary by CodeRabbit

Release Notes

• New Features

  • Added optional Redis-backed buffering for normal/low-priority crawl jobs with configurable drain intervals, batch sizes, and target backlog thresholds (disabled by default). System gracefully falls back to direct publishing when Redis is unavailable.

• Chores

  • Added Redis publisher buffer configuration variables to .env.example for buffer tuning.

[coderabbitai]

Warning

Rate limit exceeded

@juhy0987 has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 44 minutes and 59 seconds before requesting another review.

You’ve run out of usage credits. Purchase more in the billing tab.

⌛ How to resolve this issue?

After the wait time has elapsed, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

We recommend that you space out your commits to avoid hitting the rate limit.

🚦 How do rate limits work?

CodeRabbit enforces hourly rate limits for each developer per organization.

Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout.

Please see our FAQ for further information.

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: ee1726e3-d907-4e5a-8da5-cfca985d5a36

📥 Commits

Reviewing files that changed from the base of the PR and between 21d200c and a0b4009.

📒 Files selected for processing (8)

• cmd/issuetracker/main.go
• internal/scheduler/buffer_drainer.go
• pkg/queue/buffering_producer.go
• pkg/redis/job_buffer.go
• test/internal/scheduler/buffer_drainer_test.go
• test/pkg/config/config_test.go
• test/pkg/queue/buffering_producer_test.go
• test/pkg/redis/job_buffer_test.go

📝 Walkthrough

Walkthrough

This PR implements Redis-backed buffering for normal/low-priority crawl job publishing, with a periodic drainer that publishes buffered messages back to Kafka when consumer backlog is below a configured threshold. The feature is opt-in via environment variables and gracefully degrades to direct publishing when Redis is unavailable or buffering is disabled.

Changes

Redis Job Buffering & Periodic Draining

Layer / File(s)	Summary
Configuration & Parsing .env.example, pkg/config/internal/parse/parse.go, pkg/config/processor/job_buffer.go	Environment variable schema, typed parsing helpers (PositiveInt64), and JobBufferConfig with defaults and file loading. Buffering disabled by default with tunable drain interval, target backlog, batch size, and buffer length limits.
Producer Buffering & Message Serialization pkg/queue/buffering_producer.go, pkg/queue/buffered_message.go, test/pkg/queue/buffering_producer_test.go	JobBuffer interface, NoopJobBuffer opt-out fallback, BufferingProducer decorator routing normal/low-priority messages into Redis while passing high-priority and non-crawl topics directly to Kafka. Buffered-message JSON serialization with timestamp tracking. Tests verify topic-based routing, fallback on enqueue failure, batch handling, and round-trip encode/decode.
Redis Job Buffer Storage pkg/redis/job_buffer.go, test/pkg/redis/job_buffer_test.go	Redis LIST-backed JobBuffer implementation with LPUSH enqueueing, RPOP COUNT batch draining, and LTRIM size capping. Tests cover FIFO ordering, max-length trimming, empty-buffer handling, and input validation.
Buffer Drainer Scheduler internal/scheduler/buffer_drainer.go, test/internal/scheduler/buffer_drainer_test.go	Periodic scheduler that checks Kafka consumer-group backlog, drains buffered jobs up to available = targetBacklog - currentBacklog, and publishes them in batches. Re-enqueues on publish failure, performs an immediate drain cycle on startup, and gracefully stops via context cancellation. Tests validate backlog-throttled draining, re-enqueuing on failure, idle-when-empty behavior, and constructor validation.
Main Application Integration cmd/issuetracker/main.go	Initializes a shared Redis client early in startup for coordinated multi-component use. Conditionally wraps the Kafka producer with BufferingProducer when buffering is enabled and Redis is connected, with WARN fallback to direct publishing if buffering is enabled but Redis is unavailable. Refactors ProcessingLock, IngestionLock, and delayed retry scheduler wiring to use the shared Redis client. Starts BufferDrainer during initialization and stops it during graceful shutdown after scheduler.Stop() to flush buffered messages.

Sequence Diagrams

sequenceDiagram
  participant Scheduler
  participant BufferingProducer
  participant RedisBuffer
  participant KafkaProducer
  Scheduler->>BufferingProducer: Publish(message)
  alt Normal or Low Priority
    BufferingProducer->>BufferingProducer: encodeBufferedMessage()
    BufferingProducer->>RedisBuffer: EnqueueJob(topic, payload, maxLen)
    RedisBuffer-->>BufferingProducer: success or error
    alt Enqueue Fails
      BufferingProducer->>KafkaProducer: Publish(direct fallback)
    end
  else High Priority or Non-Crawl Topic
    BufferingProducer->>KafkaProducer: Publish(direct)
  end
  KafkaProducer-->>BufferingProducer: success or error

Loading

sequenceDiagram
  participant BufferDrainer
  participant BacklogChecker
  participant RedisBuffer
  participant KafkaProducer
  loop Every DrainInterval
    BufferDrainer->>BacklogChecker: Backlog(groupID)
    BacklogChecker-->>BufferDrainer: currentLag
    alt Lag >= TargetBacklog
      BufferDrainer->>BufferDrainer: Skip drain (throttled)
    else Lag < TargetBacklog
      BufferDrainer->>RedisBuffer: DrainJobs(priority, available)
      RedisBuffer-->>BufferDrainer: payloads
      BufferDrainer->>BufferDrainer: DecodeBufferedMessage(payloads)
      BufferDrainer->>KafkaProducer: PublishBatch(messages)
      alt Publish Success
        BufferDrainer->>BufferDrainer: Log drain count
      else Publish Failure
        BufferDrainer->>RedisBuffer: EnqueueJob(failed payloads)
        RedisBuffer-->>BufferDrainer: re-enqueued
      end
    end
  end

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related PRs

• EinSofINTEREST/IssueTracker#400: Refactors the publisher facade and delayed retry scheduler lifecycle management that this PR reuses in its main.go wiring and Redis client initialization.
• EinSofINTEREST/IssueTracker#179: Introduces Redis IngestionLock for publisher dedup, which this PR integrates into the shared Redis client initialization.

Suggested labels

enhancement

Poem

🐰 Buffering like a rabbit beneath the earth,
Normal and low-priority jobs find gentle rebirth,
When Kafka backlog grows, our drainer takes the lead,
Batching up the hoppers, fulfilling every need. 🌾✨

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The PR title is in Korean and references issue #510, describing the implementation of Redis buffering for normal/low crawl topics with periodic Kafka backlog draining, which aligns with the actual changeset.
Linked Issues check	✅ Passed	All core requirements from issue #510 are met: RedisJobBuffer interface with Redis/Noop implementations, BufferDrainer goroutine with backlog-aware periodic draining, Publisher normal/low buffering routing, 6 environment variables with config integration, main.go wiring for opt-in startup/shutdown, and comprehensive unit tests covering all components.
Out of Scope Changes check	✅ Passed	All changes are directly aligned with issue #510 requirements: buffer/drainer components, producer decorator, config parsing, wiring in main.go, and tests. No out-of-scope modifications detected.
Docstring Coverage	✅ Passed	Docstring coverage is 86.49% which is sufficient. The required threshold is 80.00%.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feature/#510/normal-low-redis-buffer

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[gemini-code-assist]

Code Review

This pull request implements a Redis-based buffering mechanism for normal and low priority crawl messages to prevent Kafka overloading. Key additions include a BufferingProducer for message routing, a BufferDrainer for controlled publishing based on consumer lag, and the necessary configuration and storage logic. The reviewer recommends enhancing the JobBuffer interface and its Redis implementation with batch enqueueing methods to reduce network overhead and improve performance during bulk operations.

[coderabbitai]

Actionable comments posted: 2

🧹 Nitpick comments (1)

test/internal/scheduler/buffer_drainer_test.go (1)

176-178: ⚡ Quick win

Replace fixed sleeps with eventual assertions to reduce flakiness.

These tests synchronize async work via fixed delays, which can intermittently fail in slower CI environments. Prefer require.Eventually/assert.Eventually with bounded timeout + polling interval.

Example pattern

  d.Start(ctx)
- time.Sleep(80 * time.Millisecond)
+ require.Eventually(t, func() bool {
+   return prod.count() == 5
+ }, time.Second, 10*time.Millisecond)
  cancel()
  d.Stop()

  assert.Equal(t, 5, prod.count(), "buffer 5건 모두 drain 후 underlying publish")

Also applies to: 205-207, 237-239, 263-265

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@test/internal/scheduler/buffer_drainer_test.go` around lines 176 - 178,
Replace the fragile fixed time.Sleep calls in the buffer_drainer tests with
eventual assertions (require.Eventually or assert.Eventually) that poll until
the expected condition is true within a bounded timeout; for each occurrence
around the time.Sleep + cancel() + d.Stop() pattern, assert the async work
completed (e.g., buffer length becomes 0, goroutine has exited, or a flag is
set) before calling cancel() and d.Stop(), using a short polling interval and a
generous timeout to avoid CI flakiness and apply the same change to the other
similar blocks.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@internal/scheduler/buffer_drainer.go`:
- Around line 237-246: The recovery loop uses the potentially cancelled ctx when
calling d.buffer.EnqueueJob, risking data loss if shutdown cancelled ctx; change
it to create and use a non-cancelled bounded context (e.g., context.Background()
or context.WithTimeout) for the re-enqueue calls so EnqueueJob runs even if the
original ctx is cancelled. Update the code around d.producer.PublishBatch and
the subsequent for loop that calls d.buffer.EnqueueJob to derive and pass the
new recoveryCtx (with a small timeout) instead of the original ctx, and ensure
recoveryCtx is cancelled after use.

In `@pkg/queue/buffering_producer.go`:
- Around line 72-81: NewBufferingProducer currently accepts nil for required
dependencies and stores them, which leads to nil-pointer panics later; update
NewBufferingProducer to validate inputs and fail fast: if underlying == nil or
log == nil, return an immediate panic (or explicit runtime error) with a clear
message like "NewBufferingProducer: underlying Producer is required" /
"NewBufferingProducer: log is required" so callers get a clear error at
construction time; keep the existing fallback to NoopJobBuffer for buffer but do
not allow nil underlying or nil log to be stored on the BufferingProducer struct
(referencing NewBufferingProducer, BufferingProducer.underlying and
BufferingProducer.log which are used by methods that assume non-nil).

---

Nitpick comments:
In `@test/internal/scheduler/buffer_drainer_test.go`:
- Around line 176-178: Replace the fragile fixed time.Sleep calls in the
buffer_drainer tests with eventual assertions (require.Eventually or
assert.Eventually) that poll until the expected condition is true within a
bounded timeout; for each occurrence around the time.Sleep + cancel() + d.Stop()
pattern, assert the async work completed (e.g., buffer length becomes 0,
goroutine has exited, or a flag is set) before calling cancel() and d.Stop(),
using a short polling interval and a generous timeout to avoid CI flakiness and
apply the same change to the other similar blocks.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: e73cdee4-2dc6-4112-9923-d92da6eaa73a

📥 Commits

Reviewing files that changed from the base of the PR and between 33bd925 and 21d200c.

📒 Files selected for processing (11)

• .env.example
• cmd/issuetracker/main.go
• internal/scheduler/buffer_drainer.go
• pkg/config/internal/parse/parse.go
• pkg/config/processor/job_buffer.go
• pkg/queue/buffered_message.go
• pkg/queue/buffering_producer.go
• pkg/redis/job_buffer.go
• test/internal/scheduler/buffer_drainer_test.go
• test/pkg/queue/buffering_producer_test.go
• test/pkg/redis/job_buffer_test.go

[Copilot]

Pull request overview

normal/low priority crawl 토픽의 Kafka backlog 급증 시 publish 손실을 줄이기 위해 Redis LIST 기반 버퍼와 주기적 drainer 를 추가하는 PR입니다.

Changes:

• Redis JobBuffer, BufferingProducer, buffered message 직렬화 추가
• BufferDrainer 및 main wiring/env 설정 추가
• Redis/queue/drainer 단위 테스트와 .env.example 갱신

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
pkg/redis/job_buffer.go	Redis LIST 기반 enqueue/drain/len 구현
pkg/queue/buffering_producer.go	normal/low topic 을 Redis buffer 로 라우팅하는 Producer 데코레이터 추가
pkg/queue/buffered_message.go	Redis 저장용 queue message JSON 직렬화/역직렬화 추가
internal/scheduler/buffer_drainer.go	backlog 기준 Redis buffer → Kafka drain goroutine 추가
pkg/config/processor/job_buffer.go	JobBuffer env 설정 로더 추가
pkg/config/internal/parse/parse.go	PositiveInt64 env parse helper 추가
cmd/issuetracker/main.go	Redis 공유 클라이언트 조기 초기화, buffer/drainer wiring 및 shutdown 추가
.env.example	Redis buffer 관련 환경 변수 문서화
test/pkg/redis/job_buffer_test.go	Redis JobBuffer 동작 테스트 추가
test/pkg/queue/buffering_producer_test.go	BufferingProducer 라우팅/직렬화 테스트 추가
test/internal/scheduler/buffer_drainer_test.go	BufferDrainer drain/failure 경로 테스트 추가

Comments suppressed due to low confidence (2)

internal/scheduler/buffer_drainer.go:215

• 여기서 DrainJobs 가 Redis LIST 에서 항목을 먼저 제거한 뒤 Kafka publish 를 시도하므로, 프로세스가 pop 이후 publish/재적재 전에 종료되면 해당 job 은 Redis 와 Kafka 양쪽에서 사라집니다. 같은 코드베이스의 retry queue 는 peek-publish-ack 패턴으로 crash 후 재처리를 보장하므로(pkg/redis/retry_queue.go:68-83), 이 버퍼도 processing list/ack 또는 peek 후 성공 시 제거 방식으로 바꾸지 않으면 "drop 해소" 목적과 달리 새로운 데이터 손실 창이 생깁니다.

	// 3) drain
	payloads, err := d.buffer.DrainJobs(ctx, label, n)

internal/scheduler/buffer_drainer.go:176

• drainOnce 가 backlog 조회, drain 수 계산, Redis pop, decode, Kafka publish, 실패 재적재까지 한 함수에 모두 들어 있어 변경 범위가 커지고 오류 경로 추적이 어렵습니다. 특히 손실 방지 로직은 별도 helper 로 분리해 성공/실패 경계를 명확히 하는 편이 유지보수와 테스트에 안전합니다.

func (d *BufferDrainer) drainOnce(ctx context.Context, label, topic string) error {
	// 1) 현재 backlog 조회
	checkCtx := ctx
	if d.checkTimeout > 0 {
		var cancel context.CancelFunc
		checkCtx, cancel = context.WithTimeout(ctx, d.checkTimeout)