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

cmd/issuetracker/main.go

@@ -305,7 +305,7 @@ func main() {
 			}
 			retryCfg.HeartbeatEveryNIdleTicks = retrySchedCfg.HeartbeatEveryNIdleTicks
 			redisRetry := publisher.NewRedisDelayedRetryScheduler(
-				redisClient, crawlerProducer,
+				redisClient, jobPublisher,
 				retryCfg,
 				log,
 			)
@@ -407,7 +407,7 @@ func main() {
 		log.Fatal("chromedp pool disabled (FETCHER_CHROMEDP_POOL_ENABLED=false) but goquery republish path is unconditional — enable pool or fork republish behavior in chain_handler")
 	}
 
-	manager := crawlerWorker.NewPoolManager(managerCfg, crawlerProducer, registry, contentSvc, resolver, log)
+	manager := crawlerWorker.NewPoolManager(managerCfg, jobPublisher, registry, contentSvc, resolver, log)
 
 	log.WithFields(map[string]interface{}{
 		"high_workers":   managerCfg.High.WorkerCount,

examples/kafka_pipeline/main.go

@@ -14,6 +14,7 @@ import (
 
 	"issuetracker/internal/processor/fetcher/core"
 	"issuetracker/internal/processor/fetcher/worker"
+	"issuetracker/internal/publisher"
 	"issuetracker/internal/storage"
 	"issuetracker/internal/storage/service"
 	"issuetracker/pkg/logger"
@@ -367,7 +368,8 @@ func main() {
 	handler := &testCrawlerHandler{log: log}
 	contentSvc := newMockContentService()
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, workerCount)
+	pub := publisher.New(producer, nil, log)
+	pool := worker.NewKafkaConsumerPool(consumer, pub, handler, contentSvc, workerCount)
 
 	start := time.Now()
 	pool.Start(ctx)

internal/processor/fetcher/worker/manager.go

@@ -81,7 +81,7 @@ type ManagerConfig struct {
 //  4. 종료 시 Stop(ctx) 호출
 type PoolManager struct {
 	pools    map[core.Priority]*KafkaConsumerPool
-	producer queue.Producer
+	pub      *publisher.Publisher
 	resolver PriorityResolver
 	log      *logger.Logger
 
@@ -92,9 +92,12 @@ type PoolManager struct {
 // NewPoolManager는 설정에 따라 우선순위별 KafkaConsumerPool을 생성하고 PoolManager를 반환합니다.
 //
 // NewPoolManager creates one KafkaConsumerPool per priority level (high/normal/low).
+//
+// 이슈 #390 — 구 queue.Producer 직접 주입 → publisher facade 주입으로 변경. fetcher/worker
+// 는 더 이상 queue.Producer 를 직접 보유하지 않음 (Kafka I/O 단일 출처 = publisher).
 func NewPoolManager(
 	cfg ManagerConfig,
-	producer queue.Producer,
+	pub *publisher.Publisher,
 	handler JobHandler,
 	contentSvc service.ContentService,
 	resolver PriorityResolver,
@@ -115,7 +118,7 @@ func NewPoolManager(
 
 	newPool := func(pc PoolConfig, priorityName string) *KafkaConsumerPool {
 		pool := NewKafkaConsumerPoolWithOptions(
-			pc.Consumer, producer, handler, contentSvc, pc.WorkerCount,
+			pc.Consumer, pub, handler, contentSvc, pc.WorkerCount,
 			cbRegistry, buildGate(pc.WorkerCount),
 		)
 		// heartbeat 식별자 주입 (DEBUG 레벨에서 worker pool status 출력 활성)
@@ -134,15 +137,15 @@ func NewPoolManager(
 			core.PriorityNormal: newPool(cfg.Normal, "normal"),
 			core.PriorityLow:    newPool(cfg.Low, "low"),
 		},
-		producer: producer,
+		pub:      pub,
 		resolver: resolver,
 		log:      log,
 	}
 
 	// chromedp 전용 pool wiring (Consumer + Handler 둘 다 있을 때만 활성).
 	if cfg.Chromedp.Consumer != nil && cfg.ChromedpHandler != nil {
 		chromedpPool := NewKafkaConsumerPoolWithOptions(
-			cfg.Chromedp.Consumer, producer, cfg.ChromedpHandler, contentSvc, cfg.Chromedp.WorkerCount,
+			cfg.Chromedp.Consumer, pub, cfg.ChromedpHandler, contentSvc, cfg.Chromedp.WorkerCount,
 			cbRegistry, buildGate(cfg.Chromedp.WorkerCount),
 		)
 		chromedpPool.SetPriority("chromedp")
@@ -160,35 +163,21 @@ func NewPoolManager(
 // Publish resolves the job's priority via the configured PriorityResolver,
 // updates job.Priority in-place, and publishes to the correct crawl topic
 // (crawl.high / crawl.normal / crawl.low).
+//
+// gemini PR #400 피드백 — marshal/topic/headers 구성은 publisher.PublishJob 에 위임.
+// manager 는 priority 결정 + 로깅만 책임 (priority resolver chain 통합은 Sub 6 에서).
 func (m *PoolManager) Publish(ctx context.Context, job *core.CrawlJob) error {
 	priority := m.resolver.Resolve(job)
 	job.Priority = priority
 
-	data, err := job.Marshal()
-	if err != nil {
-		return fmt.Errorf("marshal job %s: %w", job.ID, err)
-	}
-
-	topic := publisher.CrawlTopic(priority)
-
-	msg := queue.Message{
-		Topic: topic,
-		Key:   []byte(job.ID),
-		Value: data,
-		Headers: map[string]string{
-			"crawler":  job.CrawlerName,
-			"priority": fmt.Sprintf("%d", int(priority)),
-		},
-	}
-
 	m.log.WithFields(map[string]interface{}{
 		"job_id":   job.ID,
 		"crawler":  job.CrawlerName,
 		"priority": priority,
-		"topic":    topic,
+		"topic":    publisher.CrawlTopic(priority),
 	}).Info("publishing crawl job")
 
-	return m.producer.Publish(ctx, msg)
+	return m.pub.PublishJob(ctx, job)
 }
 
 // Start는 high/normal/low 모든 Pool의 goroutine을 시작합니다.

internal/processor/fetcher/worker/pool.go

@@ -62,8 +62,8 @@ var kafkaRequeuePolicy = core.RetryPolicy{
 //  3. worker goroutine들이 jobs 드레인 후 종료
 //  4. consumer.Close()
 type KafkaConsumerPool struct {
-	consumer    queue.Consumer
-	producer    queue.Producer
+	consumer    publisher.Consumer
+	pub         *publisher.Publisher
 	handler     JobHandler
 	contentSvc  service.ContentService
 	workerCount int
@@ -110,16 +110,16 @@ type jobItem struct {
 // issuetracker.normalized 토픽에 발행합니다.
 // workerCount는 동시에 실행되는 처리 goroutine 수를 결정합니다.
 func NewKafkaConsumerPool(
-	consumer queue.Consumer,
-	producer queue.Producer,
+	consumer publisher.Consumer,
+	pub *publisher.Publisher,
 	handler JobHandler,
 	contentSvc service.ContentService,
 	workerCount int,
 ) *KafkaConsumerPool {
 	// CircuitBreakerRegistry log 주입은 NewPoolManager 가 PoolManager 단위에서 처리.
 	// 본 헬퍼는 단순 호출용 (테스트/예제) 이라 logger 주입 안 함 — nil 이면 state 전이 로그 skip.
 	return NewKafkaConsumerPoolWithOptions(
-		consumer, producer, handler, contentSvc, workerCount,
+		consumer, pub, handler, contentSvc, workerCount,
 		NewCircuitBreakerRegistry(DefaultCircuitBreakerConfig, nil),
 		locks.NewNoopStageGate(),
 	)
@@ -128,15 +128,15 @@ func NewKafkaConsumerPool(
 // NewKafkaConsumerPoolWithCB는 외부에서 주입한 CircuitBreakerRegistry를 사용하는
 // KafkaConsumerPool을 생성합니다. 테스트에서 circuit breaker 동작을 제어할 때 사용합니다.
 func NewKafkaConsumerPoolWithCB(
-	consumer queue.Consumer,
-	producer queue.Producer,
+	consumer publisher.Consumer,
+	pub *publisher.Publisher,
 	handler JobHandler,
 	contentSvc service.ContentService,
 	workerCount int,
 	cbRegistry *CircuitBreakerRegistry,
 ) *KafkaConsumerPool {
 	return NewKafkaConsumerPoolWithOptions(
-		consumer, producer, handler, contentSvc, workerCount,
+		consumer, pub, handler, contentSvc, workerCount,
 		cbRegistry,
 		locks.NewNoopStageGate(),
 	)
@@ -148,8 +148,8 @@ func NewKafkaConsumerPoolWithCB(
 // gate 는 nil 허용 — nil 이면 NoopStageGate 로 fallback (단일 인스턴스 환경에서 dedup + cap 비활성).
 // 이슈 #356 — fetcher / parser / validator 가 동일 StageGate 패턴 사용.
 func NewKafkaConsumerPoolWithOptions(
-	consumer queue.Consumer,
-	producer queue.Producer,
+	consumer publisher.Consumer,
+	pub *publisher.Publisher,
 	handler JobHandler,
 	contentSvc service.ContentService,
 	workerCount int,
@@ -163,7 +163,7 @@ func NewKafkaConsumerPoolWithOptions(
 	}
 	return &KafkaConsumerPool{
 		consumer:    consumer,
-		producer:    producer,
+		pub:         pub,
 		handler:     handler,
 		contentSvc:  contentSvc,
 		workerCount: workerCount,
@@ -674,7 +674,7 @@ func (p *KafkaConsumerPool) publishNormalized(ctx context.Context, content *core
 		},
 	}
 
-	return p.producer.Publish(ctx, msg)
+	return p.pub.Forward(ctx, msg)
 }
 
 // commitMessage 는 Kafka offset 을 commit 합니다.
@@ -731,13 +731,13 @@ func (p *KafkaConsumerPool) sendToDLQ(ctx context.Context, msg *queue.Message, r
 		Headers: headers,
 	}
 
-	err := p.producer.Publish(ctx, dlqMsg)
+	err := p.pub.Forward(ctx, dlqMsg)
 	if err != nil && errors.Is(err, context.Canceled) {
 		drainCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), drainTimeout)
 		defer cancel()
 		// errors.Join 으로 최초 cancel 과 retryErr 를 모두 보존 — 호출자의
 		// errors.Is(err, context.Canceled) 분기가 안정적으로 매칭되도록 보장.
-		if retryErr := p.producer.Publish(drainCtx, dlqMsg); retryErr != nil {
+		if retryErr := p.pub.Forward(drainCtx, dlqMsg); retryErr != nil {
 			err = errors.Join(err, retryErr)
 		} else {
 			err = nil
@@ -801,7 +801,7 @@ func (p *KafkaConsumerPool) resolveRetryScheduler() publisher.RetryScheduler {
 	if h := p.retryScheduler.Load(); h != nil {
 		return h.Scheduler
 	}
-	return publisher.NewKafkaImmediateRetryScheduler(p.producer)
+	return publisher.NewKafkaImmediateRetryScheduler(p.pub)
 }
 
 // logShutdownAware 는 graceful shutdown 으로 발생한 컨텍스트성 에러를 DEBUG 로,

internal/publisher/publisher.go

@@ -17,8 +17,10 @@
 package publisher
 
 import (
+	"context"
 	"crypto/rand"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"sync/atomic"
 
@@ -29,6 +31,22 @@ import (
 	"issuetracker/pkg/urlguard"
 )
 
+// Consumer 는 Kafka 메시지 소비 인터페이스의 publisher-측 별칭입니다 (이슈 #390).
+//
+// fetcher/worker 등 다운스트림 모듈이 queue 패키지에 직접 의존하지 않도록 — Kafka I/O
+// 단일 책임 원칙 (메타 #385) 의 일환. queue.Consumer 와 100% 동일 시그니처 (type alias).
+//
+// 호출자는 publisher 가 제공하는 인스턴스나 외부 wiring 에서 queue.NewConsumer 로 생성한
+// *KafkaConsumer 를 그대로 사용. 본 패키지가 별도 factory 메소드를 제공하기 전까지는
+// queue.NewConsumer wiring 을 caller 측에서 직접 수행.
+type Consumer = queue.Consumer
+
+// Message 는 Kafka 메시지 구조체의 publisher-측 별칭입니다 (이슈 #390 피드백 — gemini).
+//
+// Consumer 별칭과 마찬가지로 다운스트림 모듈이 queue 패키지에 직접 의존하지 않고 publisher
+// API 만으로 Forward 호출 시 메시지 구성을 완성할 수 있도록 별칭화. queue.Message 와 동일.
+type Message = queue.Message
+
 // DefaultMaxRetries 는 PublishX 메소드들이 생성하는 CrawlJob 의 기본 재시도 횟수입니다
 // (CodeRabbit PR #394 피드백 — magic number 상수화).
 const DefaultMaxRetries = 3
@@ -76,6 +94,46 @@ func New(producer queue.Producer, resolver PriorityResolver, log *logger.Logger)
 // 공통 Kafka helpers (모든 PublishX 메소드가 공유)
 // ─────────────────────────────────────────────────────────────────────────────
 
+// Forward 는 미리 구성된 Message 를 내부 producer 로 그대로 발행합니다 (이슈 #390).
+//
+// 사용처 — fetcher/worker 가 Kafka I/O 책임을 publisher 로 위임하면서도 worker-특수
+// 메시지 (normalized contentRef / DLQ 등) 의 구성 / 라우팅은 worker 측에 잔존하는 경우의
+// thin pass-through. publisher 가 자체 Marshal/Topic 결정을 책임지는 PublishX 와 달리 본
+// 메소드는 호출자가 완성된 Message 를 만들어 전달합니다.
+//
+// nil guard — coderabbit PR #400 피드백. 본 메소드가 retry / worker hot path 에서 호출되므로
+// p 또는 p.producer 가 nil 일 때 panic 대신 에러 반환 (silent crash 보다 명시적 fail).
+//
+// Kafka I/O 자체는 publisher 가 단일 출처 — 호출자는 queue.Producer 를 직접 보유하지 않음.
+func (p *Publisher) Forward(ctx context.Context, msg Message) error {
+	if p == nil {
+		return errors.New("publisher: Forward called on nil *Publisher")
+	}
+	if p.producer == nil {
+		return errors.New("publisher: producer not wired")
+	}
+	return p.producer.Publish(ctx, msg)
+}
+
+// PublishJob 은 CrawlJob 을 marshal 하여 우선순위 토픽으로 발행합니다 (이슈 #390 피드백 — gemini).
+//
+// 구 manager.Publish 가 직접 수행하던 marshal / 토픽 결정 / 헤더 구성 로직을 publisher 측
+// buildMessage 헬퍼로 일원화하여 코드 중복 제거. 호출자는 priority 가 미리 결정된 job 을
+// 전달합니다 (priority resolver chain 통합은 Sub 6 에서).
+func (p *Publisher) PublishJob(ctx context.Context, job *core.CrawlJob) error {
+	if p == nil {
+		return errors.New("publisher: PublishJob called on nil *Publisher")
+	}
+	if job == nil {
+		return errors.New("publisher: PublishJob called with nil job")
+	}
+	msg, err := p.buildMessage(job)
+	if err != nil {
+		return err
+	}
+	return p.Forward(ctx, msg)
+}
+
 // buildMessage 는 CrawlJob 을 Kafka Message 로 변환합니다.
 func (p *Publisher) buildMessage(job *core.CrawlJob) (queue.Message, error) {
 	data, err := job.Marshal()