Spring Kafka Integration

@Service
@RequiredArgsConstructor
public class OrderEventPublisher {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    // Simple send — fire and forget
    public void publishOrderCreated(OrderEvent event) {
        kafkaTemplate.send("order-events", event.getOrderId(), event);
    }

    // Send to default topic (configured via spring.kafka.template.default-topic)
    public void publishDefault(OrderEvent event) {
        kafkaTemplate.sendDefault(event.getOrderId(), event);
    }

    // Full control with ProducerRecord
    public void publishWithHeaders(OrderEvent event) {
        ProducerRecord<String, OrderEvent> record = new ProducerRecord<>(
            "order-events",       // topic
            null,                 // partition (null = let partitioner decide)
            event.getOrderId(),   // key
            event,                // value
            List.of(new RecordHeader("event-type", "ORDER_CREATED".getBytes()))
        );
        kafkaTemplate.send(record);
    }

    // With callback — non-blocking confirmation
    public void publishWithCallback(OrderEvent event) {
        CompletableFuture<SendResult<String, OrderEvent>> future =
            kafkaTemplate.send("order-events", event.getOrderId(), event);

        future.whenComplete((result, ex) -> {
            if (ex != null) {
                log.error("Failed to send event: {}", event.getOrderId(), ex);
                // trigger retry or dead-letter
            } else {
                RecordMetadata metadata = result.getRecordMetadata();
                log.info("Sent to partition={} offset={}",
                    metadata.partition(), metadata.offset());
            }
        });
    }

    // Blocking send — wait for broker acknowledgment
    public void publishBlocking(OrderEvent event) throws Exception {
        SendResult<String, OrderEvent> result =
            kafkaTemplate.send("order-events", event.getOrderId(), event)
                .get(10, TimeUnit.SECONDS);  // blocks until ack
        log.info("Confirmed offset: {}", result.getRecordMetadata().offset());
    }
}

@Component
@Slf4j
public class OrderEventConsumer {

    @KafkaListener(
        topics = "order-events",
        groupId = "order-processing-group",
        concurrency = "3"  // 3 consumer threads
    )
    public void handleOrderEvent(OrderEvent event) {
        log.info("Processing order: {}", event.getOrderId());
        orderService.process(event);
    }
}

@Component
public class AdvancedConsumer {

    // Multiple topics
    @KafkaListener(topics = {"order-events", "payment-events"})
    public void multiTopic(ConsumerRecord<String, ?> record) {
        log.info("Topic={} Partition={} Offset={} Key={}",
            record.topic(), record.partition(), record.offset(), record.key());
    }

    // Custom container factory (for different deserialization or concurrency)
    @KafkaListener(
        topics = "payment-events",
        containerFactory = "paymentListenerFactory",
        groupId = "payment-group"
    )
    public void handlePayment(PaymentEvent event) {
        paymentService.process(event);
    }

    // Topic pattern matching
    @KafkaListener(topicPattern = "order-.*")
    public void handleAllOrderTopics(ConsumerRecord<String, String> record) {
        // matches order-created, order-updated, order-cancelled, etc.
    }

    // With message headers
    @KafkaListener(topics = "order-events", groupId = "audit-group")
    public void withHeaders(
            @Payload OrderEvent event,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long timestamp) {
        log.info("Event from partition={} offset={} ts={}", partition, offset, timestamp);
    }
}

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, PaymentEvent>
            paymentListenerFactory(ConsumerFactory<String, PaymentEvent> cf) {

        ConcurrentKafkaListenerContainerFactory<String, PaymentEvent> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(cf);
        factory.setConcurrency(5);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        factory.setCommonErrorHandler(new DefaultErrorHandler(
            new DeadLetterPublishingRecoverer(kafkaTemplate),
            new FixedBackOff(1000L, 3)  // 1s interval, 3 retries
        ));
        return factory;
    }
}

