Basic @KafkaListener Consumer

@Service
@Slf4j
public class OrderConsumer {

    private final OrderService orderService;

    public OrderConsumer(OrderService orderService) {
        this.orderService = orderService;
    }

    @KafkaListener(
        topics = "orders.order.created",
        groupId = "order-processor",
        concurrency = "3",     // three consumers (threads) in this container
        ackMode = "MANUAL"     // per-listener override of the container's ack mode (Spring Kafka 4.1+)
    )
    public void onOrderCreated(ConsumerRecord<String, OrderEvent> record, Acknowledgment ack) {
        log.info("Received order: partition={}, offset={}, key={}",
            record.partition(), record.offset(), record.key());

        orderService.processOrder(record.value());
        ack.acknowledge();   // MANUAL: the offset is committed with the same semantics as BATCH once you acknowledge
        // If processOrder throws, do not acknowledge: the container's error handler decides what happens next
        // (see the DefaultErrorHandler topic).
    }

    // Simple value-only listener (the message converter deserializes the payload)
    @KafkaListener(topics = "orders.order.updated")
    public void onOrderUpdated(OrderEvent event) {
        log.info("Order updated: {}", event.getOrderId());
        orderService.updateOrder(event);
    }
}