kafka basic

<!-- pom.xml -->
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
 
    <!-- Test -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

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 # Đợi tất cả ISR confirm → không mất data
      retries: 3 # Retry khi gặp transient errors
      properties:
        enable.idempotence: true # Tránh duplicate messages khi retry
        max.in.flight.requests.per.connection: 5
 
    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 # Đọc từ đầu nếu chưa có committed offset
      properties:
        spring.json.trusted.packages: '*'
 
    # KafkaAdmin: tạo các topic khai báo bằng bean NewTopic (xem phần topic configuration)
    admin:
      properties:
        bootstrap.servers: localhost:9092

// OrderEvent.java
public class OrderEvent {
 
    private String orderId;
    private String userId;
    private String product;
    private BigDecimal amount;
    private OrderStatus status;
    private LocalDateTime timestamp;
 
    public enum OrderStatus {
        CREATED, CONFIRMED, SHIPPED, DELIVERED, CANCELLED
    }
 
    // Constructor mặc định cần cho deserialization
    public OrderEvent() {}
 
    public OrderEvent(String orderId, String userId, String product,
                      BigDecimal amount, OrderStatus status) {
        this.orderId = orderId;
        this.userId = userId;
        this.product = product;
        this.amount = amount;
        this.status = status;
        this.timestamp = LocalDateTime.now();
    }
 
    // Getters & Setters (hoặc dùng Lombok @Data)
    public String getOrderId() { return orderId; }
    public void setOrderId(String orderId) { this.orderId = orderId; }
    public String getUserId() { return userId; }
    public void setUserId(String userId) { this.userId = userId; }
    public String getProduct() { return product; }
    public void setProduct(String product) { this.product = product; }
    public BigDecimal getAmount() { return amount; }
    public void setAmount(BigDecimal amount) { this.amount = amount; }
    public OrderStatus getStatus() { return status; }
    public void setStatus(OrderStatus status) { this.status = status; }
    public LocalDateTime getTimestamp() { return timestamp; }
    public void setTimestamp(LocalDateTime timestamp) { this.timestamp = timestamp; }
 
    @Override
    public String toString() {
        return "OrderEvent{orderId='%s', userId='%s', status=%s, amount=%s}"
                .formatted(orderId, userId, status, amount);
    }
}

// KafkaTopicConfig.java
@Configuration
public class KafkaTopicConfig {
 
    @Bean
    public NewTopic orderEventsTopic() {
        return TopicBuilder.name("order-events")
                .partitions(6)           // 6 partitions cho parallel processing
                .replicas(3)             // 3 replicas cho fault tolerance
                .config(TopicConfig.RETENTION_MS_CONFIG, "604800000") // 7 ngày
                .config(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2")
                .build();
    }
 
    @Bean
    public NewTopic orderEventsDLT() {
        // Dead Letter Topic — nơi chứa messages xử lý thất bại
        // Số partition phải >= topic gốc: DeadLetterPublishingRecoverer
        // mặc định gửi vào cùng số partition với record gốc
        return TopicBuilder.name("order-events.DLT")
                .partitions(6)
                .replicas(3)
                .build();
    }
}

// OrderEventProducer.java
@Service
@Slf4j
public class OrderEventProducer {
 
    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
 
    // Constructor injection — best practice
    public OrderEventProducer(KafkaTemplate<String, OrderEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
 
    /**
     * Gửi order event lên Kafka.
     * Key = userId → tất cả events của cùng user vào cùng partition
     * → đảm bảo ordering per user.
     */
    public CompletableFuture<SendResult<String, OrderEvent>> publishOrderEvent(
            OrderEvent event) {
 
        log.info("Publishing event: {}", event);
 
        return kafkaTemplate
                .send("order-events", event.getUserId(), event)
                .whenComplete((result, ex) -> {
                    if (ex == null) {
                        log.info("Event sent successfully: topic={}, partition={}, offset={}",
                                result.getRecordMetadata().topic(),
                                result.getRecordMetadata().partition(),
                                result.getRecordMetadata().offset());
                    } else {
                        log.error("Failed to send event: {}", event, ex);
                        // Ở đây bạn có thể: retry, gửi vào DLQ, alert, ...
                    }
                });
    }
 
    /**
     * Gửi event với custom headers — hữu ích cho tracing.
     */
    public void publishWithHeaders(OrderEvent event, String correlationId) {
        ProducerRecord<String, OrderEvent> record =
                new ProducerRecord<>("order-events", event.getUserId(), event);
 
        record.headers()
                .add("correlationId", correlationId.getBytes(StandardCharsets.UTF_8))
                .add("source", "order-service".getBytes(StandardCharsets.UTF_8));
 
        kafkaTemplate.send(record);
    }
}

// OrderEventConsumer.java
@Service
@Slf4j
public class OrderEventConsumer {
 