public class OrderEventSerde implements Serializer<OrderEvent>, Deserializer<OrderEvent> {

    private final ObjectMapper mapper = new ObjectMapper();

    @Override
    public byte[] serialize(String topic, OrderEvent data) {
        try {
            return mapper.writeValueAsBytes(data);
        } catch (JsonProcessingException e) {
            throw new SerializationException("Error serializing OrderEvent", e);
        }
    }

    @Override
    public OrderEvent deserialize(String topic, byte[] data) {
        try {
            return mapper.readValue(data, OrderEvent.class);
        } catch (IOException e) {
            throw new SerializationException("Error deserializing OrderEvent", e);
        }
    }
}

@Configuration
public class KafkaErrorConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<String, Object> template) {
        // Publish failed messages to DLT after retries exhausted
        DeadLetterPublishingRecoverer recoverer =
            new DeadLetterPublishingRecoverer(template,
                (record, ex) -> new TopicPartition(
                    record.topic() + ".DLT", record.partition()));

        // Exponential backoff: 1s, 2s, 4s — then DLT
        ExponentialBackOff backOff = new ExponentialBackOff(1000L, 2.0);
        backOff.setMaxElapsedTime(10000L);  // max 10s total

        DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, backOff);

        // Don't retry on these — send straight to DLT
        handler.addNotRetryableExceptions(
            DeserializationException.class,
            ValidationException.class
        );

        return handler;
    }
}

@KafkaListener(topics = "order-events.DLT", groupId = "dlt-processor")
public void processDlt(
        ConsumerRecord<String, OrderEvent> record,
        @Header(KafkaHeaders.DLT_EXCEPTION_MESSAGE) String errorMsg,
        @Header(KafkaHeaders.DLT_ORIGINAL_TOPIC) String originalTopic) {

    log.error("DLT received: topic={} key={} error={}",
        originalTopic, record.key(), errorMsg);

    // Alert ops team, store for manual review, or attempt reprocessing
    alertService.notifyDltMessage(record, errorMsg);
}

@Configuration
public class KafkaTransactionConfig {

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-tx-");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

        DefaultKafkaProducerFactory<String, Object> factory =
            new DefaultKafkaProducerFactory<>(props);
        factory.setTransactionIdPrefix("order-tx-");
        return factory;
    }
}

@Service
@RequiredArgsConstructor
public class TransactionalOrderProcessor {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    // Everything inside executeInTransaction is atomic
    public void processAndForward(OrderEvent event) {
        kafkaTemplate.executeInTransaction(ops -> {
            // Consume → process → produce in one transaction
            OrderConfirmation confirmation = orderService.confirm(event);
            ops.send("order-confirmations", event.getOrderId(), confirmation);
            ops.send("inventory-updates", event.getOrderId(),
                new InventoryDeduction(event.getItems()));
            return null;
        });
    }
}

@Configuration
public class BatchConsumerConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderEvent>
            batchFactory(ConsumerFactory<String, OrderEvent> cf) {

        ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(cf);
        factory.setBatchListener(true);  // enable batch mode
        factory.setConcurrency(3);
        return factory;
    }
}

@Component
public class BatchOrderConsumer {

    @KafkaListener(
        topics = "order-events",
        groupId = "batch-order-group",
        containerFactory = "batchFactory"
    )
    public void handleBatch(List<OrderEvent> events) {
        log.info("Received batch of {} events", events.size());

        // Process all at once — e.g., bulk insert
        orderRepository.saveAll(
            events.stream().map(this::toEntity).toList()
        );
    }

    // With full ConsumerRecords for metadata access
    @KafkaListener(
        topics = "payment-events",
        groupId = "batch-payment-group",
        containerFactory = "batchFactory"
    )
    public void handleBatchWithMetadata(
            List<ConsumerRecord<String, PaymentEvent>> records) {
        records.forEach(r ->
            log.info("partition={} offset={} key={}",
                r.partition(), r.offset(), r.key())
        );
    }
}

