Spring Boot에서 Kafka 메시지 송수신하기

1. 개요

Apache Kafka는 서비스 간 비동기 통신을 위한 인기 있는 이벤트 스트리밍 플랫폼입니다. Spring Boot는 Spring Kafka 라이브러리를 통해 Kafka를 지원하므로, KafkaProducer와 poll() 루프를 직접 관리할 필요가 없습니다.

이 글에서는 간단한 예제를 통해 배워봅니다: 주문 서비스가 OrderCreatedEvent 이벤트를 전송하고, 컨슈머가 이를 받아 재고를 예약합니다. 글을 다 읽으면 JSON 메시지 송수신 설정, Dead Letter Topic으로 에러 처리, Embedded Kafka로 테스트하는 방법을 알게 됩니다.

2. 준비하기

2.1. Maven 의존성

spring-kafka가 필요합니다. Spring Boot가 버전을 관리하므로 버전을 명시할 필요가 없습니다:

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

<!-- 테스트용 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>

spring-kafka-test는 테스트용 인메모리 Kafka 브로커를 제공합니다. Awaitility는 Thread.sleep() 없이 비동기 처리를 기다리는 데 도움이 됩니다.

2.2. 로컬에서 Kafka 실행

가장 빠른 방법은 Apache Kafka 공식 이미지를 KRaft 모드에서 사용하는 것입니다. 즉, ZooKeeper가 필요 없습니다:

docker run -d --name kafka -p 9092:9092 apache/kafka:latest

브로커는 localhost:9092에서 수신 대기합니다.

3. 데이터 모델 및 설정

전송할 이벤트는 간단한 record입니다:

package com.nonla.shop.order;

import java.math.BigDecimal;

public record OrderCreatedEvent(String orderId, String customerId, BigDecimal amount) {
}

다음으로 application.yml에서 설정합니다:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
    consumer:
      group-id: inventory-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.nonla.shop.order"

각 부분의 의미:

• bootstrap-servers는 브로커 주소입니다.
• Producer는 key에 StringSerializer, value에 JsonSerializer를 사용합니다. JsonSerializer는 Jackson을 사용하여 객체를 JSON으로 변환합니다.
• acks: all은 모든 복제본이 동기화되어 쓰기를 완료한 후에만 브로커가 확인응답하도록 강제합니다. 이 방식이 더 안전하지만 약간 느립니다.
• group-id는 컨슈머 그룹을 식별합니다. 같은 그룹의 인스턴스들은 파티션을 나누어 처리합니다.
• auto-offset-reset: earliest는 그룹에 저장된 오프셋이 없을 때 컨슈머가 토픽의 처음부터 읽도록 합니다.
• spring.json.trusted.packages는 JsonDeserializer가 객체를 생성할 수 있는 패키지를 선언합니다. 이 줄이 없으면 “The class … is not in the trusted packages” 에러가 발생합니다.

3.1. 토픽 자동 생성

Spring Boot는 자동으로 KafkaAdmin을 설정합니다. 이 bean은 컨텍스트에 선언된 모든 NewTopic을 생성합니다:

package com.nonla.shop.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

@Configuration
public class KafkaTopicConfig {

    public static final String ORDERS_TOPIC = "orders";

    @Bean
    public NewTopic ordersTopic() {
        return TopicBuilder.name(ORDERS_TOPIC)
                .partitions(3)
                .replicas(1)
                .build();
    }
}

애플리케이션 시작 시 orders 토픽(3개 파티션)이 존재하지 않으면 생성됩니다. 토픽이 이미 있으면 KafkaAdmin은 건너뛰고 기존 설정을 덮어쓰지 않습니다.

4. KafkaTemplate으로 메시지 전송

Spring Boot는 producer 설정을 바탕으로 KafkaTemplate bean을 자동으로 생성합니다. 주입하기만 하면 됩니다:

package com.nonla.shop.order;

