[REFAC#390] fetcher/worker Kafka I/O → publisher facade 위임

[juhy0987]

연관 이슈

Closes #390
부모 메타: #385 — Publisher 통합 Sub 5

구현 내용

fetcher/worker 가 queue.Producer / queue.Consumer 를 직접 보유하지 않고 *publisher.Publisher 를 통해 Kafka I/O 를 수행하도록 책임 분리. 메타 #385 의 Kafka I/O 단일 책임 원칙 일관성 확보.

publisher 패키지 신규 API

추가	설명
type Consumer = queue.Consumer (alias)	fetcher/worker 가 queue 패키지에 직접 의존하지 않도록 publisher 측 별칭
(*Publisher).Forward(ctx, msg queue.Message) error	호출자가 완성된 Message 를 publisher 내부 producer 로 그대로 발행하는 thin pass-through. PublishX (PublishSeed/PublishUpgrade) 와 달리 호출자가 토픽/마샬링을 책임지는 경우의 escape hatch

publisher.retry 시그니처 변경

- NewKafkaImmediateRetryScheduler(producer queue.Producer) *KafkaImmediateRetryScheduler
+ NewKafkaImmediateRetryScheduler(pub *Publisher) *KafkaImmediateRetryScheduler

- NewRedisDelayedRetryScheduler(client retryQueueClient, producer queue.Producer, cfg, log) *RedisDelayedRetryScheduler
+ NewRedisDelayedRetryScheduler(client retryQueueClient, pub *Publisher, cfg, log) *RedisDelayedRetryScheduler

두 구현체 내부에서 s.producer.Publish → s.pub.Forward 로 변경.

worker.pool / worker.manager 변경

• 필드 producer queue.Producer → pub *publisher.Publisher
• 필드 consumer queue.Consumer → consumer publisher.Consumer (별칭이라 동일)
• 모든 생성자 시그니처 — NewKafkaConsumerPool / WithCB / WithOptions / NewPoolManager
• publishNormalized / sendToDLQ / resolveRetryScheduler / manager.Publish 가 publisher facade 사용

cmd/issuetracker/main.go wiring

- manager := crawlerWorker.NewPoolManager(managerCfg, crawlerProducer, registry, contentSvc, resolver, log)
+ manager := crawlerWorker.NewPoolManager(managerCfg, jobPublisher, registry, contentSvc, resolver, log)

- redisRetry := publisher.NewRedisDelayedRetryScheduler(redisClient, crawlerProducer, ...)
+ redisRetry := publisher.NewRedisDelayedRetryScheduler(redisClient, jobPublisher, ...)

crawlerProducer 는 sources.RegisterAll 등 비-worker 경로에서 계속 사용 (해당 경로는 다음 sub 정리 범위).

테스트

• newTestPublisher(producer queue.Producer) *publisher.Publisher 헬퍼 도입 (test/.../worker/pool_test.go) — 실제 publisher.New 로 mockProducer 를 wrap. mockProducer 의 Publish expectation 은 pub.Forward → producer.Publish 위임 경로로 그대로 트리거.
• 동일 패턴 retryTestPub 헬퍼 추가 (test/internal/publisher/retry_test.go).
• 모든 worker.NewKafkaConsumerPool* 호출 시 producer → newTestPublisher(producer) 치환.
• TestKafkaConsumerPool_SetRetryScheduler_BypassesInlinePublish 도 동일 pub 인스턴스 공유.

CI / 머지 게이트 점검

• gofmt -l — clean
• go build ./internal/... ./cmd/... ./pkg/... ./test/... — pass
• go test -race -count=1 ./test/internal/publisher/... ./test/internal/processor/fetcher/worker/... ./test/internal/scheduler/... — 전 패키지 통과
• PR 타이틀 [REFAC#390]
• commit [REFAC]: prefix + 한국어

변경 영향 범위 + 위험도

• 영향: publisher / fetcher/worker (pool + manager) / cmd/issuetracker / 4 개 테스트 파일
• 위험도 High → Medium 동작 동등성 확보됨:
  • publisher.Forward 는 thin pass-through (단일 함수 호출)
  • 모든 기존 테스트가 새 wiring 으로 통과
  • mockProducer expectation 흐름 보존 — pool 의 publish 검증 무변경
• 다음 Sub 6 ([REFACTOR] PriorityResolver chain → publisher 측 이동 + 모든 PublishX 통과 (#385 Sub 6) #391) — PriorityResolver chain publisher 측 이동
• 다음 Sub 7/8 — parser/validate worker 동일 패턴 적용

롤백 계획

PR revert 시 모든 시그니처가 동시에 원복 — wiring 회귀 없음.

🤖 Generated with Claude Code

Summary by CodeRabbit

• Refactor
  • Unified internal Kafka publishing infrastructure to use a consistent publisher facade across retry scheduling and message processing components, improving architectural consistency without affecting end-user functionality.

[coderabbitai]

Warning

Rate limit exceeded

@juhy0987 has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 50 minutes and 27 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: 398d18c1-2de9-4613-a885-24d9009d97d7

📥 Commits

Reviewing files that changed from the base of the PR and between cc4786f and fadb6ec.

📒 Files selected for processing (4)

• examples/kafka_pipeline/main.go
• internal/processor/fetcher/worker/manager.go
• internal/publisher/publisher.go
• internal/publisher/retry.go

📝 Walkthrough

Walkthrough

Fetcher worker pool, pool manager, and retry schedulers are refactored to depend on publisher.Publisher facade instead of raw Kafka producer/consumer, unifying all Kafka I/O through the publisher layer with new Forward method and Consumer type alias.

Changes

Publisher Facade Unification for Fetcher and Retry

Layer / File(s)	Summary
Publisher facade enrichment: Forward and Consumer internal/publisher/publisher.go	Consumer type alias and Publisher.Forward(ctx, msg) method added as new public facades for downstream Kafka I/O delegation.
Retry schedulers migrate to Publisher facade internal/publisher/retry.go	KafkaImmediateRetryScheduler and RedisDelayedRetryScheduler refactored to accept *Publisher instead of queue.Producer, routing all retry publishes through Publisher.Forward.
Fetcher pool refactored to use Publisher consumer and publisher internal/processor/fetcher/worker/pool.go	KafkaConsumerPool struct and constructors updated to depend on publisher.Consumer and *Publisher instead of raw queue interfaces, with normalized/DLQ/retry publishing routed through Publisher.Forward.
Fetcher manager migrated to Publisher facade internal/processor/fetcher/worker/manager.go	PoolManager constructor and struct updated to accept and store *Publisher, with all pool creation and public Publish method delegating through the unified publisher.
Main application wiring: retry and fetcher through Publisher cmd/issuetracker/main.go	Both RedisDelayedRetryScheduler and NewPoolManager are now constructed with jobPublisher instead of raw crawlerProducer.
Fetcher worker tests rewired with Publisher wrapper test/internal/processor/fetcher/worker/pool*.go, test/internal/processor/fetcher/worker/processing_lock_pool_test.go	All pool and manager tests updated to use newTestPublisher(producer) helper that wraps mock producer into Publisher instance, preserving mock assertion paths.
Retry scheduler tests rewired with Publisher wrapper test/internal/publisher/retry_test.go	All retry scheduler tests updated to use retryTestPub(producer) helper that wraps mock producer into Publisher instance for immediate and delayed retry test scenarios.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~25 minutes

Possibly related issues

• [REFACTOR] fetcher/worker Kafka Consumer + Publish 책임 분리 (#385 Sub 5) #390 — This PR directly implements the publisher-unification and Kafka I/O separation work described in the parent refactoring issue, updating fetcher/worker to depend only on the publisher facade.

• [REFACTOR] fetcher/worker/retry_scheduler → publisher/retry.go 이동 (#385 Sub 4) #389 — Both changes migrate retry scheduler and fetcher worker publishing from queue.Producer to publisher.Publisher facade with matching constructor/wiring updates.

• [REFACTOR] validate worker + cmd/processor Kafka I/O → publisher facade (#385 Sub 8) #393 — Related refactoring that replaces direct queue usage with publisher facade and updates cmd/issuetracker/main.go wiring for a different worker pipeline.

Possibly related PRs

• EinSofINTEREST/IssueTracker#399 — Overlaps on retry scheduler migration: both move schedulers to depend on Publisher.Forward for Kafka publishes.

• EinSofINTEREST/IssueTracker#394 — Both refactor the internal/publisher facade layer and update how the publisher publishes jobs through the Kafka interface.

• EinSofINTEREST/IssueTracker#286 — Both modify internal/publisher and cmd/issuetracker/main.go wiring around publisher integration.

Suggested labels

refactor

Poem

A rabbit hops through publisher layers,
Wrapping producers, sharing prayers,
Forward() leaps, Consumer thrives,
Fetcher and Retry, unified lives,
Tests pass through—no more despair! 🐰

🚥 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 title '[REFAC#390] fetcher/worker Kafka I/O → publisher facade 위임' directly refers to the main refactoring objective: delegating fetcher/worker Kafka I/O responsibilities to the publisher facade, which is the primary focus of this PR.
Linked Issues check	✅ Passed	The PR successfully implements all core coding requirements from issue #390: introduces publisher.Consumer interface, refactors fetcher/worker to depend on publisher facade instead of queue.Producer/Consumer directly, updates retry schedulers and pool managers to use Publisher, and ensures test coverage maintains existing behavior.
Out of Scope Changes check	✅ Passed	All changes are directly aligned with issue #390 objectives: publisher package extensions (Consumer alias, Forward method), retry scheduler signature changes, worker pool/manager refactoring, main.go wiring updates, and test helper additions—no out-of-scope modifications detected.
Docstring Coverage	✅ Passed	Docstring coverage is 98.00% 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 refactor/#390/publisher-consumer

---

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 refactors the Kafka I/O architecture by introducing a publisher.Publisher facade to decouple downstream modules, such as the fetcher and worker, from the queue package. It replaces direct queue.Producer dependencies with the new Publisher and introduces a Consumer type alias to centralize I/O responsibilities. Review feedback suggests completing this abstraction by also aliasing queue.Message, reducing code duplication by reusing internal message-building logic for retries, and further strengthening encapsulation by moving message construction logic from the worker manager into the Publisher facade.

[coderabbitai]

Actionable comments posted: 1

🤖 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/publisher/publisher.go`:
- Around line 89-99: The Forward method can panic when called on a nil
*Publisher or when p.producer is nil; add explicit nil guards at the start of
Publisher.Forward to check if p == nil and if p.producer == nil and return a
descriptive error (e.g., using fmt.Errorf or errors.New) instead of calling
p.producer.Publish, preserving the existing signature and behavior for non-nil
cases and forwarding to p.producer.Publish only after the checks.

🪄 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: 65357a5e-265f-4523-b9b2-80410f34eb3c

📥 Commits

Reviewing files that changed from the base of the PR and between 6ea621e and cc4786f.

📒 Files selected for processing (10)

• cmd/issuetracker/main.go
• internal/processor/fetcher/worker/manager.go
• internal/processor/fetcher/worker/pool.go
• internal/publisher/publisher.go
• internal/publisher/retry.go
• test/internal/processor/fetcher/worker/pool_gate_test.go
• test/internal/processor/fetcher/worker/pool_retry_scheduler_test.go
• test/internal/processor/fetcher/worker/pool_test.go
• test/internal/processor/fetcher/worker/processing_lock_pool_test.go
• test/internal/publisher/retry_test.go

[Copilot]

Pull request overview

fetcher/worker가 queue.Producer/queue.Consumer를 직접 보유하던 구조를 정리하고, *publisher.Publisher facade를 통해 Kafka publish 경로를 위임하도록 리팩토링한 PR입니다(#385 메타의 Kafka I/O 단일 책임 원칙을 fetcher/worker에도 적용).

Changes:

• publisher.Consumer(= queue.Consumer alias)와 (*Publisher).Forward(ctx, msg) escape hatch API를 추가해 다운스트림에서 producer 직접 보유를 제거
• publisher.retry(KafkaImmediate/RedisDelayed)의 publish 경로를 producer.Publish → pub.Forward로 변경하고 시그니처를 *Publisher 주입 방식으로 통일
• fetcher worker pool/manager 및 main wiring, 관련 테스트들을 새 facade 주입 형태로 일괄 갱신

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
internal/publisher/publisher.go	Consumer alias 및 Forward pass-through API 추가
internal/publisher/retry.go	RetryScheduler들이 queue.Producer 대신 *Publisher를 사용하도록 변경
internal/processor/fetcher/worker/pool.go	pool이 producer 직접 보유 대신 pub.Forward로 publish 수행
internal/processor/fetcher/worker/manager.go	manager가 producer 직접 보유 대신 pub.Forward로 publish 수행
cmd/issuetracker/main.go	retry scheduler/manager wiring을 jobPublisher 주입으로 변경
test/internal/publisher/retry_test.go	retry scheduler 테스트가 *Publisher 주입 형태로 변경되도록 헬퍼 추가/치환
test/internal/processor/fetcher/worker/pool_test.go	worker pool 테스트에서 newTestPublisher 헬퍼로 publisher facade 주입
test/internal/processor/fetcher/worker/processing_lock_pool_test.go	pool 생성 시 producer → publisher facade 주입으로 변경
test/internal/processor/fetcher/worker/pool_retry_scheduler_test.go	커스텀 retry scheduler도 동일 pub 인스턴스 공유하도록 변경
test/internal/processor/fetcher/worker/pool_gate_test.go	pool 생성 시 producer → publisher facade 주입으로 변경