Batch Listener Pattern

@Service
@Slf4j
public class BatchOrderConsumer {

    // Receives every record from one poll(); ackMode MANUAL lets us acknowledge when the batch is done
    @KafkaListener(
        topics = "orders.order.created",
        groupId = "batch-processor",
        containerFactory = "batchListenerFactory",
        ackMode = "MANUAL"
    )
    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());

        orderRepository.saveAll(orders);   // one bulk insert for the whole batch

        ack.acknowledge();                 // commit after the entire batch is processed
    }

    // Value-only batch: with the factory's BATCH ack mode the offsets are committed
    // after the method returns normally
    @KafkaListener(topics = "analytics.events", containerFactory = "batchListenerFactory")
    public void onEventsBatch(List<AnalyticsEvent> events) {
        analyticsService.bulkProcess(events);
    }
}

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