🤔 메시지가 사라졌다! — Redis Pub/Sub -> Kafka로 교체한 이유 그리고 DLT를 하는 이유

배경: 배치 트리거는 어떻게 동작했는가

우리 서비스에는 대용량 이미지를 처리하는 Spring Batch 파이프라인이 있다. 사용자가 ZIP 파일을 올리면, API 서버(app-api)가 파일을 임시 저장하고 배치 서버(app-batch)에 "처리 시작해" 라는 신호를 보낸다.

이 신호를 처음에는 Redis Pub/Sub으로 구현했다.

사용자 업로드
    │
    ▼
app-api → Redis PUBLISH "batch:pending" {batchJobId}
    │
    ▼ (구독 중인 경우)
app-batch → RedisBatchEventListener.onMessage() → Spring Batch Job 실행

직관적이고 구현도 빠르다. 그리고 당분간은 잘 동작했다.

---

문제 발견: "배치가 왜 안 시작되죠?"

어느 날 QA에서 버그 리포트가 들어왔다.

"파일 업로드는 됐는데 처리가 시작이 안 됩니다."

로그를 보니 API 서버는 정상적으로 이벤트를 발행했다. 그런데 배치 서버가 처리한 흔적이 없었다. 배치 서버를 재시작한 직후에 올라온 업로드였다.

원인은 Redis Pub/Sub의 핵심 특성 때문이었다.

Redis Pub/Sub의 치명적 한계: Fire-and-Forget

API 서버:   PUBLISH batch:pending → 성공 (Redis 응답: 0 — 구독자 0명)
배치 서버:  [재시작 중... 아직 Subscribe 안 함]

→ 메시지는 Redis에 저장되지 않고 즉시 소멸

Redis Pub/Sub은 카카오톡 라이브 채팅과 같다.
채팅방에 아무도 없을 때 메시지를 보내면 💥 메시지가 사라진다!

Redis는 "구독자가 없음" 이라는 응답(0)을 돌려줄 뿐, 메시지를 어디에도 보관하지 않는다.

우편함에 비유하면, Redis Pub/Sub은 우편함이 없다. 사람이 없으면 편지가 공중에서 사라진다.

상황	Redis Pub/Sub	Kafka
구독자가 없을 때 발행	메시지 소멸	토픽에 저장됨
구독자 재시작 후	이전 메시지 수신 불가	offset 이후 메시지 재수신
브로커 장애 후	메시지 유실	재기동 후 복구

---

임시방편: BatchRecoveryScheduler

문제를 인지한 후 30분마다 DB를 폴링하는 스케줄러를 만들었다.

// ❌ 임시방편 — 복잡성만 올라갔다
@Scheduled(fixedDelay = 1800000) // 30분
public void processDelayedJobs() {
    // PENDING 상태인 작업 조회 → 이벤트 재발행
    List<BatchFileJob> pendingJobs = batchFileJobRepository
        .findByStatusAndCreatedAtBefore(BatchProcessingStatus.PENDING, 
                                        LocalDateTime.now().minusMinutes(30));
    
    for (BatchFileJob job : pendingJobs) {
        batchEventPort.publishPendingEvent(job.getId());
    }
}

이것은 진짜 해결이 아니었다. 메시지가 유실된 배치 작업은 최대 30분 뒤에야 재처리가 시작된다. 사용자 입장에서는 업로드 후 30분 동안 아무 일도 일어나지 않는 것처럼 보인다.

그리고 코드도 복잡해졌다. "왜 스케줄러가 있어요?" 라는 질문에 "메시지 유실 보완용이에요"라고 답하는 상황이 됐다.

---

☑️ 왜 Kafka인가 — 의사결정

메시지 브로커 교체를 결정할 때 세 가지 옵션을 검토했다.

옵션 A: Redis Stream 사용

Redis 5.0에서 추가된 Redis Stream은 Kafka처럼 메시지를 영속적으로 저장한다.

장점: 이미 Redis를 쓰고 있으므로 인프라 추가 없음
단점: Redis Stream은 Redis가 단일 장애점이 됨 (우리는 이미 Redis를 JWT/알림에 사용)
      Redis 장애 시 배치 트리거 + JWT 인증 + 알림이 동시에 불능

옵션 B: DB 폴링 방식으로 전환