    private final OrderService orderService;
 
    public OrderEventConsumer(OrderService orderService) {
        this.orderService = orderService;
    }
 
    /**
     * Consumer cơ bản — đủ cho hầu hết use cases.
     *
     * concurrency = 3 → Spring tạo 3 consumer threads,
     * mỗi thread xử lý 1 hoặc nhiều partitions.
     */
    @KafkaListener(
            topics = "order-events",
            groupId = "order-service",
            concurrency = "3"
    )
    public void handleOrderEvent(
            @Payload OrderEvent event,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long timestamp) {
 
        log.info("Received event: {} [partition={}, offset={}]",
                event, partition, offset);
 
        try {
            orderService.processOrder(event);
        } catch (Exception e) {
            log.error("Error processing event at offset {}: {}",
                    offset, e.getMessage());
            // Throw lại để Kafka retry hoặc gửi vào DLT
            throw e;
        }
    }
 
    /**
     * Consumer với manual acknowledgment — khi bạn cần kiểm soát
     * chính xác lúc nào offset được commit.
     */
    @KafkaListener(
            topics = "order-events",
            groupId = "order-service-manual-ack",
            containerFactory = "manualAckListenerFactory"
    )
    public void handleWithManualAck(
            OrderEvent event,
            Acknowledgment acknowledgment) {
 
        try {
            orderService.processOrder(event);
            // Chỉ commit offset khi xử lý thành công
            acknowledgment.acknowledge();
        } catch (Exception e) {
            // Chỉ "không acknowledge" thì KHÔNG đủ: consumer vẫn poll tiếp,
            // lần ack sau commit offset lớn hơn và message này bị bỏ qua.
            // nack() seek lại về message này, đợi 1s rồi giao lại.
            log.error("Processing failed, will retry: {}", e.getMessage());
            acknowledgment.nack(Duration.ofSeconds(1));
        }
    }
 
    /**
     * Batch consumer — xử lý nhiều messages cùng lúc.
     * Hiệu quả hơn khi cần bulk insert vào DB.
     */
    @KafkaListener(
            topics = "order-events",
            groupId = "order-service-batch",
            containerFactory = "batchListenerFactory"
    )
    public void handleBatch(List<OrderEvent> events) {
        log.info("Received batch of {} events", events.size());
        orderService.processBatch(events);
    }
}

// KafkaConsumerConfig.java
@Configuration
@EnableKafka
public class KafkaConsumerConfig {
 
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
 
    @Bean
    public ConsumerFactory<String, OrderEvent> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                JsonDeserializer.class);
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        return new DefaultKafkaConsumerFactory<>(props);
    }
 
    /**
     * Factory cho manual acknowledgment.
     * AckMode.MANUAL_IMMEDIATE = commit ngay khi acknowledge() được gọi.
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderEvent>
            manualAckListenerFactory() {
 
        ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties()
                .setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }
 
    /**
     * Factory cho batch processing.
     * max.poll.records = số messages tối đa mỗi lần poll.
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderEvent>
            batchListenerFactory() {
 
        ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setBatchListener(true);
        factory.getContainerProperties()
                .setAckMode(ContainerProperties.AckMode.BATCH);
        return factory;
    }
}

// KafkaErrorHandlingConfig.java
@Configuration
public class KafkaErrorHandlingConfig {
 
    /**
     * Retry 3 lần với backoff tăng dần, sau đó gửi vào Dead Letter Topic.
     *
     * Flow: message → retry 1 (1s) → retry 2 (2s) → retry 3 (4s) → DLT
     */
    @Bean
    public DefaultErrorHandler errorHandler(
            KafkaOperations<String, OrderEvent> kafkaOperations) {
 
        // Gửi message thất bại vào DLT
        DeadLetterPublishingRecoverer recoverer =
                new DeadLetterPublishingRecoverer(kafkaOperations,
                        (record, ex) -> new TopicPartition(
                                record.topic() + ".DLT",
                                record.partition()));
 
        // Retry đúng 3 lần với exponential backoff 1s, 2s, 4s
        ExponentialBackOffWithMaxRetries backOff = new ExponentialBackOffWithMaxRetries(3);
        backOff.setInitialInterval(1000L);
        backOff.setMultiplier(2.0);
 
        DefaultErrorHandler errorHandler =
                new DefaultErrorHandler(recoverer, backOff);
 
        // Không retry cho các lỗi không thể recover
        errorHandler.addNotRetryableExceptions(
                DeserializationException.class,
                ClassCastException.class,
                NullPointerException.class
        );
 
        return errorHandler;
    }
}