import com.nonla.shop.config.KafkaTopicConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);

    private final KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public CompletableFuture<SendResult<String, OrderCreatedEvent>> publish(OrderCreatedEvent event) {
        return kafkaTemplate.send(KafkaTopicConfig.ORDERS_TOPIC, event.orderId(), event)
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("주문 {} 전송 실패", event.orderId(), ex);
                        return;
                    }
                    var metadata = result.getRecordMetadata();
                    log.info("주문 {}을 파티션 {} 오프셋 {}으로 전송했습니다",
                            event.orderId(), metadata.partition(), metadata.offset());
                });
    }
}

주의할 점들:

• send(topic, key, value)는 orderId를 key로 사용합니다. Kafka는 key를 해시하여 파티션을 선택하므로, 같은 주문의 모든 메시지는 같은 파티션으로 들어갑니다. 덕분에 순서가 유지됩니다.
• Spring Kafka 3.0부터 send()는 더 이상 사용 중단된 ListenableFuture 대신 CompletableFuture<SendResult>를 반환합니다.
• send()는 비동기입니다. 호출은 즉시 반환되고, whenComplete의 콜백은 브로커 응답 시 실행됩니다.

전송이 성공하면 로그는 다음과 같습니다:

INFO  c.n.s.order.OrderEventPublisher : 주문 ORD-1001을 파티션 2 오프셋 0으로 전송했습니다

배치 작업처럼 전송이 완료될 때까지 스레드를 블로킹해야 한다면 publish(event).get(5, TimeUnit.SECONDS)을 호출할 수 있습니다. 하지만 높은 부하의 요청 처리 스레드에서는 이렇게 하지 마세요. 각 호출이 스레드를 브로커 응답 대기에 묶어두기 때문입니다.

주의: 같은 메서드에서 데이터베이스에 쓰고 Kafka로 전송하는 것은 원자적 트랜잭션이 아닙니다. 데이터베이스 커밋은 성공했으나 Kafka 전송이 실패하면 데이터가 불일치합니다. 중요한 데이터의 경우 Transactional Outbox 패턴을 고려하세요.

5. @KafkaListener로 메시지 수신

컨슈머 쪽은 @KafkaListener 어노테이션이 있는 메서드만으로 충분합니다:

package com.nonla.shop.inventory;

import com.nonla.shop.config.KafkaTopicConfig;
import com.nonla.shop.order.OrderCreatedEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class OrderCreatedListener {

    private static final Logger log = LoggerFactory.getLogger(OrderCreatedListener.class);

    private final InventoryService inventoryService;

    public OrderCreatedListener(InventoryService inventoryService) {
        this.inventoryService = inventoryService;
    }

    @KafkaListener(topics = KafkaTopicConfig.ORDERS_TOPIC, concurrency = "3")
    public void onOrderCreated(@Payload OrderCreatedEvent event,
                               @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                               @Header(KafkaHeaders.OFFSET) long offset) {
        log.info("파티션 {} 오프셋 {}에서 주문 {} 수신", partition, offset, event.orderId());
        inventoryService.reserve(event);
    }
}

InventoryService는 일반적인 서비스일 뿐입니다:

package com.nonla.shop.inventory;

import com.nonla.shop.order.OrderCreatedEvent;
import org.springframework.stereotype.Service;

@Service
public class InventoryService {

    public void reserve(OrderCreatedEvent event) {
        if (event.amount().signum() <= 0) {
            throw new IllegalArgumentException("유효하지 않은 금액: " + event.amount());
        }
        // 재고 예약...
    }
}

설명:

• topics은 수신할 토픽을 지정합니다. groupId는 spring.kafka.consumer.group-id에서 가져오거나, 어노테이션에서 직접 지정할 수 있습니다.
• concurrency = "3"은 3개의 컨슈머 스레드를 생성하고, 각 스레드가 하나의 파티션을 처리합니다. 파티션 개수보다 큰 값은 무의미합니다. 초과 스레드는 아무것도 받지 못합니다.
• @Payload는 역직렬화된 객체를 수신합니다. @Header는 파티션, 오프셋 같은 메타데이터를 가져옵니다.
• 기본적으로 메서드가 exception을 던지지 않고 완료되면 오프셋이 커밋됩니다 (AckMode.BATCH). 따라서 애플리케이션이 중간에 죽으면 메시지가 다시 처리됩니다. 이는 at-least-once 의미론이므로, 로직이 멱등성을 가져야 합니다. 같은 orderId를 두 번 처리해도 재고를 두 번 예약하면 안 됩니다.