Redis를 아예 안 쓰고, 배치 서버가 주기적으로 DB에서 PENDING 작업을 가져간다.

장점: 인프라 추가 없음, 메시지 유실 없음 (DB에 상태 저장)
단점: 폴링 주기(예: 5초)마다 불필요한 DB 쿼리
      처리 시작 지연이 폴링 주기만큼 발생

옵션 C: Kafka (선택)

장점: At-Least-Once 보장 (offset 기반 재처리)
      배치 서버 재시작 후 안 읽은 메시지 자동 재처리
      30분 폴링 스케줄러 제거 가능
단점: 인프라 추가 (Kafka 컨테이너)
      학습 비용

선택 이유: 배치 서버는 가끔 재시작된다. 재시작마다 메시지를 잃으면 안 된다. Redis Stream은 기존 Redis 의존성을 더 무겁게 만들고, DB 폴링은 근본 해결이 아니다. Kafka는 인프라를 하나 더 띄우는 비용이 있지만, 메시지 보장의 근본 해결이다.

---

헥사고날 아키텍처가 진가를 발휘한 순간

교체 작업을 시작할 때 예상했던 것보다 훨씬 쉬웠다. 이유는 BatchEventPort 인터페이스 덕분이었다.

// core-domain — 변경 없음
public interface BatchEventPort {
    void publishPendingEvent(Long batchJobId);
    boolean isAvailable();
}

API 서버의 BatchFileUploadService는 BatchEventPort만 알고 있다. Redis인지 Kafka인지 모른다.

// BatchFileUploadService — 변경 없음
@Service
public class BatchFileUploadService {
    private final BatchEventPort batchEventPort; // ← 인터페이스만 의존

    private void publishBatchEvent(Long batchJobId) {
        batchEventPort.publishPendingEvent(batchJobId); // 구현체가 뭐든 상관없음
    }
}

Redis 구현체를 Kafka 구현체로 교체한 것이 전부였다.

// ❌ 삭제: RedisBatchEventAdapter.java
@Component
@Profile("redis")
public class RedisBatchEventAdapter implements BatchEventPort {
    private final RedisTemplate<String, Long> redisTemplate;
    
    @Override
    public void publishPendingEvent(Long batchJobId) {
        redisTemplate.convertAndSend("batch:pending", batchJobId);
    }
}

// ✅ 추가: KafkaBatchEventAdapter.java
@Component
@Profile({"local", "dev"})
public class KafkaBatchEventAdapter implements BatchEventPort {
    private final KafkaTemplate<String, Long> kafkaTemplate;
    
    @Override
    public void publishPendingEvent(Long batchJobId) {
        kafkaTemplate.send(KafkaTopics.BATCH_JOBS, batchJobId.toString(), batchJobId)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("❌ Kafka 이벤트 발행 실패 - batchJobId: {}", batchJobId, ex);
                    // PENDING 상태 유지 → 기동 시 복구 스케줄러가 안전망
                } else {
                    log.info("📢 Kafka 이벤트 발행 완료 - partition: {}, offset: {}",
                            result.getRecordMetadata().partition(),
                            result.getRecordMetadata().offset());
                }
            });
    }
}

비즈니스 로직(BatchFileUploadService) 변경 줄 수: 0줄

Port & Adapter 패턴은 이론적으로 "인프라를 갈아끼울 수 있다"고 하는데, 실제로 체험하기 전까진 반신반의했다. 이번 교체에서 처음으로 그 가치를 피부로 느꼈다.

---

새 모듈 구조: infrastructure-kafka

Kafka 관련 코드는 별도 모듈로 분리했다.

core/core-infrastructure/
├── infrastructure-redis/      ← 기존 유지 (JWT, 알림 용도)
└── infrastructure-kafka/      ← 신규 추가 (배치 이벤트 전용)
    └── src/main/java/
        ├── KafkaTopics.java           ← 토픽 이름 상수
        ├── config/
        │   └── KafkaProducerConfig.java
        └── adapter/
            └── KafkaBatchEventAdapter.java

중요한 점: infrastructure-kafka는 infrastructure-redis를 의존하지 않는다. 두 모듈은 완전히 독립적이다. Redis가 죽어도 Kafka는 살아있고, Kafka가 죽어도 Redis(JWT, 알림)는 살아있다.

