[REFAC#392] parser worker + fetcher chain_handler Kafka I/O → publisher facade by juhy0987 · Pull Request #401 · EinSofINTEREST/IssueTracker

juhy0987 · 2026-05-13T02:34:51Z

연관 이슈

Closes #392
부모 메타: #385 — Publisher 통합 Sub 7

추가 범위 (사용자 요청): fetcher 의 chain_handler.go 잔여 Kafka I/O 도 동일 PR 에 묶음. 메타 #385 의 어느 sub 에도 명시되지 않았지만 "Kafka I/O 단일 책임 원칙" 일관성 차원에서 통합.

구현 내용

parser worker (Sub 7 본체)

Before	After
`consumer *queue.KafkaConsumer`	`consumer publisher.Consumer` (Sub 5 의 별칭)
`producer queue.Producer` + `publisher general.JobPublisher` (필드 2개)	`pub *publisher.Publisher` (필드 1개 — concrete 가 양쪽 인터페이스 모두 제공)
`w.producer.Publish(...)` × 3 (TopicNormalized + 2× TopicFetched LLM/pending requeue)	`w.pub.Forward(...)`
`w.publisher.PublishChained(...)` × 2 (article / category chained jobs)	`w.pub.PublishChained(...)`
`NewParserWorker(consumer, producer, ..., publisher, ...)`	`NewParserWorker(consumer, pub, ...)` (인자 2개 통합)

fetcher chain_handler (추가 범위)

Before	After
`Producer queue.Producer` 필드	`Pub *publisher.Publisher`
`h.Producer.Publish` × 2 (`republishToChromedpQueue` + `publishFetchedRef`)	`h.Pub.Forward(...)`
`NewChainHandler(..., producer queue.Producer, ...)`	`NewChainHandler(..., pub *publisher.Publisher, ...)`
`sources/registry.go` pass-through `producer queue.Producer`	`pub *publisher.Publisher` (queue 직접 의존 제거)

cmd/issuetracker/main.go wiring

-	pw := parserWorker.NewParserWorker(parserConsumer, crawlerProducer, rawSvc, contentSvc, jobPublisher, ...)
+	pw := parserWorker.NewParserWorker(parserConsumer, jobPublisher, rawSvc, contentSvc, ...)  // producer + jobPublisher 통합

-	sources.RegisterAll(ctx, registry, ..., rawSvc, crawlerProducer, fetcherResolver, ...)
+	sources.RegisterAll(ctx, registry, ..., rawSvc, jobPublisher, fetcherResolver, ...)

테스트

test/.../parser/worker/helpers_test.go newMinimalWorker(prod, log) 내부에서 publisher.New(prod, nil, log) 래핑 (Sub 5 의 newTestPublisher 패턴과 동일).
test/.../parser/worker/stage_gate_test.go 생성자 인자 갯수 / 주석 정정.
test/.../fetcher/domain/general/chain_handler_*_test.go 2 개 파일: captureProducer / stubProducer 를 publisher.New(...) 로 래핑.

CI / 머지 게이트 점검

gofmt -l — clean
go build ./internal/... ./cmd/issuetracker/ ./test/... — pass
go test -race -count=1 -timeout=180s (parser worker / fetcher domain / publisher / worker) — 전 패키지 통과
PR 타이틀 [REFAC#392]
commit [REFAC]: prefix + 한국어

변경 영향 범위 + 위험도

영향: parser worker + fetcher chain_handler + sources registry + main wiring + 4 test 파일
위험도 Medium → 동작 동등성 확보됨:
- publisher.Forward 는 thin pass-through (Sub 5 동일 검증)
- PublishChained 는 기존 publisher.PublishChained 그대로 사용
- 라이브 publish 시점/내용 무변경
다음 Sub 8 ([REFACTOR] validate worker + cmd/processor Kafka I/O → publisher facade (#385 Sub 8) #393) — validate worker + cmd/processor Kafka I/O 정리
Sub 6 ([REFACTOR] PriorityResolver chain → publisher 측 이동 + 모든 PublishX 통과 (#385 Sub 6) #391) PriorityResolver chain 이동은 별도 진행

롤백 계획

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

🤖 Generated with Claude Code

…facade (이슈 #392) 메타 #385 Sub 7 — parser worker 의 Kafka I/O 책임을 publisher facade 로 위임. 동시에 사용자 요청으로 fetcher 의 chain_handler.go 에 남아있던 Kafka I/O 잔여 책임도 함께 정리. ## parser worker - 필드 `consumer *queue.KafkaConsumer` → `consumer publisher.Consumer` (별칭 인터페이스) - 필드 `producer queue.Producer` 제거 → `pub *publisher.Publisher` 로 통합 - 필드 `publisher general.JobPublisher` 제거 → 동일 `pub` 가 PublishChained 도 제공 (concrete *publisher.Publisher 가 general.JobPublisher 만족) - 3 종 직접 publish 모두 `w.pub.Forward(...)` 위임: - TopicNormalized publish (parsed page → validator) - TopicFetched publish (LLM 재학습용 raw content requeue) - TopicFetched publish (pending URL 재발행) - 2 종 chained job publish → `w.pub.PublishChained(...)` 그대로 ## fetcher chain_handler (사용자 요청 추가) - 필드 `Producer queue.Producer` → `Pub *publisher.Publisher` - 2 종 직접 publish → `h.Pub.Forward(...)`: - `republishToChromedpQueue` (TopicCrawlChromedp 재발행) - `publishFetchedRef` (TopicFetched 로 RawContentRef 발행) - `NewChainHandler` 시그니처: producer queue.Producer → pub *publisher.Publisher ## sources/registry.go - `RegisterAll` + `buildHandler` 의 producer 파라미터 → pub *publisher.Publisher (pass-through) - queue 패키지 직접 import 제거 ## cmd/issuetracker/main.go wiring - `parser.NewParserWorker(..., crawlerProducer, ..., jobPublisher, ...)` → `parser.NewParserWorker(..., jobPublisher, ...)` (구 producer + JobPublisher 두 인자 통합) - `sources.RegisterAll(..., crawlerProducer, ...)` → `sources.RegisterAll(..., jobPublisher, ...)` ## 테스트 - parser/worker/helpers_test.go: `newMinimalWorker(prod, log)` 내부에서 `publisher.New(prod, nil, log)` 래핑 (Sub 5 의 newTestPublisher 패턴과 동일) - parser/worker/stage_gate_test.go: 생성자 인자 갯수 / 주석 정정 - chain_handler 테스트 2개: captureProducer / stubProducer 를 `publisher.New(...)` 로 래핑 ## 검증 - `go build ./internal/... ./cmd/issuetracker/ ./test/...` — pass - `go test -race -count=1 -timeout=180s` (parser/fetcher/publisher/worker 전부) — pass Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

coderabbitai · 2026-05-13T02:34:58Z

Warning

Rate limit exceeded

@juhy0987 has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 21 minutes and 11 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: be32c854-74e2-4a2d-9915-aa00b38d56d3

📥 Commits

Reviewing files that changed from the base of the PR and between ce5116f and c0a2ff3.

📒 Files selected for processing (8)

cmd/issuetracker/main.go
internal/processor/fetcher/domain/general/chain_handler.go
internal/processor/fetcher/domain/general/sources/registry.go
internal/processor/parser/worker/parser_worker.go
test/internal/processor/fetcher/domain/general/chain_handler_chromedp_routing_test.go
test/internal/processor/fetcher/domain/general/chain_handler_reparse_propagation_test.go
test/internal/processor/parser/worker/helpers_test.go
test/internal/processor/parser/worker/stage_gate_test.go

✨ Finishing Touches

🧪 Generate unit tests (beta)

Create PR with unit tests
Commit unit tests in branch refactor/#392/publisher-parser-fetcher-chain

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

_{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 infrastructure by consolidating the queue.Producer and JobPublisher into a single publisher.Publisher facade. This change is applied across the fetcher and parser components, specifically updating ChainHandler, ParserWorker, and the source registry to use the unified publisher for message forwarding and chained job publishing. The refactoring simplifies dependency management and aligns with the single responsibility principle for Kafka operations. I have no feedback to provide as no review comments were included in the request.

Copilot

Pull request overview

이 PR은 parser worker 및 fetcher의 general chain_handler에서 Kafka I/O(consume/publish)를 queue 직접 호출 대신 internal/publisher facade로 위임하도록 리팩토링하여, Kafka 발행/소비 책임을 publisher로 단일화(#385 메타 방향)합니다.

Changes:

ParserWorker의 consumer/producer/publisher 의존성을 publisher.Consumer + *publisher.Publisher로 통합하고, publish 경로를 pub.Forward(...)/pub.PublishChained(...)로 전환
general.ChainHandler의 republish/fetched_ref 발행을 queue.Producer 직접 호출에서 *publisher.Publisher 위임으로 전환
sources.RegisterAll 및 cmd/issuetracker/main.go wiring, 관련 테스트 헬퍼/케이스들을 새 시그니처에 맞게 갱신

Reviewed changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
internal/processor/parser/worker/parser_worker.go	consumer/발행 로직을 publisher facade로 위임하도록 의존성/호출부 변경
internal/processor/fetcher/domain/general/chain_handler.go	chromedp republish 및 fetched ref 발행을 `Pub.Forward`로 위임
internal/processor/fetcher/domain/general/sources/registry.go	fetcher source 등록 경로에서 `queue.Producer` 제거, `*publisher.Publisher` 전달로 전환
cmd/issuetracker/main.go	`RegisterAll`/`NewParserWorker` wiring을 publisher facade 기반으로 변경
test/internal/processor/parser/worker/stage_gate_test.go	`NewParserWorker` 시그니처 변경에 따른 인자/주석 정리
test/internal/processor/parser/worker/helpers_test.go	테스트용 producer를 `publisher.New(...)`로 래핑하여 `pub` 인자 제공
test/internal/processor/fetcher/domain/general/chain_handler_reparse_propagation_test.go	captureProducer를 publisher로 래핑하여 `Forward` 경로 검증 유지
test/internal/processor/fetcher/domain/general/chain_handler_chromedp_routing_test.go	stubProducer를 publisher로 래핑하여 불변 조건(불필요 publish 발생 시 panic) 유지

Copilot 피드백 2건 반영: 1. ParserWorker doc 의 "queue 패키지에 직접 의존하지 않음" 표현이 부정확. 파일은 토픽 상수 (queue.TopicFetched / TopicNormalized) 와 데이터 타입 (queue.Message) 사용을 위해 queue 를 그대로 import. 실제 의도는 "Kafka 구현체 (*queue.KafkaConsumer, queue.Producer) 에 직접 의존하지 않음" — publisher facade 가 그 책임 단일 담당. 2. NewParserWorker doc 의 "pub nil 이면 Forward 모두 건너뜀" 표현 정정. 실제 동작: publisher.Forward 가 nil 시 error 반환 → 호출 사이트가 ProcessMessage 실패로 분류 → commit skip → Kafka 메시지 무한 재배달 위험. 운영 환경에서 pub 는 필수 wiring, nil 은 publish 경로 비도달 단위 테스트 (예: stage gate 격리) 에만 허용. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Copilot AI review requested due to automatic review settings May 13, 2026 02:34

juhy0987 added the refactor Code refactoring label May 13, 2026

Copilot started reviewing on behalf of juhy0987 May 13, 2026 02:35 View session

gemini-code-assist Bot reviewed May 13, 2026

View reviewed changes

Copilot AI reviewed May 13, 2026

View reviewed changes

Comment thread internal/processor/parser/worker/parser_worker.go

Comment thread internal/processor/parser/worker/parser_worker.go Outdated

juhy0987 self-assigned this May 13, 2026

juhy0987 merged commit 29da4f3 into main May 13, 2026
8 checks passed

This was referenced May 13, 2026

[REFAC#391] PriorityResolver chain → publisher 측 이동 + 모든 PublishX 통과 #409

Merged

[FEATURE] Publisher 통합 모듈화 + Kafka I/O 단일 책임 분리 (메타) #385

Closed

Navigation Menu

[REFAC#392] parser worker + fetcher chain_handler Kafka I/O → publisher facade - #401

[REFAC#392] parser worker + fetcher chain_handler Kafka I/O → publisher facade#401
juhy0987 merged 2 commits into
mainfrom
refactor/#392/publisher-parser-fetcher-chain

juhy0987 commented May 13, 2026

Uh oh!

coderabbitai Bot commented May 13, 2026 •

edited

Loading

Rate limit exceeded

Uh oh!

gemini-code-assist Bot left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

juhy0987 commented May 13, 2026

연관 이슈

구현 내용

parser worker (Sub 7 본체)

fetcher chain_handler (추가 범위)

cmd/issuetracker/main.go wiring

테스트

CI / 머지 게이트 점검

변경 영향 범위 + 위험도

롤백 계획

Uh oh!

coderabbitai Bot commented May 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Rate limit exceeded

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

coderabbitai Bot commented May 13, 2026 •

edited

Loading