diff --git a/cmd/issuetracker/main.go b/cmd/issuetracker/main.go
index 85424c19..48e6caae 100644
--- a/cmd/issuetracker/main.go
+++ b/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,
diff --git a/examples/kafka_pipeline/main.go b/examples/kafka_pipeline/main.go
index da3c0c02..d730d179 100644
--- a/examples/kafka_pipeline/main.go
+++ b/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)
diff --git a/internal/processor/fetcher/worker/manager.go b/internal/processor/fetcher/worker/manager.go
index bd4bac77..1685c6a8 100644
--- a/internal/processor/fetcher/worker/manager.go
+++ b/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,7 +137,7 @@ func NewPoolManager(
 			core.PriorityNormal: newPool(cfg.Normal, "normal"),
 			core.PriorityLow:    newPool(cfg.Low, "low"),
 		},
-		producer: producer,
+		pub:      pub,
 		resolver: resolver,
 		log:      log,
 	}
@@ -142,7 +145,7 @@ func NewPoolManager(
 	// 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을 시작합니다.
diff --git a/internal/processor/fetcher/worker/pool.go b/internal/processor/fetcher/worker/pool.go
index e4295e03..cb68b14f 100644
--- a/internal/processor/fetcher/worker/pool.go
+++ b/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,8 +110,8 @@ 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,
@@ -119,7 +119,7 @@ func NewKafkaConsumerPool(
 	// 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 로,
diff --git a/internal/publisher/publisher.go b/internal/publisher/publisher.go
index 74d8a9f4..4813ebfc 100644
--- a/internal/publisher/publisher.go
+++ b/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()
diff --git a/internal/publisher/retry.go b/internal/publisher/retry.go
index 0517072e..b4ef0569 100644
--- a/internal/publisher/retry.go
+++ b/internal/publisher/retry.go
@@ -10,7 +10,6 @@ import (
 
 	"issuetracker/internal/processor/fetcher/core"
 	"issuetracker/pkg/logger"
-	"issuetracker/pkg/queue"
 	pkgredis "issuetracker/pkg/redis"
 )
 
@@ -54,43 +53,40 @@ type RetryScheduler interface {
 // 본 구현은 지적한 처리량 급감을 그대로 갖지만, Redis 미설정 환경 (단일 인스턴스
 // 개발/테스트, 통합 redis 장애) 에서 retry 자체는 동작하도록 보존합니다.
 type KafkaImmediateRetryScheduler struct {
-	producer queue.Producer
+	pub *Publisher
 }
 
-// NewKafkaImmediateRetryScheduler 는 KafkaImmediateRetryScheduler 를 생성합니다.
-func NewKafkaImmediateRetryScheduler(producer queue.Producer) *KafkaImmediateRetryScheduler {
-	return &KafkaImmediateRetryScheduler{producer: producer}
+// NewKafkaImmediateRetryScheduler 는 KafkaImmediateRetryScheduler 를 생성합니다 (이슈 #390 —
+// 구 queue.Producer 직접 주입 → publisher facade 주입으로 변경. Kafka I/O 단일 책임 원칙
+// 일관성 보장).
+func NewKafkaImmediateRetryScheduler(pub *Publisher) *KafkaImmediateRetryScheduler {
+	return &KafkaImmediateRetryScheduler{pub: pub}
 }
 
 // Enqueue 는 job 을 priority 토픽에 즉시 publish 합니다.
+//
+// gemini PR #400 피드백 — buildMessage 를 재사용하여 crawler / priority 기본 헤더 누락 + 코드
+// 중복을 한꺼번에 해소. retry 전용 헤더 (retry-count / last-error) 는 그 위에 덮어쓰기.
 func (s *KafkaImmediateRetryScheduler) Enqueue(ctx context.Context, job *core.CrawlJob, lastErr error) error {
-	data, err := job.Marshal()
+	msg, err := s.pub.buildMessage(job)
 	if err != nil {
-		return fmt.Errorf("marshal job for retry: %w", err)
-	}
-
-	msg := queue.Message{
-		Topic:   CrawlTopic(job.Priority),
-		Key:     []byte(job.ID),
-		Value:   data,
-		Headers: retryHeaders(job, lastErr),
+		return fmt.Errorf("build retry message: %w", err)
 	}
+	applyRetryHeaders(msg.Headers, job, lastErr)
 
-	if err := s.producer.Publish(ctx, msg); err != nil {
+	if err := s.pub.Forward(ctx, msg); err != nil {
 		return fmt.Errorf("publish retry job %s: %w", job.ID, err)
 	}
 	return nil
 }
 
-// retryHeaders 는 retry-count / last-error 표준 헤더를 구성합니다 — 두 구현체가 공유.
-func retryHeaders(job *core.CrawlJob, lastErr error) map[string]string {
-	h := map[string]string{
-		"retry-count": fmt.Sprintf("%d", job.RetryCount),
-	}
+// applyRetryHeaders 는 buildMessage 가 부착한 기본 헤더 위에 retry 전용 헤더를 덮어씁니다 —
+// 두 RetryScheduler 구현체가 공유.
+func applyRetryHeaders(h map[string]string, job *core.CrawlJob, lastErr error) {
+	h["retry-count"] = fmt.Sprintf("%d", job.RetryCount)
 	if lastErr != nil {
 		h["last-error"] = lastErr.Error()
 	}
-	return h
 }
 
 // ─────────────────────────────────────────────────────────────────────────────
@@ -150,24 +146,25 @@ func DefaultRedisRetrySchedulerConfig() RedisRetrySchedulerConfig {
 //
 // 라이프사이클: New → Run(ctx) (별도 goroutine) → ctx cancel → goroutine 정리 → Stop 대기.
 type RedisDelayedRetryScheduler struct {
-	client   retryQueueClient
-	producer queue.Producer
-	cfg      RedisRetrySchedulerConfig
-	log      *logger.Logger
-	wg       sync.WaitGroup
+	client retryQueueClient
+	pub    *Publisher
+	cfg    RedisRetrySchedulerConfig
+	log    *logger.Logger
+	wg     sync.WaitGroup
 
 	// idleTicks: 연속 idle (due 0) tick 수. due 발견 시 0 으로 reset.
 	// pollOnce 만 접근 — goroutine 단일 진입이라 race 없음.
 	idleTicks int
 }
 
-// NewRedisDelayedRetryScheduler 는 RedisDelayedRetryScheduler 를 생성합니다.
+// NewRedisDelayedRetryScheduler 는 RedisDelayedRetryScheduler 를 생성합니다 (이슈 #390 —
+// 구 queue.Producer 직접 주입 → publisher facade 주입으로 변경).
 //
 // 보정 규칙:
 //   - PollInterval / BatchSize / RepublishFailureBackoff: 0 또는 음수 → default 적용.
 //   - HeartbeatEveryNIdleTicks: **0 은 legacy (매 tick 로깅) 로 유효한 값** 이므로
 //     음수일 때만 default (60) 로 보정. 따라서 0 을 명시하면 default 가 적용되지 않음.
-func NewRedisDelayedRetryScheduler(client retryQueueClient, producer queue.Producer, cfg RedisRetrySchedulerConfig, log *logger.Logger) *RedisDelayedRetryScheduler {
+func NewRedisDelayedRetryScheduler(client retryQueueClient, pub *Publisher, cfg RedisRetrySchedulerConfig, log *logger.Logger) *RedisDelayedRetryScheduler {
 	def := DefaultRedisRetrySchedulerConfig()
 	if cfg.PollInterval <= 0 {
 		cfg.PollInterval = def.PollInterval
@@ -182,10 +179,10 @@ func NewRedisDelayedRetryScheduler(client retryQueueClient, producer queue.Produ
 		cfg.HeartbeatEveryNIdleTicks = def.HeartbeatEveryNIdleTicks
 	}
 	return &RedisDelayedRetryScheduler{
-		client:   client,
-		producer: producer,
-		cfg:      cfg,
-		log:      log,
+		client: client,
+		pub:    pub,
+		cfg:    cfg,
+		log:    log,
 	}
 }
 
@@ -337,14 +334,20 @@ func (s *RedisDelayedRetryScheduler) republish(ctx context.Context, item pkgredi
 		lastErr = errors.New(entry.LastErr)
 	}
 
-	msg := queue.Message{
-		Topic:   CrawlTopic(job.Priority),
-		Key:     []byte(item.JobID),
-		Value:   entry.JobBytes,
-		Headers: retryHeaders(&job, lastErr),
+	// gemini PR #400 피드백 — buildMessage 재사용으로 crawler/priority 기본 헤더 부착.
+	// Value 는 이미 Redis 에 보관된 entry.JobBytes 를 그대로 사용 (re-marshal 회피).
+	msg, err := s.pub.buildMessage(&job)
+	if err != nil {
+		s.log.WithFields(map[string]interface{}{
+			"job_id": item.JobID,
+		}).WithError(err).Error("failed to build retry message, dropping")
+		s.ackOrLog(ctx, item.JobID, "drop on build message failure")
+		return
 	}
+	msg.Value = entry.JobBytes
+	applyRetryHeaders(msg.Headers, &job, lastErr)
 
-	if err := s.producer.Publish(ctx, msg); err != nil {
+	if err := s.pub.Forward(ctx, msg); err != nil {
 		// Kafka publish 실패 — backoff 적용한 ScheduledAt 으로 EnqueueRetry 재호출하여
 		// 같은 jobID 의 score 를 미래로 overwrite (즉시 재peek 회피).
 		// EnqueueRetry 가 실패해도 ZSET 의 원본 항목이 남아있어 다음 폴에 재peek + 재시도.
diff --git a/test/internal/processor/fetcher/worker/pool_gate_test.go b/test/internal/processor/fetcher/worker/pool_gate_test.go
index 7893ccdf..ef6a9b1e 100644
--- a/test/internal/processor/fetcher/worker/pool_gate_test.go
+++ b/test/internal/processor/fetcher/worker/pool_gate_test.go
@@ -21,7 +21,7 @@ func TestKafkaConsumerPool_Gate_BlocksRSSURL_CommitsAndSkipsHandler(t *testing.T
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 	gate, _ := urlguard.NewGate(urlguard.Default(), logger.New(logger.DefaultConfig()))
 	pool.SetGate(gate)
 
@@ -49,7 +49,7 @@ func TestKafkaConsumerPool_Gate_AllowsArticleURL_DelegatesToHandler(t *testing.T
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 	gate, _ := urlguard.NewGate(urlguard.Default(), logger.New(logger.DefaultConfig()))
 	pool.SetGate(gate)
 
@@ -77,7 +77,7 @@ func TestKafkaConsumerPool_NoGate_LegacyBehavior(t *testing.T) {
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 	// SetGate 호출 없음
 
 	job := newTestJob()
@@ -104,7 +104,7 @@ func TestKafkaConsumerPool_Gate_AllowAllGuard_DelegatesAll(t *testing.T) {
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 	gate, _ := urlguard.NewGate(urlguard.AllowAllGuard{}, logger.New(logger.DefaultConfig()))
 	pool.SetGate(gate)
 
@@ -132,7 +132,7 @@ func TestKafkaConsumerPool_Gate_RaceFreeUpdate(t *testing.T) {
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 
 	// 동시에 여러 번 SetGate 호출 — atomic.Pointer 가 보장
 	g1, _ := urlguard.NewGate(urlguard.Default(), nil)
diff --git a/test/internal/processor/fetcher/worker/pool_retry_scheduler_test.go b/test/internal/processor/fetcher/worker/pool_retry_scheduler_test.go
index a991452d..a08db8a6 100644
--- a/test/internal/processor/fetcher/worker/pool_retry_scheduler_test.go
+++ b/test/internal/processor/fetcher/worker/pool_retry_scheduler_test.go
@@ -59,10 +59,11 @@ func TestKafkaConsumerPool_SetRetryScheduler_BypassesInlinePublish(t *testing.T)
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pub := newTestPublisher(producer)
+	pool := worker.NewKafkaConsumerPool(consumer, pub, handler, contentSvc, 1)
 
 	q := newPoolRetryFakeQueue()
-	customScheduler := publisher.NewRedisDelayedRetryScheduler(q, producer, publisher.RedisRetrySchedulerConfig{
+	customScheduler := publisher.NewRedisDelayedRetryScheduler(q, pub, publisher.RedisRetrySchedulerConfig{
 		PollInterval:            time.Hour,
 		BatchSize:               1,
 		RepublishFailureBackoff: time.Hour,
diff --git a/test/internal/processor/fetcher/worker/pool_test.go b/test/internal/processor/fetcher/worker/pool_test.go
index 1b30acf3..804b41c3 100644
--- a/test/internal/processor/fetcher/worker/pool_test.go
+++ b/test/internal/processor/fetcher/worker/pool_test.go
@@ -12,11 +12,21 @@ import (
 
 	"issuetracker/internal/processor/fetcher/core"
 	"issuetracker/internal/processor/fetcher/worker"
+	"issuetracker/internal/publisher"
 	"issuetracker/internal/storage"
 	"issuetracker/internal/storage/service"
+	"issuetracker/pkg/logger"
 	"issuetracker/pkg/queue"
 )
 
+// newTestPublisher 는 pool/manager 가 publisher facade 만 의존하도록 변경된 후 (이슈 #390)
+// 기존 mockProducer 검증 흐름을 유지하기 위한 thin helper 입니다.
+// 실제 *publisher.Publisher 를 생성하되 내부 producer 로 mockProducer 를 주입 — pool 이
+// pub.Forward → producer.Publish 로 위임하므로 mock 의 expectations 가 그대로 작동.
+func newTestPublisher(producer queue.Producer) *publisher.Publisher {
+	return publisher.New(producer, nil, logger.New(logger.DefaultConfig()))
+}
+
 // ─────────────────────────────────────────────────────────────────────────────
 // Mock 구현체
 // ─────────────────────────────────────────────────────────────────────────────
@@ -186,7 +196,7 @@ func TestKafkaConsumerPool_ProcessJob_PublishesToNormalized(t *testing.T) {
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 
 	job := newTestJob()
 	content := newTestContent()
@@ -223,7 +233,7 @@ func TestKafkaConsumerPool_ProcessJob_MultipleContents_PublishesAll(t *testing.T
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 
 	job := newTestJob()
 	c1 := newTestContent()
@@ -260,7 +270,7 @@ func TestKafkaConsumerPool_ProcessJob_EmptyContents_CommitsOnly(t *testing.T) {
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 
 	job := newTestJob()
 	msg := marshaledJobMsg(t, job)
@@ -283,7 +293,7 @@ func TestKafkaConsumerPool_ProcessJob_HandlerError_SendsToDLQ(t *testing.T) {
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 
 	job := newTestJob()
 	job.RetryCount = job.MaxRetries // 이미 최대 재시도 도달
@@ -313,7 +323,7 @@ func TestKafkaConsumerPool_ProcessJob_HandlerError_RequeuesAndCommits(t *testing
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 
 	job := newTestJob()
 	job.RetryCount = 1 // MaxRetries(3) 미달
@@ -345,7 +355,7 @@ func TestKafkaConsumerPool_PollMessages_UnmarshalFails_DLQSuccess_Commits(t *tes
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 
 	// 유효하지 않은 JSON을 담은 메시지로 unmarshal 실패 유발
 	malformed := &queue.Message{
@@ -375,7 +385,7 @@ func TestKafkaConsumerPool_PollMessages_UnmarshalFails_DLQFails_SkipsCommit(t *t
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 
 	malformed := &queue.Message{
 		Topic: queue.TopicCrawlNormal,
@@ -410,7 +420,7 @@ func TestKafkaConsumerPool_ProcessJob_CircuitOpen_SendsToDLQ(t *testing.T) {
 	// 첫 번째 실패로 circuit을 미리 open 상태로 만듭니다.
 	cbRegistry.Get("test-crawler").RecordFailure()
 
-	pool := worker.NewKafkaConsumerPoolWithCB(consumer, producer, handler, contentSvc, 1, cbRegistry)
+	pool := worker.NewKafkaConsumerPoolWithCB(consumer, newTestPublisher(producer), handler, contentSvc, 1, cbRegistry)
 
 	job := newTestJob()
 	msg := marshaledJobMsg(t, job)
@@ -436,7 +446,7 @@ func TestKafkaConsumerPool_ProcessJob_DLQPublishFails_SkipsCommit(t *testing.T)
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 
 	job := newTestJob()
 	job.RetryCount = job.MaxRetries // DLQ 경로
@@ -464,7 +474,7 @@ func TestKafkaConsumerPool_ProcessJob_RequeuePublishFails_SkipsCommit(t *testing
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 
 	job := newTestJob()
 	job.RetryCount = 1 // MaxRetries(3) 미달 → requeue 경로
@@ -492,7 +502,7 @@ func TestKafkaConsumerPool_ProcessJob_NormalizedMessageHasContentRef(t *testing.
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 
 	job := newTestJob()
 	content := newTestContent()
@@ -540,7 +550,7 @@ func TestKafkaConsumerPool_CommitMessage_DrainRetry_OnCanceled_Succeeds(t *testi
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 
 	job := newTestJob()
 	content := newTestContent()
@@ -570,7 +580,7 @@ func TestKafkaConsumerPool_SendToDLQ_DrainRetry_OnCanceled_Succeeds(t *testing.T
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 
 	job := newTestJob()
 	job.RetryCount = job.MaxRetries // DLQ 경로 강제
@@ -601,7 +611,7 @@ func TestKafkaConsumerPool_RequeueWithRetry_DrainRetry_OnCanceled_Succeeds(t *te
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 
 	job := newTestJob()
 	job.RetryCount = 1 // requeue 경로 (MaxRetries 미달)
@@ -635,7 +645,7 @@ func TestKafkaConsumerPool_CommitMessage_DrainRetry_BothFail_StillTwoCalls(t *te
 	handler := new(mockJobHandler)
 	contentSvc := new(mockContentService)
 
-	pool := worker.NewKafkaConsumerPool(consumer, producer, handler, contentSvc, 1)
+	pool := worker.NewKafkaConsumerPool(consumer, newTestPublisher(producer), handler, contentSvc, 1)
 
 	job := newTestJob()
 	content := newTestContent()
diff --git a/test/internal/processor/fetcher/worker/processing_lock_pool_test.go b/test/internal/processor/fetcher/worker/processing_lock_pool_test.go
index bfd5cd39..74ef8251 100644
--- a/test/internal/processor/fetcher/worker/processing_lock_pool_test.go
+++ b/test/internal/processor/fetcher/worker/processing_lock_pool_test.go
@@ -72,7 +72,7 @@ func TestKafkaConsumerPool_StageGate_AlreadyAcquired_SkipsWithoutCommit(t *testi
 	gate := new(mockStageGate)
 
 	pool := worker.NewKafkaConsumerPoolWithOptions(
-		consumer, producer, handler, contentSvc, 1,
+		consumer, newTestPublisher(producer), handler, contentSvc, 1,
 		worker.NewCircuitBreakerRegistry(worker.DefaultCircuitBreakerConfig, nil),
 		gate,
 	)
@@ -103,7 +103,7 @@ func TestKafkaConsumerPool_StageGate_Acquired_ReleasedAfterProcessing(t *testing
 	gate := new(mockStageGate)
 
 	pool := worker.NewKafkaConsumerPoolWithOptions(
-		consumer, producer, handler, contentSvc, 1,
+		consumer, newTestPublisher(producer), handler, contentSvc, 1,
 		worker.NewCircuitBreakerRegistry(worker.DefaultCircuitBreakerConfig, nil),
 		gate,
 	)
@@ -141,7 +141,7 @@ func TestKafkaConsumerPool_StageGate_AcquireError_ProceedsWithoutGate(t *testing
 	gate := new(mockStageGate)
 
 	pool := worker.NewKafkaConsumerPoolWithOptions(
-		consumer, producer, handler, contentSvc, 1,
+		consumer, newTestPublisher(producer), handler, contentSvc, 1,
 		worker.NewCircuitBreakerRegistry(worker.DefaultCircuitBreakerConfig, nil),
 		gate,
 	)
diff --git a/test/internal/publisher/retry_test.go b/test/internal/publisher/retry_test.go
index 89860700..4c0e87eb 100644
--- a/test/internal/publisher/retry_test.go
+++ b/test/internal/publisher/retry_test.go
@@ -41,6 +41,13 @@ func (m *retryMockProducer) Close() error {
 	return args.Error(0)
 }
 
+// retryTestPub 는 RetryScheduler 가 *publisher.Publisher 의존성으로 변경된 후 (이슈 #390)
+// 기존 mockProducer 검증 흐름을 유지하기 위한 thin helper 입니다.
+// retryMockProducer 를 내부 producer 로 가진 실제 *publisher.Publisher 를 생성합니다.
+func retryTestPub(producer *retryMockProducer) *publisher.Publisher {
+	return publisher.New(producer, nil, logger.New(logger.DefaultConfig()))
+}
+
 // retryTestJob 은 RetryScheduler 테스트 전용 fixture 입니다.
 func retryTestJob(priority core.Priority) *core.CrawlJob {
 	return &core.CrawlJob{
@@ -74,7 +81,7 @@ func TestKafkaImmediateRetryScheduler_PublishesToPriorityTopic(t *testing.T) {
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
 			producer := new(retryMockProducer)
-			sched := publisher.NewKafkaImmediateRetryScheduler(producer)
+			sched := publisher.NewKafkaImmediateRetryScheduler(retryTestPub(producer))
 
 			job := retryTestJob(tc.priority)
 			lastErr := errors.New("upstream 503")
@@ -97,7 +104,7 @@ func TestKafkaImmediateRetryScheduler_PublishesToPriorityTopic(t *testing.T) {
 // jobID 를 포함한 wrap 된 에러가 반환되는지 검증.
 func TestKafkaImmediateRetryScheduler_PublishError_Wrapped(t *testing.T) {
 	producer := new(retryMockProducer)
-	sched := publisher.NewKafkaImmediateRetryScheduler(producer)
+	sched := publisher.NewKafkaImmediateRetryScheduler(retryTestPub(producer))
 
 	job := retryTestJob(core.PriorityNormal)
 	publishErr := errors.New("kafka unavailable")
@@ -113,7 +120,7 @@ func TestKafkaImmediateRetryScheduler_PublishError_Wrapped(t *testing.T) {
 // last-error 헤더가 누락됨을 검증 (운영 보호 — nil deref 회피).
 func TestKafkaImmediateRetryScheduler_NilLastErr_OmitsHeader(t *testing.T) {
 	producer := new(retryMockProducer)
-	sched := publisher.NewKafkaImmediateRetryScheduler(producer)
+	sched := publisher.NewKafkaImmediateRetryScheduler(retryTestPub(producer))
 
 	job := retryTestJob(core.PriorityNormal)
 
@@ -260,7 +267,7 @@ func retrySchedTestLogger() *logger.Logger { return logger.New(logger.DefaultCon
 func TestRedisDelayedRetryScheduler_Enqueue_StoresJobAndLastErr(t *testing.T) {
 	q := newFakeRetryQueue()
 	prod := new(retryMockProducer)
-	sched := publisher.NewRedisDelayedRetryScheduler(q, prod, publisher.DefaultRedisRetrySchedulerConfig(), retrySchedTestLogger())
+	sched := publisher.NewRedisDelayedRetryScheduler(q, retryTestPub(prod), publisher.DefaultRedisRetrySchedulerConfig(), retrySchedTestLogger())
 
 	job := retryTestJob(core.PriorityNormal)
 	require.NoError(t, sched.Enqueue(context.Background(), job, errors.New("upstream 503")))
@@ -286,7 +293,7 @@ func TestRedisDelayedRetryScheduler_Run_PublishesDueItems(t *testing.T) {
 	prod := new(retryMockProducer)
 	cfg := publisher.DefaultRedisRetrySchedulerConfig()
 	cfg.PollInterval = 10 * time.Millisecond
-	sched := publisher.NewRedisDelayedRetryScheduler(q, prod, cfg, retrySchedTestLogger())
+	sched := publisher.NewRedisDelayedRetryScheduler(q, retryTestPub(prod), cfg, retrySchedTestLogger())
 
 	job := retryTestJob(core.PriorityHigh)
 	job.ScheduledAt = time.Now().Add(-time.Second)
@@ -319,7 +326,7 @@ func TestRedisDelayedRetryScheduler_Run_FutureItemsNotPublished(t *testing.T) {
 	prod := new(retryMockProducer)
 	cfg := publisher.DefaultRedisRetrySchedulerConfig()
 	cfg.PollInterval = 10 * time.Millisecond
-	sched := publisher.NewRedisDelayedRetryScheduler(q, prod, cfg, retrySchedTestLogger())
+	sched := publisher.NewRedisDelayedRetryScheduler(q, retryTestPub(prod), cfg, retrySchedTestLogger())
 
 	job := retryTestJob(core.PriorityNormal)
 	job.ScheduledAt = time.Now().Add(10 * time.Second)
@@ -346,7 +353,7 @@ func TestRedisDelayedRetryScheduler_AckOnlyAfterSuccessfulPublish(t *testing.T)
 	cfg := publisher.DefaultRedisRetrySchedulerConfig()
 	cfg.PollInterval = 10 * time.Millisecond
 	cfg.RepublishFailureBackoff = time.Hour
-	sched := publisher.NewRedisDelayedRetryScheduler(q, prod, cfg, retrySchedTestLogger())
+	sched := publisher.NewRedisDelayedRetryScheduler(q, retryTestPub(prod), cfg, retrySchedTestLogger())
 
 	job := retryTestJob(core.PriorityNormal)
 	job.ScheduledAt = time.Now().Add(-time.Second)
@@ -382,7 +389,7 @@ func TestRedisDelayedRetryScheduler_AckCalledAfterPublishSuccess(t *testing.T) {
 	prod := new(retryMockProducer)
 	cfg := publisher.DefaultRedisRetrySchedulerConfig()
 	cfg.PollInterval = 10 * time.Millisecond
-	sched := publisher.NewRedisDelayedRetryScheduler(q, prod, cfg, retrySchedTestLogger())
+	sched := publisher.NewRedisDelayedRetryScheduler(q, retryTestPub(prod), cfg, retrySchedTestLogger())
 
 	job := retryTestJob(core.PriorityNormal)
 	job.ScheduledAt = time.Now().Add(-time.Second)
@@ -410,7 +417,7 @@ func TestRedisDelayedRetryScheduler_Run_PublishFailure_ReEnqueues(t *testing.T)
 	cfg := publisher.DefaultRedisRetrySchedulerConfig()
 	cfg.PollInterval = 10 * time.Millisecond
 	cfg.RepublishFailureBackoff = 5 * time.Second
-	sched := publisher.NewRedisDelayedRetryScheduler(q, prod, cfg, retrySchedTestLogger())
+	sched := publisher.NewRedisDelayedRetryScheduler(q, retryTestPub(prod), cfg, retrySchedTestLogger())
 
 	job := retryTestJob(core.PriorityNormal)
 	job.ScheduledAt = time.Now().Add(-time.Second)
@@ -442,7 +449,7 @@ func TestRedisDelayedRetryScheduler_Run_PeekError_LogsAndContinues(t *testing.T)
 	prod := new(retryMockProducer)
 	cfg := publisher.DefaultRedisRetrySchedulerConfig()
 	cfg.PollInterval = 10 * time.Millisecond
-	sched := publisher.NewRedisDelayedRetryScheduler(q, prod, cfg, retrySchedTestLogger())
+	sched := publisher.NewRedisDelayedRetryScheduler(q, retryTestPub(prod), cfg, retrySchedTestLogger())
 
 	ctx, cancel := context.WithCancel(context.Background())
 	done := make(chan struct{})
@@ -466,7 +473,7 @@ func TestRedisDelayedRetryScheduler_RepublishOnShutdown_UsesDrainCtx(t *testing.
 	cfg := publisher.DefaultRedisRetrySchedulerConfig()
 	cfg.PollInterval = 10 * time.Millisecond
 	cfg.RepublishFailureBackoff = 5 * time.Second
-	sched := publisher.NewRedisDelayedRetryScheduler(q, prod, cfg, retrySchedTestLogger())
+	sched := publisher.NewRedisDelayedRetryScheduler(q, retryTestPub(prod), cfg, retrySchedTestLogger())
 
 	job := retryTestJob(core.PriorityNormal)
 	job.ScheduledAt = time.Now().Add(-time.Second)
@@ -498,7 +505,7 @@ func TestRedisDelayedRetryScheduler_StartStop_NoWaitGroupPanic(t *testing.T) {
 	prod := new(retryMockProducer)
 	cfg := publisher.DefaultRedisRetrySchedulerConfig()
 	cfg.PollInterval = 10 * time.Millisecond
-	sched := publisher.NewRedisDelayedRetryScheduler(q, prod, cfg, retrySchedTestLogger())
+	sched := publisher.NewRedisDelayedRetryScheduler(q, retryTestPub(prod), cfg, retrySchedTestLogger())
 
 	ctx, cancel := context.WithCancel(context.Background())
 	sched.Start(ctx)
@@ -523,7 +530,7 @@ func TestRedisDelayedRetryScheduler_StartStop_NoWaitGroupPanic(t *testing.T) {
 func TestRedisDelayedRetryScheduler_DefaultConfigBoundary(t *testing.T) {
 	q := newFakeRetryQueue()
 	prod := new(retryMockProducer)
-	sched := publisher.NewRedisDelayedRetryScheduler(q, prod, publisher.RedisRetrySchedulerConfig{}, retrySchedTestLogger())
+	sched := publisher.NewRedisDelayedRetryScheduler(q, retryTestPub(prod), publisher.RedisRetrySchedulerConfig{}, retrySchedTestLogger())
 
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
@@ -585,7 +592,7 @@ func TestRedisDelayedRetryScheduler_IdleHeartbeat_CompressedEveryN(t *testing.T)
 	cfg := publisher.DefaultRedisRetrySchedulerConfig()
 	cfg.PollInterval = 10 * time.Millisecond
 	cfg.HeartbeatEveryNIdleTicks = 5
-	sched := publisher.NewRedisDelayedRetryScheduler(q, prod, cfg, captureDebugLogger(buf))
+	sched := publisher.NewRedisDelayedRetryScheduler(q, retryTestPub(prod), cfg, captureDebugLogger(buf))
 
 	ctx, cancel := context.WithCancel(context.Background())
 	done := make(chan struct{})
@@ -620,7 +627,7 @@ func TestRedisDelayedRetryScheduler_IdleHeartbeat_LegacyEveryTick(t *testing.T)
 	cfg := publisher.DefaultRedisRetrySchedulerConfig()
 	cfg.PollInterval = 10 * time.Millisecond
 	cfg.HeartbeatEveryNIdleTicks = 0
-	sched := publisher.NewRedisDelayedRetryScheduler(q, prod, cfg, captureDebugLogger(buf))
+	sched := publisher.NewRedisDelayedRetryScheduler(q, retryTestPub(prod), cfg, captureDebugLogger(buf))
 
 	ctx, cancel := context.WithCancel(context.Background())
 	done := make(chan struct{})
@@ -651,7 +658,7 @@ func TestRedisDelayedRetryScheduler_IdleTicksResetOnDue(t *testing.T) {
 	cfg := publisher.DefaultRedisRetrySchedulerConfig()
 	cfg.PollInterval = 10 * time.Millisecond
 	cfg.HeartbeatEveryNIdleTicks = 100
-	sched := publisher.NewRedisDelayedRetryScheduler(q, prod, cfg, captureDebugLogger(buf))
+	sched := publisher.NewRedisDelayedRetryScheduler(q, retryTestPub(prod), cfg, captureDebugLogger(buf))
 
 	prod.On("Publish", mock.Anything, mock.Anything).Return(nil)