---

Consumer 구현: 3단계 처리 흐름

배치 서버의 Kafka Consumer는 단순히 메시지를 받아서 Job을 실행하는 게 아니다. Race Condition이 있다.

API 서버: DB에 BatchFileJob 저장 → 트랜잭션 커밋
             ↓ (수 ms 이내)
          Kafka에 메시지 발행

배치 서버: Kafka 메시지 수신
             ↓ (즉시)
          DB에서 BatchFileJob 조회 → ???
          아직 API 서버 트랜잭션이 커밋 안 됐을 수 있음!

트랜잭션 커밋과 Kafka 메시지 발행 사이에 틈이 있다. 배치 서버가 메시지를 받는 속도가 더 빠를 수 있다.

// ✅ 3단계 처리로 Race Condition 해결
@KafkaListener(
    topics = KafkaTopics.BATCH_JOBS,
    groupId = "batch-worker",
    containerFactory = "batchKafkaListenerContainerFactory"
)
public void onBatchJobReceived(Long batchJobId, Acknowledgment ack) {
    try {
        // 1단계: DB 조회 — 최대 3회 재시도 (API 커밋 대기)
        BatchFileJob batchJob = findBatchJobWithRetry(batchJobId);
        
        // 2단계: 상태 변경 — REQUIRES_NEW 트랜잭션으로 즉시 커밋
        updateStatusToProcessing(batchJobId);
        
        // 3단계: Job 실행 — 트랜잭션 밖에서 실행
        launchBatchJob(batchJob);
        
        ack.acknowledge(); // ✅ 성공 시만 offset 커밋
    } catch (Exception e) {
        log.error("❌ 배치 이벤트 처리 실패 - batchJobId: {} | 재전달 예정", batchJobId, e);
        // ack 미호출 → Kafka가 동일 메시지를 재전달함 (At-Least-Once 보장)
    }
}

// DB 조회 재시도 (Race Condition 대응)
private BatchFileJob findBatchJobWithRetry(Long batchJobId) {
    for (int i = 0; i < 3; i++) {
        Optional<BatchFileJob> job = batchFileJobRepository.findById(batchJobId);
        if (job.isPresent()) return job.get();
        
        log.warn("⚠️ BatchFileJob 조회 실패 (재시도 {}/3) - batchJobId: {}", i + 1, batchJobId);
        Thread.sleep(500); // 500ms 대기 후 재시도
    }
    throw new IllegalStateException("BatchFileJob을 찾을 수 없습니다: " + batchJobId);
}

ack.acknowledge()를 수동으로 호출하는 것이 핵심이다. 성공했을 때만 offset을 커밋한다. 실패하면 Kafka가 같은 메시지를 재전달한다. 이것이 At-Least-Once 보장의 실제 동작이다.

---

⭐ DLT(Dead Letter Topic): 계속 실패하는 메시지는 어떻게 처리하나

네트워크 오류나 DB 일시 장애로 배치 처리가 실패하면 Kafka가 재전달한다. 그런데 파일이 손상됐거나 처리 자체가 불가능한 경우라면? 무한 재전달이 반복된다.

이를 막기 위해 DLT(Dead Letter Topic) 를 도입했다.

배치 처리 실패 (예외 발생 → ack 미호출)
    │
    ▼
DefaultErrorHandler 재시도: 3초 간격, 최대 3회
    │
    ├── 3회 재시도 중 성공 → 정상 처리
    │
    └── 3회 모두 실패
            ↓
    DeadLetterPublishingRecoverer → batch-jobs.DLT 토픽에 발행
            ↓
    DltBatchEventListener.onDltMessage() 수신
            ↓
    BatchFileJob.status = FAILED + errorMessage 기록
    ack.acknowledge() → 무한 루프 방지

// Consumer 설정 — 재시도 + DLT
@Bean
public DefaultErrorHandler errorHandler(KafkaTemplate<?, ?> kafkaTemplate) {
    // 재처리 불가 예외는 즉시 DLT로
    DeadLetterPublishingRecoverer recoverer = 
        new DeadLetterPublishingRecoverer(kafkaTemplate);
    
    // 3초 간격, 최대 3회 재시도
    FixedBackOff backOff = new FixedBackOff(3000L, 3L);
    
    return new DefaultErrorHandler(recoverer, backOff);
}

