Batch Listener Pattern

@Service
@Slf4j
public class BatchOrderConsumer {

    // Batch listener receives all records from one poll()
    @KafkaListener(
        topics = "orders.order.created",
        groupId = "batch-processor",
        containerFactory = "batchListenerFactory"  // Uses BATCH ack mode
    )
    public void onOrdersBatch(
        List<ConsumerRecord<String, OrderEvent>> records,
        Acknowledgment ack
    ) {
        log.info("Processing batch of {} orders", records.size());

        List<Order> orders = records.stream()
            .map(r -> orderMapper.toOrder(r.value()))
            .collect(Collectors.toList());

        // Bulk insert — much more efficient than one-by-one
        orderRepository.saveAll(orders);

        // Commit after entire batch processed
        ack.acknowledge();
    }

    // Even simpler: value-only batch
    @KafkaListener(topics = "analytics.events", containerFactory = "batchListenerFactory")
    public void onEventsBatch(List<AnalyticsEvent> events) {
        analyticsService.bulkProcess(events);
        // Auto-ack after method returns (with BATCH ack mode)
    }
}

// Config
@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> batchListenerFactory(
    ConsumerFactory<String, Object> cf
) {
    var factory = new ConcurrentKafkaListenerContainerFactory<String, Object>();
    factory.setConsumerFactory(cf);
    factory.setBatchListener(true); // KEY: enables batch mode
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);
    factory.getContainerProperties().setPollTimeout(3000);
    return factory;
}