3 · Use
@Service
class PaymentEventPublisher {
private static final Logger log = LoggerFactory.getLogger(PaymentEventPublisher.class);
private final KafkaTemplate<String, PaymentCreated> kafka;
// constructor omitted
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) // only after the DB commit
void on(PaymentCreated event) {
kafka.send("payments.events", event.paymentId().toString(), event) // key → partition → order
.whenComplete((result, ex) -> {
if (ex != null) log.error("send failed for {}", event.paymentId(), ex);
});
}
}
@Component
class NotificationListener {
private final ProcessedEvents processed; // table with a unique event_id
private final Mailer mailer;
// constructor omitted
@KafkaListener(id = "notifications", topics = "payments.events")
void on(PaymentCreated event) {
if (!processed.markIfNew(event.eventId())) return; // duplicate delivery → skip
mailer.sendReceipt(event);
}
}
@Bean
DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
return new DefaultErrorHandler(new DeadLetterPublishingRecoverer(template),
new FixedBackOff(1000L, 3)); // 3 retries, then payments.events-dlt
}