3. Idempotent Consumer and DLT

package boxoffice.notification;

import java.math.BigDecimal;
import java.time.Instant;
import java.util.UUID;

/** The consumer's own view of the event: unknown fields are ignored (tolerant reader). */
public record OrderEvent(
        UUID eventId,
        String eventType,
        UUID orderId,
        String showId,
        String holdId,
        String customerEmail,
        BigDecimal amount,
        Instant occurredAt) {
}

package boxoffice.notification;

import org.springframework.mail.SimpleMailMessage;
import org.springframework.mail.javamail.JavaMailSender;
import org.springframework.stereotype.Component;

@Component
public class EmailSender {

    private final JavaMailSender mail;
    private final NotificationProperties props;

    public EmailSender(JavaMailSender mail, NotificationProperties props) {
        this.mail = mail;
        this.props = props;
    }

    public void orderReceived(OrderEvent e) {
        if (e.customerEmail().startsWith("poison")) {
            // a permanent failure: retrying will never help
            throw new IllegalArgumentException("mailbox rejected: " + e.customerEmail());
        }
        simulateProviderLatency();
        SimpleMailMessage m = new SimpleMailMessage();
        m.setFrom("tickets@boxoffice.local");
        m.setTo(e.customerEmail());
        m.setSubject("BoxOffice: we received your order " + e.orderId());
        m.setText("Order " + e.orderId() + " for show " + e.showId() + " (R$ " + e.amount()
                + ") is waiting for payment.");
        mail.send(m);
    }

    /** A real e-mail API takes tens to hundreds of ms per call; Mailpit takes ~1 ms. */
    private void simulateProviderLatency() {
        try {
            Thread.sleep(props.emailLatency());
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(ex);
        }
    }
}

package boxoffice.notification;

import java.time.Duration;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.bind.DefaultValue;

@ConfigurationProperties(prefix = "boxoffice.notification")
public record NotificationProperties(
        @DefaultValue("50ms") Duration emailLatency,
        @DefaultValue("true") boolean idempotent) {
}

package boxoffice.notification;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.properties.ConfigurationPropertiesScan;

@SpringBootApplication
@ConfigurationPropertiesScan
public class NotificationApplication {

    public static void main(String[] args) {
        SpringApplication.run(NotificationApplication.class, args);
    }
}

package boxoffice.notification;

import tools.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class OrderEventListener {

    private final ObjectMapper mapper;
    private final NotificationService notifications;

    public OrderEventListener(ObjectMapper mapper, NotificationService notifications) {
        this.mapper = mapper;
        this.notifications = notifications;
    }

    @KafkaListener(topics = "order-events", groupId = "notification")
    public void onOrderEvent(ConsumerRecord<String, String> record) {
        OrderEvent event = mapper.readValue(record.value(), OrderEvent.class);
        if (event.eventId() == null || event.eventType() == null) {
            throw new IllegalArgumentException("not a BoxOffice event: " + record.value());
        }
        if (!"OrderPlaced".equals(event.eventType())) {
            return; // other event types are not for this consumer (yet)
        }
        notifications.handle(event);
    }
}

package boxoffice.notification;

import tools.jackson.core.JacksonException;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.ExponentialBackOffWithMaxRetries;

@Configuration
public class KafkaConfig {

    /** Picked up by Spring Boot's listener container factory (one CommonErrorHandler bean). */
    @Bean
    DefaultErrorHandler errorHandler(KafkaTemplate<String, String> template) {
        // after the retries, publish to <topic>-dlt, same partition, with the exception in headers
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);

        ExponentialBackOffWithMaxRetries backOff = new ExponentialBackOffWithMaxRetries(3);
        backOff.setInitialInterval(500L);  // 0.5 s, 1 s, 2 s
        backOff.setMultiplier(2.0);
        backOff.setMaxInterval(5_000L);

        DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, backOff);
        // permanent failures: retrying is pointless, go straight to the DLT
        handler.addNotRetryableExceptions(JacksonException.class, IllegalArgumentException.class);
        return handler;
    }
}

package boxoffice.notification;

import java.nio.charset.StandardCharsets;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.stereotype.Component;

@Component
public class DeadLetterListener {

    private static final Logger log = LoggerFactory.getLogger(DeadLetterListener.class);

    private final JdbcClient jdbc;

    public DeadLetterListener(JdbcClient jdbc) {
        this.jdbc = jdbc;
    }

    @KafkaListener(topics = "order-events-dlt", groupId = "notification-dlt")
    public void onDeadLetter(ConsumerRecord<String, String> record) {
        // the listener's exception is wrapped: the CAUSE class says what really failed
        Header cause = record.headers().lastHeader(KafkaHeaders.DLT_EXCEPTION_CAUSE_FQCN);
        String why = cause == null ? "?" : new String(cause.value(), StandardCharsets.UTF_8);
        log.warn("dead letter at {}-{}@{}: {}", record.topic(), record.partition(), record.offset(), why);
        jdbc.sql("""
                INSERT INTO dead_letters (topic, partition, record_offset, record_key, payload, reason)
                VALUES (?, ?, ?, ?, ?, ?)
                """)
                .params(record.topic(), record.partition(), record.offset(), record.key(), record.value(), why)
                .update();
    }
}

package boxoffice.notification;

import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;

/** The synchronous "before" path, used by order in NOTIFY_MODE=sync. */
@RestController
public class NotificationController {

    private final NotificationService notifications;

    public NotificationController(NotificationService notifications) {
        this.notifications = notifications;
    }

    @PostMapping("/notifications")
    @ResponseStatus(HttpStatus.NO_CONTENT)
    public void send(@RequestBody OrderEvent event) {
        notifications.handle(event);
    }
}