[FEAT#5] kafka 연결 - #7
Conversation
- Makefile에 Kafka 파이프라인 실행 명령어 추가 - Kafka 관련 설정 및 Consumer/Producer 구현 - CrawlJob 및 ProcessingMessage 구조체 정의 - KafkaConsumerPool을 통한 작업 처리 로직 추가 - in-memory mock을 사용한 Kafka 파이프라인 예제 추가
…/kafka-connection
| type Config struct { | ||
| Brokers []string | ||
| GroupID string | ||
| ReadTimeout time.Duration | ||
| WriteTimeout time.Duration |
There was a problem hiding this comment.
queue.Config includes ReadTimeout, but it is not used anywhere in the producer/consumer constructors. Either wire it into the kafka-go reader/writer configuration (or enforce it via contexts) or remove it to avoid a misleading/unused configuration knob.
| raw, err := p.handler.Handle(ctx, item.job) | ||
| if err != nil { | ||
| // 재시도 횟수 초과 시 DLQ로 전송, 아니면 재큐잉 | ||
| if item.job.RetryCount >= item.job.MaxRetries { | ||
| p.sendToDLQ(ctx, item.msg, err) |
There was a problem hiding this comment.
On handler failure, the code requeues or sends to DLQ but does not commit the original consumed message. This leaves the offset uncommitted, so the same job will be reprocessed and can generate duplicate retry/DLQ messages. Consider making sendToDLQ/requeueWithRetry return an error and committing the original message only after the compensating publish succeeds.
| func (p *KafkaConsumerPool) processJob(ctx context.Context, item jobItem) error { | ||
| log := logger.FromContext(ctx) | ||
|
|
||
| log.WithFields(map[string]interface{}{ | ||
| "job_id": item.job.ID, |
There was a problem hiding this comment.
This PR adds substantial new worker-pool / retry / DLQ behavior, but there are no automated tests covering cases like handler error → requeue/DLQ, malformed message → DLQ+commit, and shutdown draining without panics. Since the repo already uses package-level tests under /test with coverage, adding similar tests for internal/crawler/worker would help prevent regressions.
| chrome-start chrome-stop chrome-status run-example-docker | ||
| chrome-start chrome-stop chrome-status run-example-docker \ | ||
| run-kafka-pipeline \ | ||
| kafka-start kafka-stop kafka-clean kafka-status kafka-logs kafka-topics |
There was a problem hiding this comment.
Makefile defines kafka-describe and kafka-scale-partitions targets, but they are not included in the .PHONY list at the top. If a file with the same name exists, make may skip running the recipe; add these targets to .PHONY for consistency with the other kafka-* commands.
| kafka-start kafka-stop kafka-clean kafka-status kafka-logs kafka-topics | |
| kafka-start kafka-stop kafka-clean kafka-status kafka-logs kafka-topics kafka-describe kafka-scale-partitions |
| github.com/mattn/go-isatty v0.0.19 // indirect | ||
| github.com/pierrec/lz4/v4 v4.1.15 // indirect | ||
| github.com/pmezard/go-difflib v1.0.0 // indirect | ||
| github.com/segmentio/kafka-go v0.4.50 // indirect |
There was a problem hiding this comment.
github.com/segmentio/kafka-go is imported by module code (pkg/queue), so it should be a direct dependency in go.mod (not marked as // indirect). Running go mod tidy after adding the new package should correct this and keep dependency metadata accurate.
Gemini code review #6, #7, #8 통합 처리. Resolver (#6, #7): - DefaultMaxCacheEntries (10,000) — cache 폭증 시 OOM 방어 - WithMaxCacheEntries(n) Option — 운영자 override - evictExpiringSoon: 가득 차면 가장 만료 임박 entry 제거 (LRU 가 아니라 expiry-order eviction — 단순하지만 본 패키지의 부하 패턴 (소수 host 반복) 에 충분, 외부 의존성 회피) - log 필드 제거 + WithLogger Option 제거 — 실제 미사용. 향후 cache hit/miss 로깅이 필요해지면 재추가. pgParsingRuleRepository (#8): - log 필드 제거 — 모든 메소드에서 미사용 - NewParsingRuleRepository 시그니처는 보존 (다른 Repository 와 일관성, 향후 logging 추가 대비) - 인자는 받되 _ = log 로 명시적으로 무시 전체 race 테스트 23 패키지 통과. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Coderabbit code review 7건 통합 처리. #1 .vscode/settings.json 추적 제외: - gh CLI 자동 승인이 commit 에 포함됨 (다른 contributor 영향) - .vscode/settings.json 제거 + .gitignore 의 ".vscode/" 활성화 #2 hasRequiredSelector 헬퍼 (parser.go): - nil 만 검사하면 zero-value selector (CSS 빈 문자열) 가 ErrParseFailure 로 잘못 분류 - nil + CSS trim 후 빈 문자열 모두 ErrEmptySelector 로 명확 분류 - ParsePage: Title + MainContent 둘 다 필수 - ParseLinks: ItemContainer + ItemLink 둘 다 필수 #4 TargetType 주석 수정: - "article" | "list" → "page" | "list" (도메인 일반화 commit 후 미반영 부분) #5, #6 자연키 ↔ lookup 키 정렬 (migration 006): - 이전 UNIQUE: (source_name, host_pattern, target_type, version) → FindActive 가 (host_pattern, target_type) 만 lookup 하면 동일 host/type/version 의 두 source row 가 활성화될 때 nondeterministic - 신규 UNIQUE: (host_pattern, target_type, version) → resolver lookup 키와 정렬, 의도치 않은 두 row 활성화 schema 단계 차단 - source_name 은 metadata 로 보존 (어느 source 가 등록했는지 추적) #7 test require.NoError: - 이전 _, _ := r.Resolve(...) 에러 무시 → call-count assertion 만 검증 - 모든 Resolve 호출에 require.NoError 추가 — 회귀 가시성 강화 #3 (wildcard host_pattern 매칭) 은 별도 후속 — 본 PR 범위 밖, 후속 issue 권장. 전체 race 테스트 23 패키지 통과. 라이브 DB migration 재적용 + UNIQUE 제약 검증 완료. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…er.CrawlTopic 통합 + 불필요 counter 제거 gemini + coderabbit + Copilot 리뷰 반영: 1. RetrySchedulerHolder 필드 rename — `S` → `Scheduler` - atomic.Pointer 호환을 위해 노출하되 명확한 이름으로 가독성 개선 (gemini #1, #3, #4) - publisher/retry.go 정의 + pool.go SetRetryScheduler / resolveRetryScheduler 의 참조 갱신 2. CrawlTopic 단일화 (gemini #2 + coderabbit) - publisher 의 `crawlTopic` (unexported) → `CrawlTopic` (exported) 로 노출 - worker 의 `topicForPriority` 제거 — pool.requeueWithRetry / manager.Publish 가 publisher.CrawlTopic 직접 호출 - scheduler/throttle.go 의 중복 `crawlTopic` 제거 — publisher.CrawlTopic 사용 - Kafka I/O 책임이 publisher 단일 출처라는 메타 #385 원칙과 일관 3. pool_retry_scheduler_test.go 의 불필요 counter 제거 (Copilot) - poolRetrySchedulerCounter atomic.Int32 + sync/atomic import 삭제 - 테스트 검증 로직과 무관한 placeholder 였음 4. fakeRetryQueue 의 `// ScheduledAt 정렬 유지` 주석 복원 (gemini #7) Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
연관 이슈
구현 내용
Kafka 추상화 (pkg/queue)
크롤 작업 모델 (internal/crawler/core)
핸들러 레지스트리 (internal/crawler/handler)
워커 풀 및 라우팅 (internal/crawler/worker)
Kafka 인프라 (deployments/docker)
TODO
논의 사항