// DLT 리스너 — 최종 실패 처리
@KafkaListener(topics = "batch-jobs.DLT", groupId = "batch-worker-dlt")
public void onDltMessage(Long batchJobId, Acknowledgment ack) {
    try {
        BatchFileJob job = batchFileJobRepository.findById(batchJobId)
            .orElseThrow();
        job.updateStatus(BatchProcessingStatus.FAILED);
        job.updateErrorMessage("DLT 격리: 재시도 3회 소진");
        batchFileJobRepository.save(job);
        
        log.error("💀 DLT 처리 완료 - batchJobId: {} FAILED 확정", batchJobId);
    } finally {
        ack.acknowledge(); // 항상 호출 — DLT 무한 루프 방지
    }
}

재시도 계층을 3단계로 분리한 것도 의도적이다:

계층	위치	대상	횟수
내부 재시도	findBatchJobWithRetry()	API DB 커밋 대기 (Race Condition)	3회 / 500ms
외부 재시도	DefaultErrorHandler	네트워크/DB 일시 장애	3회 / 3000ms
DLT 격리	DltBatchEventListener	재시도 소진, 복구 불가	최종 확정

---

⭐ Kafka 설정: KRaft 모드 (Zookeeper 없음)

☑️ Kafka 3.x부터는 Zookeeper 없이 동작하는 KRaft(Kafka Raft) 모드를 지원한다. 개발/스테이징 환경에서는 컨테이너 하나로 충분하다.

# docker-compose-local.yml
kafka:
  image: apache/kafka:3.8.1
  environment:
    KAFKA_NODE_ID: 1
    KAFKA_PROCESS_ROLES: broker,controller  # 하나의 노드가 두 역할 담당
    KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
    KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
    KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
    KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
    KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
  healthcheck:
    test: ["/opt/kafka/bin/kafka-topics.sh", "--bootstrap-server", "localhost:9092", "--list"]
    interval: 10s
    retries: 5

Zookeeper + Kafka 두 컨테이너를 띄웠던 이전과 달리, 이제 Kafka 하나만 있으면 된다.

---

마이그레이션 결과 비교

항목	Redis Pub/Sub (이전)	Kafka (이후)
배치 서버 재시작 중 메시지	유실	offset에 보관 → 재시작 후 자동 처리
배치 시작 지연	최대 30분 (스케줄러)	수 초 이내
코드 복잡도	BatchRecoveryScheduler + Redis 리스너	Kafka 리스너만
처리 실패 시	로그만 남음	DLT → DB FAILED 확정
메시지 보장	Fire-and-Forget	At-Least-Once

BatchRecoveryScheduler의 30분 폴링 로직은 제거했다. 기동 시 누락된 QUEUED/IN_PROGRESS 작업을 복구하는 로직은 남겼다. Kafka는 메시지를 보장하지만, 앱 크래시로 인해 Kafka 메시지를 아예 수신 못 한 경우는 Kafka로 해결이 안 되기 때문이다.

---

마치며 — 이 경험에서 얻은 것

1. 메시지 브로커를 선택할 때는 "구독자가 없을 때 어떻게 되나"를 먼저 물어라

Redis Pub/Sub의 Fire-and-Forget 특성은 문서에 명시되어 있다. 처음 도입할 때 이것을 간과했다. "단순한 이벤트 전달"에 Redis를 쓰는 건 적합하지만, "반드시 처리되어야 하는 이벤트"에는 맞지 않는다.

2. Port & Adapter 패턴은 "인프라 교체"의 순간에 빛난다

이 교체를 통해 BatchEventPort 인터페이스가 왜 있어야 하는지를 실감했다. 인터페이스 없이 RedisTemplate을 서비스에서 직접 쓰고 있었다면, 교체 범위가 비즈니스 로직까지 퍼졌을 것이다.

3. 임시방편이 코드를 어떻게 썩히는지 보았다

BatchRecoveryScheduler는 합리적인 임시방편이었다. 그런데 그것이 있는 한 "메시지 유실 문제가 해결됐다"는 착각이 생긴다. 근본 원인을 해결하지 않고 위에 코드를 쌓으면, 나중에는 어느 코드가 왜 있는지 설명하기 어려워진다.

---