// OrderController.java
@RestController
@RequestMapping("/api/orders")
public class OrderController {
 
    private final OrderEventProducer producer;
 
    public OrderController(OrderEventProducer producer) {
        this.producer = producer;
    }
 
    @PostMapping
    public ResponseEntity<Map<String, String>> createOrder(
            @RequestBody CreateOrderRequest request) {
 
        OrderEvent event = new OrderEvent(
                UUID.randomUUID().toString(),
                request.getUserId(),
                request.getProduct(),
                request.getAmount(),
                OrderEvent.OrderStatus.CREATED
        );
 
        producer.publishOrderEvent(event);
 
        return ResponseEntity.accepted()
                .body(Map.of(
                        "orderId", event.getOrderId(),
                        "status", "ACCEPTED",
                        "message", "Order event published to Kafka"
                ));
    }
}

// OutboxEvent entity — lưu event vào cùng DB transaction với business data
@Entity
@Table(name = "outbox_events")
public class OutboxEvent {
 
    @Id
    private String id;
    private String aggregateType;  // "Order"
    private String aggregateId;    // orderId
    private String eventType;      // "OrderCreated"
 
    @Column(columnDefinition = "TEXT")
    private String payload;        // JSON serialized event
 
    private LocalDateTime createdAt;
    private boolean published;
 
    // constructors, getters, setters...
}
 
// Service — save business data + outbox event trong cùng 1 transaction
@Service
@Transactional
@RequiredArgsConstructor
public class OrderService {
 
    private final OrderRepository orderRepository;
    private final OutboxRepository outboxRepository;
    private final ObjectMapper objectMapper;
 
    public Order createOrder(CreateOrderRequest request) {
        // 1. Save business data
        Order order = new Order(request);
        orderRepository.save(order);
 
        // 2. Save outbox event (cùng transaction)
        OutboxEvent outboxEvent = new OutboxEvent();
        outboxEvent.setId(UUID.randomUUID().toString());
        outboxEvent.setAggregateType("Order");
        outboxEvent.setAggregateId(order.getId());
        outboxEvent.setEventType("OrderCreated");
        try {
            outboxEvent.setPayload(objectMapper.writeValueAsString(order));
        } catch (JsonProcessingException e) {
            // Checked exception: @Transactional mặc định KHÔNG rollback
            // với checked exception, nên bọc lại thành unchecked
            throw new IllegalStateException("Cannot serialize order", e);
        }
        outboxEvent.setCreatedAt(LocalDateTime.now());
        outboxEvent.setPublished(false);
        outboxRepository.save(outboxEvent);
 
        return order;
        // Cả hai save cùng commit hoặc cùng rollback
    }
}
 
// Scheduler — poll outbox table và publish lên Kafka
@Component
@Slf4j
@RequiredArgsConstructor
public class OutboxPublisher {
 
    private final OutboxRepository outboxRepository;
    // KafkaTemplate riêng dùng StringSerializer: payload đã là JSON sẵn,
    // đi qua JsonSerializer nữa sẽ thành chuỗi JSON bị bọc trong dấu nháy
    private final KafkaTemplate<String, String> kafkaTemplate;
 
