Kafka 기초: 메시지 큐 vs 이벤트 스트리밍

# 토픽 생성 시 replication factor 설정 예시
# 3개의 Broker에 복제본 유지
kafka-topics.sh --create \
  --topic order-events \
  --partitions 6 \
  --replication-factor 3

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

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                    # 모든 복제본에 쓰기 완료 확인
      retries: 3                   # 실패 시 재시도 횟수
      properties:
        enable.idempotence: true   # 멱등성 보장 (중복 전송 방지)
        max.in.flight.requests.per.connection: 5
    consumer:
      group-id: order-service-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      auto-offset-reset: earliest  # Consumer Group 최초 시작 시 처음부터 읽기
      properties:
        spring.json.trusted.packages: "com.example.kafka.event"
    listener:
      ack-mode: manual             # 수동 커밋으로 메시지 유실 방지

public record OrderEvent(
    String orderId,
    String userId,
    String productName,
    int quantity,
    long totalPrice,
    OrderStatus status,
    LocalDateTime occurredAt
) {
    public enum OrderStatus {
        CREATED, PAID, SHIPPED, DELIVERED, CANCELLED
    }
 
    public static OrderEvent created(String orderId, String userId,
                                     String productName, int quantity, long totalPrice) {
        return new OrderEvent(
            orderId, userId, productName, quantity, totalPrice,
            OrderStatus.CREATED, LocalDateTime.now()
        );
    }
}

@Service
@RequiredArgsConstructor
@Slf4j
public class OrderEventProducer {
 
    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
 
    private static final String TOPIC = "order-events";
 
    /**
     * 주문 이벤트를 Kafka에 발행한다.
     * orderId를 파티션 키로 사용하여 같은 주문의 이벤트가
     * 동일 파티션에 순서대로 저장되도록 보장한다.
     */
    public CompletableFuture<SendResult<String, OrderEvent>> publish(OrderEvent event) {
        log.info("Publishing order event: orderId={}, status={}",
                event.orderId(), event.status());
 
        return kafkaTemplate.send(TOPIC, event.orderId(), event)
                .thenApply(result -> {
                    RecordMetadata metadata = result.getRecordMetadata();
                    log.info("Event sent successfully: topic={}, partition={}, offset={}",
                            metadata.topic(), metadata.partition(), metadata.offset());
                    return result;
                })
                .exceptionally(ex -> {
                    log.error("Failed to send event: orderId={}", event.orderId(), ex);
                    throw new RuntimeException("Kafka send failed", ex);
                });
    }
 
    /**
     * 특정 파티션에 직접 메시지를 보내는 예시
     */
    public void publishToPartition(OrderEvent event, int partition) {
        kafkaTemplate.send(TOPIC, partition, event.orderId(), event);
    }
}

@Service
@Slf4j
public class OrderEventConsumer {
 
    /**
     * order-events 토픽을 구독하여 주문 이벤트를 처리한다.
     * 수동 ACK 모드를 사용하여 처리 완료 후에만 offset을 커밋한다.
     */
    @KafkaListener(
        topics = "order-events",
        groupId = "order-service-group",
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void consume(OrderEvent event,
                        @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                        @Header(KafkaHeaders.OFFSET) long offset,
                        Acknowledgment acknowledgment) {
        try {
            log.info("Received event: orderId={}, status={}, partition={}, offset={}",
                    event.orderId(), event.status(), partition, offset);
 
            // 비즈니스 로직 처리
            processOrderEvent(event);
 
            // 처리 성공 시 offset 커밋
            acknowledgment.acknowledge();
            log.info("Event processed successfully: orderId={}", event.orderId());
 
        } catch (Exception e) {
            log.error("Failed to process event: orderId={}", event.orderId(), e);
            // 처리 실패 시 acknowledge를 호출하지 않으면
            // 해당 메시지는 다시 소비된다.
            // 필요에 따라 DLT(Dead Letter Topic)로 전송할 수도 있다.
        }
    }
 
    private void processOrderEvent(OrderEvent event) {
        switch (event.status()) {
            case CREATED -> handleOrderCreated(event);
            case PAID -> handleOrderPaid(event);
            case SHIPPED -> handleOrderShipped(event);
            case CANCELLED -> handleOrderCancelled(event);
            default -> log.warn("Unknown order status: {}", event.status());
        }
    }
 
    private void handleOrderCreated(OrderEvent event) {
        log.info("Processing new order: {} - {} x{}",
                event.orderId(), event.productName(), event.quantity());
        // 재고 확인, 결제 요청 등
    }
 
    private void handleOrderPaid(OrderEvent event) {
        log.info("Order paid: {}", event.orderId());
        // 배송 준비 시작
    }
 
    private void handleOrderShipped(OrderEvent event) {
        log.info("Order shipped: {}", event.orderId());
        // 배송 추적 시작
    }
 
    private void handleOrderCancelled(OrderEvent event) {
        log.info("Order cancelled: {}", event.orderId());
        // 환불 처리, 재고 복구
    }
}

@Service
@Slf4j
public class NotificationConsumer {
 
