이벤트 기반: Kafka 기초

Topic: 메시지 분류 (예: orders, payments)
Partition: Topic 내 분할 (병렬 처리)
Producer: 메시지 발행
Consumer: 메시지 소비
Consumer Group: 같은 Topic을 함께 소비하는 그룹

특징:
- 영속성 (디스크 저장)
- 순서 보장 (Partition 내)
- 수평 확장 (Partition 추가)
- 재처리 가능 (offset reset)

implementation 'org.springframework.kafka:spring-kafka'

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  # 모든 replica 확인
    consumer:
      group-id: order-service
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      auto-offset-reset: earliest
      properties:
        spring.json.trusted.packages: "com.example.event.*"

public record OrderPlacedEvent(
    Long orderId,
    Long userId,
    Long productId,
    int quantity,
    BigDecimal totalAmount,
    LocalDateTime placedAt
) {}


@Service
@RequiredArgsConstructor
public class OrderService {
    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final OrderRepository orderRepo;

    @Transactional
    public Order placeOrder(OrderRequest req) {
        Order order = orderRepo.save(new Order(req));

        OrderPlacedEvent event = new OrderPlacedEvent(
            order.getId(), order.getUserId(),
            order.getProductId(), order.getQuantity(),
            order.getTotal(), LocalDateTime.now()
        );

        kafkaTemplate.send("orders", String.valueOf(order.getUserId()), event);
        // partition key — 같은 사용자의 주문은 같은 partition (순서 보장)
        return order;
    }
}

@Component
@RequiredArgsConstructor
@Slf4j
public class InventoryEventListener {
    private final InventoryService inventoryService;

    @KafkaListener(topics = "orders", groupId = "inventory-service")
    public void handleOrderPlaced(OrderPlacedEvent event) {
        log.info("Order received: {}", event);

        try {
            inventoryService.decrementStock(event.productId(), event.quantity());
        } catch (Exception e) {
            log.error("재고 감소 실패: {}", event, e);
            // 재시도 또는 DLQ로
            throw e;
        }
    }
}


// 여러 Consumer Group — 같은 이벤트 다중 처리
@KafkaListener(topics = "orders", groupId = "notification-service")
public void notifyUser(OrderPlacedEvent event) {
    notificationService.sendOrderConfirmation(event);
}

@KafkaListener(topics = "orders", groupId = "analytics-service")
public void recordOrder(OrderPlacedEvent event) {
    analyticsService.track("order_placed", event);
}

@Configuration
public class KafkaConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(
        ConsumerFactory<String, Object> consumerFactory,
        KafkaTemplate<String, Object> kafkaTemplate
    ) {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);

        // 재시도 + DLQ
        DefaultErrorHandler errorHandler = new DefaultErrorHandler(
            new DeadLetterPublishingRecoverer(kafkaTemplate),
            new FixedBackOff(1000L, 3)  // 1초 간격, 3회 재시도
        );
        factory.setCommonErrorHandler(errorHandler);
        return factory;
    }
}


// DLQ Listener
@KafkaListener(topics = "orders.DLT", groupId = "dlt-handler")
public void handleDeadLetter(OrderPlacedEvent event) {
    log.error("Dead letter: {}", event);
    // Slack 알림 + 수동 처리
}

@Component
@RequiredArgsConstructor
public class OrderEventListener {
    private final RedisTemplate<String, String> redis;
    private final OrderProcessor processor;

    @KafkaListener(topics = "orders")
    public void handle(OrderPlacedEvent event) {
        String key = "processed:" + event.orderId();
        Boolean isNew = redis.opsForValue().setIfAbsent(key, "1", Duration.ofDays(7));

        if (Boolean.FALSE.equals(isNew)) {
            log.info("Already processed: {}", event.orderId());
            return;
        }

        processor.process(event);
    }
}