Kafka Producer를 제너럴하게 리팩토링

⚠️ 문제 상황

기존에는 Kafka 메시지를 전송하기 위해 CouponIssueProducer 내부에 쿠폰 발급 전용 전송 로직이 하드코딩되어 있었다.

// RedisQueueWorker
couponIssueProducer.sendIssueEvent(couponId, userId);

// CouponIssueProducer
private final KafkaTemplate<String, CouponIssueEventDto> kafkaTemplate;
private static final String TOPIC = "coupon.issue";

public void sendIssueEvent(Long couponId, Long userId)  {
    CouponIssueEventDto event = new CouponIssueEventDto(couponId, userId);

    kafkaTemplate.send(TOPIC, couponId.toString(), event)
            .toCompletableFuture().whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Kafka 메시지 전송 실패: couponId={}, userId={}, error={}", couponId, userId, ex.getMessage());
                } else {
                    log.info("[Kafka] 쿠폰 발급 이벤트 전송 : couponId={}, userId={}", couponId, userId);
                }
            });
}

이 방식은 특정 이벤트에만 고정되어 있어, 다른 이벤트(DLQ, 실패 이벤트 등)를 전송하려면 매번 새로운 메서드를 추가해야 했고 토픽 이름도 고정되어 있어 재사용성과 확장성에 제약이 있었다.

🔄️ 개선

이러한 문제를 해결하기 위해 Kafka 전송 로직을 제네릭 + 파라미터화된 토픽 구조로 리팩토링했다.

// RedisQueueWorker
couponIssueProducer.send("coupon.issue", String.valueOf(couponId), new CouponIssueEventDto(couponId, userId));

// CouponIssueProducer
private final KafkaTemplate<String, Object> kafkaTemplate;

public <T> void send(String topic, String key, T payload) {
    kafkaTemplate.send(topic, key, payload)
            .toCompletableFuture()
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Kafka 전송 실패: topic={}, key={}, payload={}, error={}", topic, key, payload, ex.getMessage());
                } else {
                    log.info("[Kafka] 쿠폰 발급 전송: topic={}, key={}, payload={}", topic, key, payload);
                }
            });
}

• 이벤트 타입 추상화
  기존에는 CouponIssueEventDto만 전송할 수 있었지만 제네릭 타입 <T>를 활용하면서 다양한 도메인의 이벤트 객체도 전송 가능해졌다. 이제 실패 이벤트나 사용자 가입 이벤트 같은 새로운 타입의 객체도 같은 메서드를 통해 전송할 수 있기 때문에 코드 재사용성과 확장성이 크게 향상되었다.

• 토픽 동적 처리
  기존에는 토픽이 coupon.issue로 고정되어 있어 다른 이벤트를 보내려면 토픽 이름을 변경하기 위한 별도 메서드가 필요했다. 하지만 이제는 호출 시점에 topic을 파라미터로 전달함으로써 다양한 목적의 메시지를 원하는 토픽으로 전송할 수 있게 되었고 운영 중인 시스템에서 새로운 Kafka 주제를 도입하거나 분리할 때 유연하게 대응할 수 있다.

• 코드 중복 제거
  이벤트 전송 로직을 하나의 send() 메서드로 통합하면서 기존처럼 이벤트마다 개별 메서드를 만들 필요가 없어졌다. 이는 유지보수 시 반복적인 코드 작성을 방지하고 이벤트 로직이 변경되더라도 단일 메서드만 수정하면 되어 관리가 훨씬 쉬워졌다.

• 로깅 및 예외 처리 일관성 확보
  Kafka 전송 결과에 대한 whenComplete() 블록을 통해 모든 전송 성공/실패 상황을 한 곳에서 처리할 수 있게 되었다. 각 이벤트마다 다른 방식으로 로깅하거나 예외를 다뤄야 했던 이전 방식보다 지금은 전송 결과를 일관되게 기록할 수 있어 디버깅이나 장애 분석 시에도 유리하다.

📝 배운점

Kafka 메시지 전송 로직을 추상화하면서 기존의 하드코딩된 방식의 한계점을 체감할 수 있었다. 이번 리팩토링을 통해 이벤트 기반 시스템을 확장성과 재사용성 측면에서 한 단계 더 유연하게 설계할 수 있었고 앞으로 Kafka를 사용하는 도메인이 많아질수록 이 구조가 큰 도움이 될 것이라 확신한다.

또한 단일 메서드로 이벤트 전송을 통일함으로써 코드 구조가 더욱 깔끔해졌고 장애 시 로깅과 대응도 쉬워지는 등 실질적인 운영 효율성도 함께 개선되었다.