Kafka Interceptor로 MDC 로깅 분리하기

@Component
public class CouponIssueProducer {
    // ...
    public void issue(final UUID userId, final UUID couponId) {
        CouponIssueMessage payload = new CouponIssueMessage(userId, couponId);
        // 비즈니스 로직에 섞여 있는 MDC 코드
        String globalTraceId = MDC.get("globalTraceId");

        ProducerRecord<String, Object> record = new ProducerRecord<>(KafkaTopic.COUPON_ISSUE.getTopicName(), payload);

        // 헤더를 직접 추가하는 로직
        if (globalTraceId != null) {
            record.headers().add(GLOBAL_TRACE_ID_HEADER, globalTraceId.getBytes(StandardCharsets.UTF_8));
        }

        kafkaTemplate.send(record).whenComplete((result, ex) -> {
            //...
        }).join();
    }
}

@Component
public class CouponIssueConsumer {
    
    // ...
    @KafkaListener(topics = "...", groupId = "...")
    public void listener(final CouponIssueMessage message,
                         @Header(name = GLOBAL_TRACE_ID_KEY, required = false) final String globalTraceId,
                         final Acknowledgment ack) {
        
        // 비즈니스 로직에 섞여 있는 MDC 설정 코드
        if (!Objects.isNull(globalTraceId)) {
            MDC.put(GLOBAL_TRACE_ID_KEY, globalTraceId);
        }
        log.info("발급 처리 메시지 수신: {}", message);

        try {
            couponIssuanceService.process(message);
            ack.acknowledge();
        } catch (Exception e) {
            // ...
        } finally {
            // try-finally로 직접 MDC를 제거해야 하는 번거로움
            MDC.remove(GLOBAL_TRACE_ID_KEY);
        }
    }
}

@Configuration
public class KafkaProducerConfig {
    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        // ... (부트스트랩 서버, 직렬화 등 필수 설정)
        // 여기에 인터셉터 클래스를 지정해준다.
        config.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MdcProducerInterceptor.class.getName());
        return new DefaultKafkaProducerFactory<>(config);
    }
    // ...
}

// Spring Bean이 아니므로 @Component를 붙이지 않는다.
public class MdcProducerInterceptor implements ProducerInterceptor<String, Object> {

    private static final String GLOBAL_TRACE_ID_KEY = "globalTraceId";

    /**
     * Kafka 브로커로 메시지를 보내기 전에 호출합니다.
     */
    @Override
    public ProducerRecord<String, Object> onSend(ProducerRecord<String, Object> record) {
        String globalTraceId = MDC.get(GLOBAL_TRACE_ID_KEY);
        // MDC에서 Trace ID를 꺼내 헤더에 추가한다.
        if (globalTraceId != null) {
            record.headers().add(GLOBAL_TRACE_ID_KEY, globalTraceId.getBytes(StandardCharsets.UTF_8));
        }
        return record;
    }

    // 나머지 onAcknowledgement, close, configure 메서드는 비워둔다.
    // ...
}

@Component // Spring Bean으로 등록해서 의존성 주입이 가능하게 한다.
public class MdcConsumerRecordInterceptor implements RecordInterceptor<String, CouponIssueMessage> {
    // ... (내용은 아래에서 설명)
}

@Configuration
public class KafkaConsumerConfig {
    // ... (ConsumerFactory 설정 생략)

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, CouponIssueMessage> kafkaListenerContainerFactory(
            ConsumerFactory<String, CouponIssueMessage> consumerFactory,
            // RecordInterceptor를 Spring Bean으로 주입받는다.
            RecordInterceptor<String, CouponIssueMessage> mdcRecordInterceptor
    ) {
        ConcurrentKafkaListenerContainerFactory<String, CouponIssueMessage> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        // ... (AckMode, Concurrency 등 기타 설정)

        // (중요) ListenerContainerFactory에 인터셉터를 등록한다.
        factory.setRecordInterceptor(mdcRecordInterceptor);
        return factory;
    }
}

@Component // Spring Bean으로 등록해서 의존성 주입이 가능하게 한다.
public class MdcConsumerRecordInterceptor implements RecordInterceptor<String, CouponIssueMessage> {

    private static final String GLOBAL_TRACE_ID_KEY = "globalTraceId";

    // @KafkaListener 메서드가 실행되기 직전에 호출된다.
    @Override
    public ConsumerRecord<String, CouponIssueMessage> intercept(
            @NonNull ConsumerRecord<String, CouponIssueMessage> record,
            @NonNull Consumer<String, CouponIssueMessage> consumer) {
        // 메시지 헤더에서 Trace ID를 꺼내 MDC에 설정한다.
        Header header = record.headers().lastHeader(GLOBAL_TRACE_ID_KEY);
        if (Objects.nonNull(header)) {
            String globalTraceId = new String(header.value(), StandardCharsets.UTF_8);
            MDC.put(GLOBAL_TRACE_ID_KEY, globalTraceId);
        }
        return record;
    }

    // @KafkaListener 메서드가 성공적으로 완료된 후 호출된다. (예외 없을 때)
    @Override
    public void success(
            @NonNull ConsumerRecord<String, CouponIssueMessage> record,
            @NonNull Consumer<String, CouponIssueMessage> consumer) {
        // 여기서 MDC를 정리한다.
        MDC.remove(GLOBAL_TRACE_ID_KEY);
    }

    // @KafkaListener 메서드에서 예외가 발생했을 때 호출된다.
    @Override
    public void failure(
            @NonNull ConsumerRecord<String, CouponIssueMessage> record,
            @NonNull Exception exception,
            @NonNull Consumer<String, CouponIssueMessage> consumer) {
        // 여기서도 마찬가지로 MDC를 정리한다.
        MDC.remove(GLOBAL_TRACE_ID_KEY);
    }
}

@Component
public class CouponIssueProducer {
    // ...
    public void issue(final UUID userId, final UUID couponId) {
        CouponIssueMessage payload = new CouponIssueMessage(userId, couponId);
        ProducerRecord<String, Object> record = new ProducerRecord<>(KafkaTopic.COUPON_ISSUE.getTopicName(), payload);
        // MDC 관련 코드가 완전히 사라졌다!
        kafkaTemplate.send(record).whenComplete((result, ex) -> {
            // ...
        }).join();
    }
}

@Component
public class CouponIssueConsumer {
    // ...
    @KafkaListener(topics = "...", groupId = "...")
    public void listener(final CouponIssueMessage message, final Acknowledgment ack) {
        // MDC 설정, 헤더 파라미터, try-finally 구문이 모두 사라졌다!
        log.info("발급 처리 메시지 수신: {}", message);
        try {
            couponIssuanceService.process(message);
            ack.acknowledge();
        } catch (Exception e) {
            log.error("메시지 처리 실패, 재처리를 위해 커밋하지 않음: {}", message, e);
        }
    }
}