Kafka DLT (Dead Letter Topic)

원본 토픽: realtime-transaction-batch
DLT 토픽: realtime-transaction-batch.DLT

Producer → 원본 토픽 → Consumer (메인 리스너)
                         ↓ 실패
                    DefaultErrorHandler
                         ↓ 재시도 (FixedBackOff)
                         ↓ 최종 실패
                    DeadLetterPublishingRecoverer
                         ↓
                    {topic}.DLT → DLT 전용 리스너

TopicBuilder
    .name(topic + ".DLT")
    .partitions(1)           // 실패 메시지는 소량, 순서 보장 유리
    .replicas(3)             // 유실 방지
    .config(TopicConfig.RETENTION_MS_CONFIG, Duration.ofDays(14).toMillis())
    .config(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2")
    .build();

@Bean("dltKafkaListenerContainerFactory")
public ConcurrentKafkaListenerContainerFactory<String, Object> dltFactory(
    ConsumerFactory<String, Object> consumerFactory) {
    var factory = new ConcurrentKafkaListenerContainerFactory<String, Object>();
    factory.setConsumerFactory(consumerFactory);
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
    // DLT에서는 재발행 없이 로그만
    factory.setCommonErrorHandler(new DefaultErrorHandler(new FixedBackOff(0L, 0L)));
    return factory;
}

@Header(name = KafkaHeaders.EXCEPTION_MESSAGE, required = false) String errorMessage

@Bean
public NewTopic realtimeTransactionDltTopic() {
    return TopicBuilder
        .name(topic + ".DLT")
        .partitions(1)
        .replicas(3)
        .config(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(Duration.ofDays(14).toMillis()))
        .config(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2")
        .build();
}

catch (Exception e) {
    log.error("[DLT] 최종 실패 - txNo: {}", dto.transactionNo(), e);
    telegramProvider.sendAlert(...);  // 반드시 알림
    acknowledgment.acknowledge();     // 더 이상 재시도 없음
}

// 1초 간격, 최대 2회 재시도 → 실패 시 DLT
new DefaultErrorHandler(
    new DeadLetterPublishingRecoverer(kafkaTemplate),
    new FixedBackOff(1000L, 2L)
);