@RetryableTopic (Non-Blocking Retry)

@Service
@Slf4j
public class OrderConsumer {

    @RetryableTopic(
        attempts = "4",           // the original attempt plus 3 retries
        backOff = @BackOff(
            delay = 1_000,        // 1 second initial delay
            multiplier = 2,       // doubles each retry: 1 s, 2 s, 4 s
            maxDelay = 10_000     // capped at 10 seconds
        ),
        dltTopicSuffix = "-dlt",
        // With an exponential back off the retry topics are named by delay:
        // orders.order.created-retry-1000, -retry-2000, -retry-4000, and orders.order.created-dlt
        include = {TransientServiceException.class},   // retry only these
        exclude = {InvalidEventException.class}        // never retry these
    )
    @KafkaListener(topics = "orders.order.created", groupId = "order-processor")
    public void onOrderCreated(OrderEvent event, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        log.info("Processing order from topic: {}", topic);
        orderService.processOrder(event); // a thrown exception sends the record to the next retry topic
    }

    // Must be in the same class as the listener
    @DltHandler
    public void handleDlt(OrderEvent event, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        log.error("DLT received from topic {}: orderId={}", topic, event.getOrderId());
        // Alert, store for manual review, or count it
    }
}