Kafka 입문

1plugins { id 'java' }
2plugins { id 'org.springframework.boot' version '3.3.0' }
3
4dependencies {
5  implementation 'org.springframework.boot:spring-boot-starter-web'
6  implementation 'org.springframework.kafka:spring-kafka'
7  implementation 'org.springframework.boot:spring-boot-starter-validation'
8  testImplementation 'org.springframework.kafka:spring-kafka-test'
9}
10
11# application.yml
12spring.kafka.bootstrap-servers: localhost:9092
13spring.kafka.producer.key-serializer: org.apache.kafka.common.serialization.StringSerializer
14spring.kafka.producer.value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
15spring.kafka.consumer.group-id: notification-service
16spring.kafka.consumer.auto-offset-reset: earliest
17spring.kafka.consumer.key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
18spring.kafka.consumer.value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
19spring.kafka.consumer.properties.spring.json.trusted.packages: com.example.kafka
20spring.kafka.listener.ack-mode: manual

• 6번 Spring Boot에서 KafkaTemplate과 @KafkaListener를 쓰기 위한 핵심 의존성입니다.
• 12번 로컬 개발이면 보통 Docker Kafka의 9092 포트를 바라봅니다.
• 14, 18번 Java 객체를 JSON으로 보내고, Consumer에서 다시 객체로 받기 위한 설정입니다.
• 19번 JsonDeserializer가 역직렬화해도 되는 패키지를 제한합니다.
• 20번 Consumer가 처리를 끝낸 뒤 직접 offset commit을 호출하게 만듭니다.

1package com.example.kafka.order;
2
3import java.time.Instant;
4import java.util.UUID;
5
6public record OrderCreatedEvent(
7    String eventId,
8    String orderId,
9    String userId,
10    long amount,
11    Instant occurredAt
12) {
13  public static OrderCreatedEvent from(Order order) {
14    return new OrderCreatedEvent(
15        UUID.randomUUID().toString(),
16        order.id(),
17        order.userId(),
18        order.amount(),
19        Instant.now()
20    );
21  }
22}
23
24public final class KafkaTopics {
25  public static final String ORDER_CREATED = "order.created";
26  private KafkaTopics() {}
27}

• 6번 record를 쓰면 JSON으로 보낼 필드를 명확히 고정할 수 있습니다.
• 7번 eventId는 중복 처리 방지에 씁니다. Consumer가 같은 eventId를 이미 처리했는지 확인할 수 있습니다.
• 8번 orderId는 Kafka message key로 쓰기 좋습니다. 같은 주문 이벤트를 같은 partition으로 보내기 위함입니다.
• 14번 도메인 객체 Order에서 이벤트 객체로 변환합니다. DB Entity를 그대로 메시지로 보내지 않습니다.
• 25번 Topic 이름은 상수화해서 Producer와 Consumer가 같은 문자열을 공유하게 합니다.

1@Slf4j
2@Component
3@RequiredArgsConstructor
4public class OrderEventProducer {
5  private final KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;
6
7  public void publishOrderCreated(Order order) {
8    OrderCreatedEvent event = OrderCreatedEvent.from(order);
9    String topic = KafkaTopics.ORDER_CREATED;
10    String key = event.orderId();
11
12    kafkaTemplate.send(topic, key, event)
13        .whenComplete((result, ex) -> {
14          if (ex != null) {
15            log.error("publish failed. topic={}, key={}", topic, key, ex);
16            return;
17          }
18          var meta = result.getRecordMetadata();
19          log.info("published. partition={}, offset={}", meta.partition(), meta.offset());
20        });
21  }
22}

• 5번 key는 String, value는 OrderCreatedEvent인 KafkaTemplate을 주입받습니다.
• 8번 비즈니스 객체를 이벤트 메시지로 변환합니다.
• 10번 key를 orderId로 잡으면 같은 주문 관련 이벤트가 같은 partition에 들어갈 가능성이 높아집니다.
• 12번 send(topic, key, value) 형태로 보냅니다. 성공/실패는 callback에서 확인합니다.
• 18-19번 운영 로그에는 partition과 offset을 남기면 추적이 쉬워집니다.

