Kafka KRaft 실습 Spring Boot 마이크로서비스

Spring Boot 마이크로서비스에서의 실전 Kafka 구현

중급에서 고급 개발자들은 KRaft를 사용해 Kafka 클러스터를 빠르게 배포하고 온라인 스토어를 위한 네 개의 Spring Boot 서비스와 통합할 수 있습니다. order-service는 REST API를 통해 주문을 받고 order-placed 이벤트를 게시합니다. inventory-service는 재고를 예약하고 inventory-reserved 또는 inventory-rejected 이벤트를 생성합니다. analytics-service와 notification-service는 이러한 이벤트를 비동기적으로 처리합니다.

요구사항: Java, Spring Boot, Docker, Docker Compose, Postman. 의존성: Spring for Apache Kafka, Spring Web.

Kafka 클러스터를 위한 Docker Compose

KRaft를 사용해 Zookeeper 없이 Kafka 8.1.0 브로커 3개를 배포합니다. 각 브로커는 브로커와 컨트롤러 역할을 모두 수행합니다. 토픽, 파티션, 컨슈머 그룹을 모니터링하기 위해 포트 8086에서 Kafka-UI를 실행합니다.

docker-compose.yaml 파일을 생성하세요:

services:
  kafka-0:
    image: confluentinc/cp-kafka:8.1.0
    container_name: kafka-0
    hostname: kafka-0
    ports:
      - "29092:29092"
    environment:
      CLUSTER_ID: ZETVJENWRjaECDUEExP_eg
      KAFKA_NODE_ID: 0
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: INTERNAL://:9091,EXTERNAL://:29092,CONTROLLER://:9093
      KAFKA_CONTROLLER_QUORUM_VOTERS: 0@kafka-0:9093,1@kafka-1:9093,2@kafka-2:9093
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka-0:9091,EXTERNAL://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
  # kafka-1 (포트 29093), kafka-2 (포트 29094)에 대해서도 동일하게
  kafka-ui:
    image: provectuslabs/kafka-ui:v0.7.2
    container_name: kafka-ui
    ports:
      - "8086:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka-0:9091,kafka-1:9091,kafka-2:9091
      DYNAMIC_CONFIG_ENABLED: true

시작 명령: docker compose up -d. 클러스터는 3개의 파티션과 x3 복제로 로드할 준비가 됩니다.

Spring Boot 서비스 구성

공통 프로듀서 설정

모든 서비스의 application.properties에서:

spring.kafka.bootstrap-servers=localhost:29092,localhost:29093,localhost:29094
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

프로그램 방식 토픽 생성

config 패키지의 KafkaConfig 클래스:

@Configuration
public class KafkaConfig {
    @Bean
    public NewTopic orderPlacedTopic() {
        return TopicBuilder.name("order-placed")
            .partitions(3)
            .replicas(3)
            .config(TopicConfig.RETENTION_MS_CONFIG, "86400000")
            .config(TopicConfig.RETENTION_BYTES_CONFIG, "524288000")
            .build();
    }
}

보존 기간: 토픽당 24시간, 500MB. inventory-service의 inventory-reserved, inventory-rejected도 동일합니다.

Order-Service 구현 (프로듀서)

DTO와 이벤트

public record OrderRequest(String email, String productName, Integer quantity) {}

public record OrderPlacedEvent(String orderId, String email, String productName, Integer quantity) {}

비즈니스 로직

@Service
public class OrderService {
    private final KafkaTemplate<String, OrderPlacedEvent> kafkaTemplate;

    public OrderService(KafkaTemplate<String, OrderPlacedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public String placeOrder(OrderRequest orderRequest) {
        String orderId = UUID.randomUUID().toString();
        OrderPlacedEvent event = new OrderPlacedEvent(orderId, orderRequest.email(), 
            orderRequest.productName(), orderRequest.quantity());
        kafkaTemplate.send("order-placed", event);
        return orderId;
    }
}

REST 컨트롤러는 orderId와 함께 즉시 201 Created를 반환하며, 처리는 비동기적으로 발생합니다.

Inventory-Service: 프로듀서 + 컨슈머

order-placed를 수신하고 재고 수준을 확인한 후 inventory-reserved 또는 inventory-rejected를 게시합니다. @KafkaListener와 두 번째 KafkaTemplate가 필요합니다.

서비스 역할:

• 순수 프로듀서: order-service
• 순수 컨슈머: analytics-service, notification-service
• 프로듀서+컨슈머: inventory-service

컨슈머 구현 (analytics, notification)

@KafkaListener(topics = "inventory-reserved")
public void handleInventoryReserved(InventoryReservedEvent event) {
    // Analytics logic
}

Spring Kafka는 오프셋을 자동으로 관리하며, 컨슈머 그룹은 서비스 이름으로 형성됩니다.

플로우 테스트

• JSON { "email": "[email protected]", "productName": "laptop", "quantity": 1 }로 /orders에 POST
• Kafka-UI에서 모니터링: order-placed → inventory-reserved
• analytics와 notification이 이벤트를 병렬로 수신

핵심 포인트:

• KRaft는 Zookeeper 없이 배포를 단순화합니다
• 3 파티션 + x3 복제 = 장애 내성
• 값에 JsonSerializer, 키에 StringSerializer
• 개발에 적합한 24시간/500MB 보존 기간
• KafkaTemplate이 저수준 API를 추상화합니다

— Editorial Team