key, timestamp, 커스텀 header가 필요하면 파라미터로 ConsumerRecord<String, OrderCreatedEvent>를 전체 받을 수 있습니다.

6. 에러 처리 및 Dead Letter Topic

listener가 exception을 던지면, 기본 DefaultErrorHandler는 9회 재시도 후 로그를 남기고 메시지를 버립니다. 실제로는 보통 에러 메시지를 별도 토픽으로 옮겨서 나중에 검토합니다:

package com.nonla.shop.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

@Configuration
public class KafkaErrorConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> kafkaTemplate) {
        var recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
        var handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 2));
        handler.addNotRetryableExceptions(IllegalArgumentException.class);
        return handler;
    }
}

Spring Boot가 이 CommonErrorHandler bean을 기본 컨테이너 팩토리에 자동으로 연결합니다. 위 설정의 동작:

• FixedBackOff(1000L, 2)는 최대 2회 재시도하고, 매번 1초씩 간격을 둡니다.
• 재시도를 모두 소진하면 DeadLetterPublishingRecoverer가 메시지를 orders.DLT 토픽으로 전송합니다. 이름은 원본 토픽 이름에 .DLT 접미사를 붙인 것이고, 메시지는 원본 파티션을 유지합니다. 따라서 DLT 토픽은 최소한 원본 토픽과 같은 수의 파티션이 필요합니다.
• IllegalArgumentException은 데이터 에러이므로 재시도해도 소용없습니다. handler는 메시지를 바로 DLT로 보냅니다.

amount = -5인 메시지는 다음과 같은 로그를 출력합니다:

ERROR o.s.k.l.DefaultErrorHandler : Backoff none exhausted for orders-1@4
INFO  c.n.s.inventory.DltListener  : DLT의 에러 메시지: ORD-1002

주의: JSON 손상 같이 역직렬화 중에 발생한 에러는 listener에 도달하지 않습니다. 처리하지 않으면 컨슈머가 해당 메시지에서 영구적으로 멈출 수 있습니다. 이를 “독약(poison pill)“이라고 합니다. 역직렬화 래퍼를 ErrorHandlingDeserializer로 감싸세요:

spring:
  kafka:
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer

이 설정으로 역직렬화 에러도 DefaultErrorHandler로 전달되고, 에러 메시지도 DLT로 들어갑니다.

7. Embedded Kafka로 테스트하기

@EmbeddedKafka는 인메모리 브로커를 시작하므로 테스트 시 Docker가 필요 없습니다. 다음 예제는 전체 흐름을 테스트합니다: OrderEventPublisher로 전송한 후, InventoryService가 올바르게 이벤트를 받는지 확인합니다.

package com.nonla.shop.order;

import com.nonla.shop.inventory.InventoryService;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.context.bean.override.mockito.MockitoBean;

import java.math.BigDecimal;
import java.time.Duration;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.mockito.Mockito.verify;

@SpringBootTest(properties = "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}")
@EmbeddedKafka(partitions = 3, topics = "orders")
class OrderKafkaIntegrationTest {

    @Autowired
    private OrderEventPublisher publisher;

    @MockitoBean
    private InventoryService inventoryService;

    @Test
    void whenOrderPublished_thenListenerReservesInventory() throws Exception {
        var event = new OrderCreatedEvent("ORD-1001", "CUS-42", new BigDecimal("250000"));

        var result = publisher.publish(event).get();
        assertThat(result.getRecordMetadata().topic()).isEqualTo("orders");

        var captor = ArgumentCaptor.forClass(OrderCreatedEvent.class);
        await().atMost(Duration.ofSeconds(10))
                .untilAsserted(() -> verify(inventoryService).reserve(captor.capture()));

        assertThat(captor.getValue()).isEqualTo(event);
    }
}

