Spring Kafka Hata Yönetimi — Bozuk Mesaj Partition'ı Durdurur mu?

1
@Configuration
2
class KafkaErrorConfig {
3

4
    // Boot wires a CommonErrorHandler bean into its listener container factory.
5
    @Bean
6
    DefaultErrorHandler kafkaErrorHandler(KafkaTemplate<Object, Object> dltTemplate) {
7
        var recoverer = new DeadLetterPublishingRecoverer(dltTemplate,
8
                // Same partition number on the DLT: create card-transactions-dlt with at least as many partitions.
9
                (record, ex) -> new TopicPartition(record.topic() + "-dlt", record.partition()));
10

11
        var backOff = new ExponentialBackOffWithMaxRetries(3);   // 1s, 2s, 4s, then recover
12
        backOff.setInitialInterval(1_000);
13
        backOff.setMultiplier(2.0);
14

15
        var handler = new DefaultErrorHandler(recoverer, backOff);
16
        // Data that can never succeed goes straight to the DLT: no retries, no blocked partition.
17
        handler.addNotRetryableExceptions(InvalidTransactionException.class, MethodArgumentNotValidException.class);
18
        return handler;
19
    }
20

21
    // The DLT receives raw bytes for records that failed to deserialize,
22
    // and CardTransactionCaptured objects for everything else.
23
    @Bean
24
    KafkaTemplate<Object, Object> dltTemplate(ProducerFactory<Object, Object> producerFactory) {
25
        var serializer = new DelegatingByTypeSerializer(Map.of(
26
                byte[].class, new ByteArraySerializer(),
27
                CardTransactionCaptured.class, new JsonSerializer<>()));
28
        var factory = new DefaultKafkaProducerFactory<Object, Object>(
29
                producerFactory.getConfigurationProperties(), new StringSerializer(), serializer);
30
        return new KafkaTemplate<>(factory);
31
    }
32
}

1
@Component
2
class CardTransactionListener {
3

4
    private final ProcessedEvents processed;
5
    private final Ledger ledger;
6

7
    CardTransactionListener(ProcessedEvents processed, Ledger ledger) {
8
        this.processed = processed;
9
        this.ledger = ledger;
10
    }
11

12
    @KafkaListener(topics = "card-transactions", groupId = "ledger")
13
    @Transactional
14
    public void on(CardTransactionCaptured event) {
15
        // 1. Idempotency first: a retry or a redelivery must not debit the card twice.
16
        if (!processed.markIfFirst(event.eventId())) return;
17

18
        // 2. Validation: a permanent problem, classified as not retryable.
19
        //    Retrying a negative amount or an unknown currency will never make it valid.
20
        if (event.amount().signum() <= 0 || !SupportedCurrencies.contains(event.currency())) {
21
            throw new InvalidTransactionException("unpostable transaction " + event.transactionId());
22
        }
23

24
        // 3. The side effect: a double-entry posting (card account debit, merchant
25
        //    settlement credit). A transient failure here (DB deadlock) is retried;
26
        //    the transaction rolls back, so the processed-marker is rolled back too.
27
        ledger.post(event.cardAccountId(), event.merchantId(), event.amount(), event.currency());
28
    }
29
}

1
@Repository
2
class ProcessedEvents {
3

4
    private final JdbcTemplate jdbc;
5

6
    ProcessedEvents(JdbcTemplate jdbc) {
7
        this.jdbc = jdbc;
8
    }
9

10
    /**
11
     * Returns true only for the first delivery of an event.
12
     * Table: processed_events(event_id UUID PRIMARY KEY, processed_at TIMESTAMP NOT NULL)
13
     */
14
    boolean markIfFirst(UUID eventId) {
15
        int inserted = jdbc.update("""
16
                INSERT INTO processed_events (event_id, processed_at)
17
                VALUES (?, now())
18
                ON CONFLICT (event_id) DO NOTHING
19
                """, eventId);
20
        return inserted == 1;
21
    }
22
}

1
@Component
2
class CardTransactionDltListener {
3

4
    private static final Logger log = LoggerFactory.getLogger(CardTransactionDltListener.class);
5

6
    private final MeterRegistry meters;
7

8
    CardTransactionDltListener(MeterRegistry meters) {
9
        this.meters = meters;
10
    }
11

12
    // Raw bytes on purpose: a record that could not be deserialized must still be readable here.
13
    @KafkaListener(topics = "card-transactions-dlt", groupId = "ledger-dlt",
14
            properties = "value.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer")
15
    public void on(ConsumerRecord<String, byte[]> record) {
16
        String error = header(record, KafkaHeaders.DLT_EXCEPTION_MESSAGE);
17
        String source = header(record, KafkaHeaders.DLT_ORIGINAL_TOPIC);
18

19
        log.error("dead letter from {} key={} error={}", source, record.key(), error);
20
        meters.counter("ledger.dlt.records", "source", String.valueOf(source)).increment();
21
        // No automatic replay: an unposted card transaction means the customer's balance
22
        // and the scheme's settlement file will disagree. The reconciliation team fixes
23
        // the data or the code, then replays on purpose.
24
    }
25

26
    private static String header(ConsumerRecord<?, ?> record, String name) {
27
        Header header = record.headers().lastHeader(name);
28
        return header == null ? null : new String(header.value(), StandardCharsets.UTF_8);
29
    }
30
}

1
public record CardTransactionCaptured(
2
        UUID eventId,
3
        String transactionId,        // the authorization's reference, for reconciliation
4
        long cardAccountId,
5
        String merchantId,
6
        BigDecimal amount,
7
        String currency,
8
        Instant occurredAt) {
9
}

1
spring:
2
  kafka:
3
    bootstrap-servers: kafka:9092
4
    consumer:
5
      group-id: ledger
6
      auto-offset-reset: earliest
7
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
8
      # The wrapper catches deserialization failures and hands them to the error handler.
9
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
10
      properties:
11
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
12
        spring.json.trusted.packages: com.bank.events
13
        spring.json.value.default.type: com.bank.events.CardTransactionCaptured
14
        # Retries block the partition; keep total back-off well below this.
15
        max.poll.interval.ms: 300000
16
    producer:
17
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
18
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer