Summer Kafka

Summer Kafka — Consumer Idempotency & Error Handling

Gate: Verify summer-core loaded + io.f8a.summer:summer-platform in build.gradle. Producers must be on summer-data-outbox 0.3.1+ — messages must carry ob.lsn header.

Modules: summer-kafka-consumer | summer-kafka-consumer-autoconfigure (both added 0.3.1) Package: io.f8a.summer.kafka.consumer.* Activation: Auto when summer-kafka-consumer-autoconfigure is on the classpath. There is no enabled flag — presence of the dependency is the opt-in. Disable by dropping the dependency.

Tracks LATEST stable (0.3.5). First shipped 0.3.1; no 0.3.5 overlay (no kafka-consumer changes since 0.3.1).

Why it exists

Pre-0.3.1: every consumer needed its own dedup table (one row per event) or a domain-unique key + DuplicateKeyException. Both scale with message volume and produce hot indexes. OutboxConsumerIdempotency watermarks by (consumer_group, topic, partition) — matching Kafka's native offset granularity — so storage cost is O(groups × topics × partitions), typically a few hundred rows per service regardless of throughput.

Debezium preserves PostgreSQL LSN order within a topic-partition. Scheduler-mode publishers emit createdAt.toEpochNanos() as best-effort — stable under normal load, not authoritative under extreme concurrency. Strict ordering → use CDC mode.

Schema — outbox_consumer_watermark

One row per (group, topic, partition). Schema must exist before consumers run; ConsumerWatermarkValidator checks on startup (disable: f8a.kafka.consumer.idempotency.validate-on-startup=false).

CREATE TABLE outbox_consumer_watermark (
    consumer_group  VARCHAR(255) NOT NULL,
    topic           VARCHAR(255) NOT NULL,
    partition       INTEGER      NOT NULL,
    last_lsn        BIGINT       NOT NULL,
    last_event_id   UUID,
    last_updated_at TIMESTAMPTZ  NOT NULL DEFAULT now(),
    PRIMARY KEY (consumer_group, topic, partition)
);

Column names load-bearing — ConsumerWatermarkValidator checks exact names: consumer_group, topic, partition, last_lsn, last_event_id, last_updated_at. Using lsn/updated_at instead fails boot. DDL also in summer-data/references/ddl-scripts.md.

Public API

Listener pattern

One rule: recordProcessed must commit in the same transaction as the business write. Crash between commits → duplicate-processing window. Reactive: wrap with TransactionalOperator.transactional(...).

@Component
@RequiredArgsConstructor
public class OrderEventListener {

  private final OrderService orderService;
  private final OutboxConsumerIdempotency idempotency;
  private final TransactionalOperator tx;

  @KafkaListener(topics = "orders.created", groupId = "${spring.application.name}")
  public Mono<Void> onCreated(
      @Payload OrderCreatedEvent event,
      @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
      @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
      @Header("ob.lsn") long lsn,
      @Header(value = "ob.eid", required = false) String eid) {

    var ctx = new IdempotencyContext(
        "${spring.application.name}", topic, partition, lsn, eid);

    return idempotency.isProcessed(ctx)
        .flatMap(seen -> seen
            ? Mono.empty()
            : orderService.handle(event)
                .then(idempotency.recordProcessed(ctx)))
        .as(tx::transactional);
  }
}

Imperative consumers use @Transactional instead — same rule, same wiring.

Error handling (DefaultErrorHandler + DLT)

Autoconfigure wires:

• Exponential-backoff retry (ExponentialBackOffWithMaxRetries). Retries up to maxAttempts; delay = initialInterval × multiplier^(n-1) capped at maxInterval.
• On exhaustion, DeadLetterPublishingRecoverer forwards to <topic>.DLT via dltKafkaTemplate.

f8a:
  kafka:
    consumer:
      idempotency:
        enabled: true                # default; presence of the dep is the real switch
        validate-on-startup: true
      retry:
        max-attempts: 3
        initial-interval: 1s
        multiplier: 2.0
        max-interval: 30s

Marking exceptions non-retryable

Register a DefaultErrorHandlerCustomizer bean — don't replace the whole handler:

@Bean
DefaultErrorHandlerCustomizer markValidationFatal() {
  return handler -> handler.addNotRetryableExceptions(
      MethodArgumentNotValidException.class,
      DeserializationException.class);
}

Custom DLT producer

Override default fire-and-forget template only if idempotence on DLT writes needed (rare):

@Bean("dltKafkaTemplate")
KafkaTemplate<?, ?> dltKafkaTemplate(ProducerFactory<?, ?> pf) {
  KafkaTemplate<?, ?> t = new KafkaTemplate<>(pf);
  t.setObservationEnabled(true);
  return t;
}

JSON consumer factory helpers

For typed ConcurrentKafkaListenerContainerFactory instead of Spring Boot default:

@Bean("orderListenerContainerFactory")
ConcurrentKafkaListenerContainerFactory<String, OrderCreatedEvent> orderListenerFactory(
    KafkaProperties props, DefaultErrorHandler errorHandler) {
  var cf = SummerKafkaConsumerFactories.jsonConsumerFactory(
      props, OrderCreatedEvent.class);
  return SummerKafkaConsumerFactories.jsonListenerContainerFactory(props, cf, errorHandler);
}

Reference from listener: @KafkaListener(topics = "...", containerFactory = "orderListenerContainerFactory").

Gradle

implementation 'io.f8a.summer:summer-kafka-consumer'
implementation 'io.f8a.summer:summer-kafka-consumer-autoconfigure'
implementation 'io.f8a.summer:summer-data-r2dbc'   // for R2dbcOutboxConsumerIdempotency

Rules

• Always call idempotency.recordProcessed(ctx) in the same reactive transaction as the business write. Separate .subscribe() or flatMap outside transaction breaks the contract.
• Never skip watermark validator. Missing/misshapen outbox_consumer_watermark → silently re-processes from offset 0 on restart.
• Never invent own (consumerGroup, topic, partition) keying. Use KafkaHeaders.RECEIVED_TOPIC/RECEIVED_PARTITION directly — not static topics = value (misleading for pattern listeners).
• Always bind @Header("ob.lsn") long — 8-byte big-endian since 0.3.1. Older string-encoded headers produce wrong values silently. Producers must be on summer-data-outbox 0.3.1+.
• Always point DLT consumers at <topic>.DLT — where DeadLetterPublishingRecoverer writes.
• Never set idempotency.enabled=true without the dependency — no effect without autoconfigure jar.

References

• references/versions/0.3.1.md — module introduction + ob.lsn binary header change
• references/listener-examples.md — full listener wiring, imperative-mode, multi-topic patterns

Full feature × version table: summer-core/references/version-matrix.md

Related Skills

• summer-data — producer side (OutboxService.saveEvent, KafkaOutboxPublisher, ob.lsn header)
• summer-core — gate; Txid/Ufid types in event payloads
• summer-rest — handlers that initiate writes published via outbox