    @Scheduled(fixedDelay = 1000) // Poll mỗi giây
    @Transactional
    public void publishPendingEvents() {
        List<OutboxEvent> events =
                outboxRepository.findByPublishedFalseOrderByCreatedAtAsc();
 
        for (OutboxEvent event : events) {
            try {
                kafkaTemplate.send(
                        event.getAggregateType().toLowerCase() + "-events",
                        event.getAggregateId(),
                        event.getPayload()
                ).get(); // Blocking wait để đảm bảo gửi thành công
 
                event.setPublished(true);
                outboxRepository.save(event);
            } catch (Exception e) {
                log.error("Failed to publish outbox event: {}", event.getId(), e);
                break; // Dừng lại, retry ở lần poll tiếp theo
            }
        }
    }
}

// Các events cho Order aggregate
public sealed interface OrderDomainEvent {
    String orderId();
 
    record OrderCreated(String orderId, String userId,
                        String product, BigDecimal amount) implements OrderDomainEvent {}
 
    record OrderConfirmed(String orderId, LocalDateTime confirmedAt)
            implements OrderDomainEvent {}
 
    record OrderShipped(String orderId, String trackingNumber)
            implements OrderDomainEvent {}
 
    record OrderCancelled(String orderId, String reason)
            implements OrderDomainEvent {}
}
 
// Rebuild state từ events
public class OrderAggregate {
 
    private String orderId;
    private String userId;
    private OrderStatus status;
    private BigDecimal amount;
    private String trackingNumber;
 
    public static OrderAggregate replayEvents(List<OrderDomainEvent> events) {
        OrderAggregate aggregate = new OrderAggregate();
        events.forEach(aggregate::apply);
        return aggregate;
    }
 
    private void apply(OrderDomainEvent event) {
        switch (event) {
            case OrderDomainEvent.OrderCreated e -> {
                this.orderId = e.orderId();
                this.userId = e.userId();
                this.amount = e.amount();
                this.status = OrderStatus.CREATED;
            }
            case OrderDomainEvent.OrderConfirmed e -> {
                this.status = OrderStatus.CONFIRMED;
            }
            case OrderDomainEvent.OrderShipped e -> {
                this.status = OrderStatus.SHIPPED;
                this.trackingNumber = e.trackingNumber();
            }
            case OrderDomainEvent.OrderCancelled e -> {
                this.status = OrderStatus.CANCELLED;
            }
        }
    }
}

// Orchestrator-based Saga
@Service
@Slf4j
@RequiredArgsConstructor
public class OrderSagaOrchestrator {
 
    private final KafkaTemplate<String, SagaEvent> kafkaTemplate;
 
    @KafkaListener(topics = "saga-order-reply", groupId = "saga-orchestrator")
    public void handleSagaReply(SagaEvent event) {
        switch (event.getStep()) {
            case "PAYMENT_SUCCESS" -> {
                log.info("Payment successful, reserving inventory...");
                kafkaTemplate.send("saga-inventory-request",
                        event.getOrderId(),
                        SagaEvent.of(event.getOrderId(), "RESERVE_INVENTORY"));
            }
            case "INVENTORY_RESERVED" -> {
                log.info("Inventory reserved, initiating shipping...");
                kafkaTemplate.send("saga-shipping-request",
                        event.getOrderId(),
                        SagaEvent.of(event.getOrderId(), "CREATE_SHIPMENT"));
            }
            case "PAYMENT_FAILED" -> {
                log.warn("Payment failed, cancelling order...");
                kafkaTemplate.send("saga-order-request",
                        event.getOrderId(),
                        SagaEvent.of(event.getOrderId(), "CANCEL_ORDER"));
            }
            case "INVENTORY_FAILED" -> {
                log.warn("Inventory failed, refunding payment...");
                kafkaTemplate.send("saga-payment-request",
                        event.getOrderId(),
                        SagaEvent.of(event.getOrderId(), "REFUND_PAYMENT"));
            }
        }
    }
}

// Real-time order analytics với Kafka Streams
@Configuration
@EnableKafkaStreams
public class OrderAnalyticsStream {
 