각 부분의 역할:

• ${spring.embedded.kafka.brokers}는 @EmbeddedKafka가 제공하는 property입니다. 임베디드 브로커 주소를 포함하며, 이를 사용해 bootstrap-servers를 오버라이드합니다.
• @MockitoBean (Spring Boot 3.4부터)은 실제 InventoryService를 mock으로 대체하여 검증을 쉽게 합니다. 구버전에서는 @MockBean을 사용하세요.
• publish(event).get()은 브로커 확인응답을 기다리므로 메시지가 확실히 저장되었음을 보장합니다.
• Listener는 다른 스레드에서 실행되므로 await().untilAsserted(...)로 최대 10초 동안 verify를 재시도합니다.
• OrderCreatedEvent는 record이므로 equals()가 값으로 비교합니다. assertThat(...).isEqualTo(event)는 JSON이 올바르게 직렬화되고 역직렬화되었는지 확인합니다.

auto-offset-reset: earliest 설정 덕분에 테스트가 안정적입니다. 컨슈머가 그룹 join 후 메시지가 전송되도 메시지를 읽을 수 있습니다.

8. 흔한 에러와 Kafka를 쓰면 안 되는 경우

자주 만나는 에러들:

• trusted.packages 누락: JsonDeserializer가 미지의 클래스 생성을 거부합니다. 올바른 패키지를 선언하세요. 모든 producer를 제어할 때만 "*"을 사용하세요.
• Producer와 Consumer의 패키지 차이: JsonSerializer는 기본적으로 클래스 전체 이름을 __TypeId__ header에 씁니다. Consumer 서비스에 같은 이름이지만 다른 패키지의 클래스가 있으면 역직렬화 실패합니다. 이 경우 Consumer에서 spring.json.use.type.headers: false와 spring.json.value.default.type: com.nonla.inventory.OrderCreatedEvent를 설정하세요.
• Listener 실행 시간 초과: 한 번의 poll() 처리가 max.poll.interval.ms (기본값 5분)를 넘으면 브로커는 컨슈머를 죽은 것으로 봅니다. 결과적으로 rebalance가 일어나고 메시지가 중복 처리됩니다. max.poll.records를 줄이거나 로직을 최적화하세요.
• concurrency가 파티션 개수보다 큼: 초과 스레드는 유휴 상태입니다. 처리량을 늘리려면 먼저 파티션 개수를 늘려야 합니다.
• null key로 전송: 메시지가 파티션에 균등하게 분산되므로 엔티티별 순서 보장을 잃습니다.

Kafka가 항상 좋은 선택은 아닙니다:

• 동기 request/response가 필요하고 즉시 결과를 받아야 합니다. 이 경우 REST나 gRPC를 사용하세요.
• 시스템이 작고, 처리량이 낮으며, 이벤트 히스토리 보관이 불필요합니다. 이때 Kafka 클러스터 운영은 불필요한 비용이고, RabbitMQ나 데이터베이스 큐 테이블로 충분할 수 있습니다.
• 메시지별 복잡한 라우팅, 또는 메시지별로 지연이나 우선순위를 다르게 설정해야 합니다. 이는 전통적인 메시지 브로커의 강점입니다.

9. 결론

이 글에서는 Spring Kafka를 JSON 직렬화로 설정하고 NewTopic으로 토픽을 생성했습니다. KafkaTemplate으로 key를 포함한 메시지를 전송하고, @KafkaListener (concurrency 포함)로 수신했습니다. DefaultErrorHandler와 Dead Letter Topic으로 에러를 처리하고, @EmbeddedKafka와 Awaitility로 통합 테스트를 작성했습니다.

프로덕션 배포 시 세 가지를 잊지 마세요: listener는 멱등성을 가져야 하고, ErrorHandlingDeserializer로 역직렬화 에러를 처리해야 하며, 이벤트 전송이 데이터베이스의 데이터와 일관성을 유지해야 합니다.