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"  // Creates 3 consumer threads
    )
    public void onOrderCreated(
        ConsumerRecord<String, OrderEvent> record,
        Acknowledgment ack  // Injected with ack-mode: MANUAL or BATCH
    ) {
        log.info("Received order: partition={}, offset={}, key={}",
            record.partition(), record.offset(), record.key());

        try {
            orderService.processOrder(record.value());
            ack.acknowledge();  // Commit offset after successful processing
        } catch (NonRetryableException e) {
            log.error("Non-retryable error, sending to DLT", e);
            ack.acknowledge(); // Don't block - error handler sends to DLT
            throw e;           // Let DefaultErrorHandler handle it
        }
    }

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