Implementing Message Queue Logic

@Service @RequiredArgsConstructor
public class OrderProducer {
    private final KafkaTemplate<String, OrderPlaced> kafka;
    public void publish(OrderPlaced e) {
        kafka.send("orders", e.orderId().toString(), e);
    }
}

@KafkaListener(topics = "orders", groupId = "billing")
public void onOrder(OrderPlaced e, Acknowledgment ack) {
    billing.charge(e);
    ack.acknowledge();   // manual commit after success
}

@RetryableTopic(
    attempts = "5",
    backoff = @Backoff(delay = 1000, multiplier = 2.0),
    dltStrategy = DltStrategy.FAIL_ON_ERROR)
@KafkaListener(topics = "orders")
public void onOrder(OrderPlaced e) { billing.charge(e); }