@Configuration
public class ManualAckConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderEvent>
            manualAckFactory(ConsumerFactory<String, OrderEvent> cf) {

        ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(cf);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }
}

@KafkaListener(
    topics = "payment-events",
    groupId = "payment-manual-group",
    containerFactory = "manualAckFactory"
)
public void processPayment(
        ConsumerRecord<String, PaymentEvent> record,
        Acknowledgment ack) {

    try {
        paymentService.process(record.value());
        ack.acknowledge();  // commit offset only after success
    } catch (RetriableException e) {
        // Don't ack — message will be re-delivered
        ack.nack(Duration.ofSeconds(5));  // retry after 5s
    }
}

@Configuration
@EnableKafkaStreams
public class KafkaStreamsConfig {

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration streamsConfig() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-streams-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
        return new KafkaStreamsConfiguration(props);
    }
}

@Component
public class OrderStreamTopology {

    @Autowired
    public void buildPipeline(StreamsBuilder builder) {
        // Real-time order aggregation by customer
        KStream<String, OrderEvent> orders = builder.stream(
            "order-events",
            Consumed.with(Serdes.String(), new JsonSerde<>(OrderEvent.class))
        );

        // Count orders per customer in 5-minute windows
        KTable<Windowed<String>, Long> orderCounts = orders
            .groupBy((key, order) -> order.getCustomerId())
            .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
            .count(Materialized.as("order-counts-store"));

        // Detect high-value orders and route to priority topic
        orders
            .filter((key, order) -> order.getTotal().compareTo(BigDecimal.valueOf(1000)) > 0)
            .to("high-value-orders",
                Produced.with(Serdes.String(), new JsonSerde<>(OrderEvent.class)));

        // Join orders with payments
        KStream<String, PaymentEvent> payments = builder.stream("payment-events");
        orders.join(
            payments,
            (order, payment) -> new OrderPaymentPair(order, payment),
            JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(10)),
            StreamJoined.with(Serdes.String(),
                new JsonSerde<>(OrderEvent.class),
                new JsonSerde<>(PaymentEvent.class))
        ).to("order-payment-joined");
    }
}

@SpringBootTest
@EmbeddedKafka(
    partitions = 3,
    topics = {"order-events", "order-events.DLT"},
    brokerProperties = {"listeners=PLAINTEXT://localhost:9092"}
)
class OrderEventFlowTest {

    @Autowired
    private KafkaTemplate<String, OrderEvent> kafkaTemplate;

    @Autowired
    private OrderEventConsumer consumer;

    @Test
    void shouldProcessOrderEvent() throws Exception {
        OrderEvent event = new OrderEvent("order-123", "user-1",
            List.of(new OrderItem("SKU-001", 2)));

        kafkaTemplate.send("order-events", event.getOrderId(), event).get();

        // Wait for async consumer
        await().atMost(Duration.ofSeconds(10))
            .untilAsserted(() ->
                verify(orderService).process(argThat(e ->
                    e.getOrderId().equals("order-123"))));
    }
}

@SpringBootTest
@Testcontainers
class KafkaIntegrationTest {

    @Container
    static KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("confluentinc/cp-kafka:7.5.0"))
        .withKraft();  // KRaft mode — no ZooKeeper needed

    @DynamicPropertySource
    static void kafkaProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
    }

    @Autowired
    private KafkaTemplate<String, OrderEvent> kafkaTemplate;

    @Test
    void shouldConsumeAndProcessEvent() {
        OrderEvent event = new OrderEvent("order-456", "user-2",
            List.of(new OrderItem("SKU-002", 1)));

        kafkaTemplate.send("order-events", event.getOrderId(), event);

        await().atMost(Duration.ofSeconds(15))
            .untilAsserted(() ->
                assertThat(orderRepository.findById("order-456")).isPresent());
    }
}