    @Bean
    public KStream<String, OrderEvent> orderAnalytics(
            StreamsBuilder streamsBuilder) {
 
        // Đọc từ topic
        KStream<String, OrderEvent> orderStream = streamsBuilder
                .stream("order-events",
                        Consumed.with(Serdes.String(), orderEventSerde()));
 
        // 1. Filter: chỉ lấy orders đã confirmed
        KStream<String, OrderEvent> confirmedOrders = orderStream
                .filter((key, event) ->
                        event.getStatus() == OrderEvent.OrderStatus.CONFIRMED);
 
        // 2. Tính tổng revenue theo user (windowed aggregation)
        KTable<Windowed<String>, BigDecimal> revenuePerUser = confirmedOrders
                .groupByKey()
                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
                .aggregate(
                        () -> BigDecimal.ZERO,
                        (userId, event, total) -> total.add(event.getAmount()),
                        Materialized.with(Serdes.String(), bigDecimalSerde())
                );
 
        // 3. Detect high-value orders → gửi alert
        orderStream
                .filter((key, event) ->
                        event.getAmount().compareTo(new BigDecimal("10000")) > 0)
                .to("high-value-order-alerts",
                        Produced.with(Serdes.String(), orderEventSerde()));
 
        // 4. Branch: phân loại orders theo status
        Map<String, KStream<String, OrderEvent>> branches = orderStream
                .split(Named.as("order-"))
                .branch((key, event) ->
                        event.getStatus() == OrderEvent.OrderStatus.CANCELLED,
                        Branched.as("cancelled"))
                .branch((key, event) ->
                        event.getStatus() == OrderEvent.OrderStatus.SHIPPED,
                        Branched.as("shipped"))
                .defaultBranch(Branched.as("other"));
 
        branches.get("order-cancelled")
                .to("cancelled-orders");
 
        return orderStream;
    }
}

# Prometheus metrics quan trọng nhất
# (tên metric cụ thể tuỳ exporter: kafka_exporter, JMX exporter, Micrometer...)
 
# Consumer lag — số messages chưa được xử lý (tên theo kafka_exporter)
# Nếu lag tăng liên tục → consumer không kịp xử lý
kafka_consumergroup_lag{consumergroup="order-service", topic="order-events"}
 
# Producer request latency
kafka_producer_request_latency_avg
 
# Khoảng cách giữa 2 lần poll: nếu max tiến sát max.poll.interval.ms
# là sắp bị kick khỏi group
kafka_consumer_time_between_poll_max
 
# Under-replicated partitions — DẤU HIỆU NGUY HIỂM (tên theo JMX exporter)
kafka_server_replicamanager_underreplicatedpartitions

// Spring Boot Actuator + Micrometer: Boot tự gắn metrics cho producer/consumer
// factory do Boot tạo. Factory tự khai báo (như consumerFactory ở trên)
// thì phải gắn listener bằng tay, khai báo listener thành bean riêng là không đủ.
@Bean
public ConsumerFactory<String, OrderEvent> consumerFactory(MeterRegistry registry) {
    DefaultKafkaConsumerFactory<String, OrderEvent> factory =
            new DefaultKafkaConsumerFactory<>(consumerProps());
    factory.addListener(new MicrometerConsumerListener<>(registry));
    return factory;
}

@Service
@Slf4j
@RequiredArgsConstructor
public class IdempotentOrderConsumer {
 
    private final ProcessedEventRepository processedEventRepo;
    private final OrderService orderService;
 
    @KafkaListener(topics = "order-events", groupId = "order-service")
    @Transactional
    public void handle(OrderEvent event,
                       @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                       @Header(KafkaHeaders.OFFSET) long offset) {
 
        // Tạo unique ID từ partition + offset (hoặc dùng event ID)
        String eventId = "order-events-%d-%d".formatted(partition, offset);
 
        // Check đã xử lý chưa
        if (processedEventRepo.existsById(eventId)) {
            log.warn("Duplicate event detected, skipping: {}", eventId);
            return;
        }
 
        // Xử lý business logic
        orderService.processOrder(event);
 
        // Đánh dấu đã xử lý (cùng transaction với business logic)
        processedEventRepo.save(new ProcessedEvent(eventId, LocalDateTime.now()));
    }
}

# Producer tuning
spring:
  kafka:
    producer:
      # Batch messages trước khi gửi → tăng throughput
      batch-size: 32768 # 32KB
      # Đợi tối đa 5ms để gom batch
      properties:
        linger.ms: 5 # Kafka 4.0 trở đi mặc định đã là 5ms (trước đó là 0)
        # Compression — giảm network bandwidth
        compression.type: lz4 # lz4 = fast, snappy = balanced, zstd = best ratio
        # Buffer memory cho producer
        buffer.memory: 67108864 # 64MB
 
    consumer:
      # Số messages tối đa mỗi lần poll
      properties:
        max.poll.records: 500
        # Thời gian tối đa giữa 2 lần poll
        # Nếu vượt quá → consumer bị coi là dead → rebalance
        max.poll.interval.ms: 300000 # 5 phút
        # Fetch size
        fetch.min.bytes: 1024
        fetch.max.wait.ms: 500