1package com.example.kafka.order;
2
3import lombok.RequiredArgsConstructor;
4import org.springframework.stereotype.Service;
5import org.springframework.transaction.annotation.Transactional;
6
7@Service
8@RequiredArgsConstructor
9public class OrderService {
10  private final OrderRepository orderRepository;
11  private final OrderEventProducer orderEventProducer;
12
13  @Transactional
14  public OrderResponse createOrder(CreateOrderRequest request) {
15    Order order = Order.create(
16        request.userId(),
17        request.items(),
18        request.amount()
19    );
20
21    Order savedOrder = orderRepository.save(order);
22    orderEventProducer.publishOrderCreated(savedOrder);
23
24    return OrderResponse.from(savedOrder);
25  }
26}

• 13번 주문 저장은 트랜잭션 안에서 처리합니다.
• 21번 DB에 주문이 먼저 남아야 Consumer가 나중에 주문 상세를 조회할 수 있습니다.
• 22번 저장된 주문 기준으로 이벤트를 발행합니다.
• 중요 아주 엄격하게 하려면 21번과 22번 사이 실패를 막기 위해 Outbox 패턴을 씁니다. 입문 자료에서는 연결 지점을 먼저 이해하면 됩니다.

1@Slf4j
2@Component
3@RequiredArgsConstructor
4public class NotificationConsumer {
5  private final ProcessedEventRepository processedEventRepository;
6  private final NotificationService notificationService;
7
8  @KafkaListener(topics = KafkaTopics.ORDER_CREATED, groupId = "notification-service")
9  public void handle(OrderCreatedEvent event, Acknowledgment ack) {
10    log.info("consume. eventId={}, orderId={}", event.eventId(), event.orderId());
11
12    if (processedEventRepository.existsByEventId(event.eventId())) {
13      ack.acknowledge();
14      return;
15    }
16
17    notificationService.sendOrderCreatedMessage(event.userId(), event.orderId(), event.amount());
18    processedEventRepository.save(event.eventId());
19    ack.acknowledge();
20  }
21}

• 8번 이 Consumer는 order.created Topic을 notification-service group으로 읽습니다.
• 9번 Acknowledgment는 수동 offset commit 핸들입니다.
• 12-15번 같은 이벤트가 재전달될 수 있으므로 eventId로 중복 처리를 막습니다.
• 17번 Consumer는 자기 책임인 “알림 발송”만 처리합니다.
• 19번 성공적으로 처리한 뒤 commit합니다. 실패하면 commit하지 않아 재처리될 수 있습니다.

1public void handle(OrderCreatedEvent event, Acknowledgment ack) {
2  log.info("consume. eventId={}, orderId={}",
3      event.eventId(), event.orderId());
4
5  if (processedEventRepository.existsByEventId(event.eventId())) {
6    ack.acknowledge();
7    return;
8  }
9
10  notificationService.sendOrderCreatedMessage(
11      event.userId(),
12      event.orderId(),
13      event.amount()
14  );
15
16  processedEventRepository.save(event.eventId());
17  ack.acknowledge();
18}

• 5번 Consumer는 같은 메시지를 두 번 받을 수 있다고 가정합니다. 그래서 eventId로 이미 처리했는지 먼저 봅니다.
• 6번 이미 처리한 이벤트라면 비즈니스 로직을 다시 실행하지 않고 offset만 commit합니다.
• 10-14번 실제 담당 업무인 알림 발송만 수행합니다. 결제나 분석 로직을 여기 넣지 않습니다.
• 16번 처리 완료 eventId를 저장해 다음 중복 수신을 막습니다.
• 17번 모든 처리가 끝난 뒤 commit합니다. 이 줄 전에 예외가 나면 commit하지 않아 재처리 여지가 남습니다.