    private final NotificationService notificationService;
 
    public NotificationConsumer(NotificationService notificationService) {
        this.notificationService = notificationService;
    }
 
    @KafkaListener(
        topics = "order-events",
        groupId = "notification-group"  // 다른 Consumer Group
    )
    public void consume(OrderEvent event, Acknowledgment acknowledgment) {
        try {
            switch (event.status()) {
                case CREATED -> notificationService.sendOrderConfirmation(event.userId(), event.orderId());
                case SHIPPED -> notificationService.sendShippingNotification(event.userId(), event.orderId());
                case CANCELLED -> notificationService.sendCancellationNotification(event.userId(), event.orderId());
                default -> { /* 알림 불필요 */ }
            }
            acknowledgment.acknowledge();
        } catch (Exception e) {
            log.error("Notification failed for order: {}", event.orderId(), e);
        }
    }
}

@Configuration
@EnableKafka
public class KafkaConsumerConfig {
 
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderEvent>
            kafkaListenerContainerFactory(ConsumerFactory<String, OrderEvent> consumerFactory) {
 
        ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
 
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(3);  // 3개의 Consumer 스레드
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
 
        // 에러 핸들러: 3회 재시도 후 DLT로 전송
        factory.setCommonErrorHandler(new DefaultErrorHandler(
            new DeadLetterPublishingRecoverer(new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(Map.of()))),
            new FixedBackOff(1000L, 3L)
        ));
 
        return factory;
    }
}

@RestController
@RequestMapping("/api/orders")
@RequiredArgsConstructor
public class OrderController {
 
    private final OrderEventProducer orderEventProducer;
 
    @PostMapping
    public ResponseEntity<String> createOrder(@RequestBody CreateOrderRequest request) {
        String orderId = UUID.randomUUID().toString();
 
        OrderEvent event = OrderEvent.created(
            orderId,
            request.userId(),
            request.productName(),
            request.quantity(),
            request.totalPrice()
        );
 
        orderEventProducer.publish(event);
 
        return ResponseEntity.ok(orderId);
    }
 
    public record CreateOrderRequest(
        String userId,
        String productName,
        int quantity,
        long totalPrice
    ) {}
}

# Consumer Group과 파티션의 관계 예시
#
# Topic: order-events (6 partitions)
#
# Consumer Group: order-service-group
#   Consumer-1 → Partition 0, 1
#   Consumer-2 → Partition 2, 3
#   Consumer-3 → Partition 4, 5
#
# Consumer Group: notification-group
#   Consumer-A → Partition 0, 1, 2
#   Consumer-B → Partition 3, 4, 5
#
# 두 그룹은 완전히 독립적으로 동일한 메시지를 소비한다

// Consumer Lag를 프로그래밍으로 확인하는 예시
@Component
@RequiredArgsConstructor
public class KafkaLagMonitor {
 
    private final KafkaAdmin kafkaAdmin;
 
    public Map<TopicPartition, Long> getConsumerLag(String groupId, String topic) {
        try (AdminClient client = AdminClient.create(kafkaAdmin.getConfigurationProperties())) {
            // Consumer Group의 현재 offset 조회
            Map<TopicPartition, OffsetAndMetadata> offsets =
                client.listConsumerGroupOffsets(groupId)
                      .partitionsToOffsetAndMetadata()
                      .get();
 
            // 토픽의 최신 offset 조회
            Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> endOffsets =
                client.listOffsets(
                    offsets.keySet().stream()
                        .collect(Collectors.toMap(tp -> tp, tp -> OffsetSpec.latest()))
                ).all().get();
 
            // Lag 계산
            return offsets.entrySet().stream()
                .filter(e -> e.getKey().topic().equals(topic))
                .collect(Collectors.toMap(
                    Map.Entry::getKey,
                    e -> endOffsets.get(e.getKey()).offset() - e.getValue().offset()
                ));
        } catch (Exception e) {
            throw new RuntimeException("Failed to get consumer lag", e);
        }
    }
}