# docker-compose.yml
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
 
  kafka:
    image: confluentinc/cp-kafka:7.5.0
    depends_on:
      - zookeeper
    ports:
      - '9092:9092'
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      # 2 listener: app trên máy host gọi localhost:9092,
      # container khác (kafka-ui) gọi kafka:29092
      KAFKA_LISTENERS: INTERNAL://0.0.0.0:29092,EXTERNAL://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:29092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
 
  # UI để visualize Kafka — rất hữu ích khi debug
  # (provectuslabs/kafka-ui đã ngừng phát triển, kafbat/kafka-ui là bản fork tiếp tục)
  kafka-ui:
    image: ghcr.io/kafbat/kafka-ui:latest
    ports:
      - '8080:8080'
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092

// Sai: xử lý mất 10 phút nhưng max.poll.interval chỉ 5 phút
@KafkaListener(topics = "heavy-processing")
public void handle(Event event) {
    heavyProcessing(event); // Mất 10 phút
    // → Consumer bị kick khỏi group → rebalance → message bị re-deliver
}
 
// Đúng: max.poll.interval phải lớn hơn (max.poll.records × thời gian xử lý 1 record)
// Ở đây mỗi record mất 10 phút → poll 1 record mỗi lần, interval 15 phút
@KafkaListener(topics = "heavy-processing",
        properties = {
            "max.poll.interval.ms=900000",  // 15 phút
            "max.poll.records=1"             // 1 record mỗi lần poll
        })
public void handle(Event event) {
    heavyProcessing(event);
}

// Sai: retries có thể làm đảo thứ tự messages
// Message A gửi → fail → retry
// Message B gửi → success
// Message A retry → success
// Kết quả: B trước A → sai thứ tự!
 
// Đúng: bật idempotence (mặc định bật từ Kafka 3.0), giữ đúng thứ tự
// với max.in.flight.requests.per.connection tối đa 5
spring.kafka.producer.properties.enable.idempotence=true
// Hoặc set max.in.flight.requests.per.connection=1 (nhưng giảm throughput)

# ✅ Giải pháp: ErrorHandlingDeserializer
spring:
  kafka:
    consumer:
      key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer

// Không có groupId: Spring lấy spring.kafka.consumer.group-id chung.
// Nếu cũng không có thì consumer lỗi (InvalidGroupIdException) khi subscribe.
// Nhiều listener khác mục đích dùng chung một group cũng là bug khó thấy.
@KafkaListener(topics = "events")
 
// Luôn set groupId cố định, mỗi mục đích xử lý một group
@KafkaListener(topics = "events", groupId = "my-service")
 
// Đừng sinh groupId ngẫu nhiên mỗi lần chạy: mỗi lần restart là một group mới,
// không có offset đã commit → đọc lại từ đầu (earliest) hoặc bỏ lỡ message (latest)

// Version 1: OrderEvent ban đầu
{
    "orderId": "abc-123",
    "amount": 99.99
}
 
// Version 2: Thêm field mới (consumer cũ vẫn đọc được ✅)
{
    "orderId": "abc-123",
    "amount": 99.99,
    "currency": "USD"    // Mới — consumer cũ bỏ qua field này (nếu deserializer không fail on unknown)
}
 
// Version 3: Đổi tên field (BREAKING CHANGE ❌)
{
    "orderId": "abc-123",
    "totalAmount": 99.99  // Consumer cũ tìm "amount" → null
}

// Avro schema evolution với Spring
// (KafkaAvroSerializer nằm trong io.confluent:kafka-avro-serializer, repo Maven của Confluent)
@Configuration
public class AvroConfig {
 
    @Bean
    public ProducerFactory<String, GenericRecord> avroProducerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                KafkaAvroSerializer.class);
        props.put("schema.registry.url", "http://localhost:8081");
        return new DefaultKafkaProducerFactory<>(props);
    }
}