KIP-360 — producer epoch bump

Status: implemented — see the KIP index.

What the KIP changes in Apache Kafka

KIP-360 (Kafka 2.5) lets a producer recover from state loss without losing its identity. InitProducerId v3+ can carry the producer's current PID and epoch; the coordinator responds by bumping the epoch on the same PID rather than minting a new producer, so retried batches from the old session are fenced (PRODUCER_FENCED) instead of poisoning the log, and UNKNOWN_PRODUCER_ID becomes retriable. When the 16-bit epoch is exhausted, the coordinator rotates to a fresh PID.

How kaas implements it

The rejoin contract lives in crates/kaas-coordinator/src/txn_state.rs::get_or_allocate_with_timeout: the first InitProducerId for a transactional.id allocates a fresh PID at epoch 0; every subsequent call returns the same PID with epoch + 1. At epoch == i16::MAX the entry rotates to a fresh PID at epoch 0, matching the KIP's exhaustion rule. The timeout reaper's abort path also bumps the epoch, so a producer returning after its transaction was reaped is fenced rather than resumed.

The bump is then propagated in two rings (fencing details):

• Cross-partition, in-process: after any bump to epoch > 0, the handler (crates/kaas-broker/src/handlers/init_producer_id.rs) calls the engine's fence_producer_epoch, which advances the PID's epoch and clears the dedupe window on every partition this broker leads — a zombie batch is rejected even on partitions the new session hasn't touched.
• Cross-broker, via the shared volume: the bump is appended to this broker's outbound fence file, /data/__cluster/producer_fences/from-<broker>.json (crates/kaas-coordinator/src/fence_log.rs); peers' FenceWatcher (crates/kaas-broker/src/fence_watcher.rs) polls the directory every 2 s and applies newer (pid, epoch) pairs. Polling is deliberate: inotify does not fire for another NFS client's writes.

Where kaas differs from Apache: the v3+ request's producer_id / producer_epoch fields are decoded but not validated — every InitProducerId for a known transactional.id bumps, unconditionally. The single-writer guarantee is preserved (the highest epoch wins and everything older is fenced), but Apache's claimed-identity check — which lets the coordinator reject a stale caller with PRODUCER_FENCED rather than hand it a newer epoch — has no counterpart here. In the boot window or dev mode, before a TxnStateStore is wired, the handler falls back to a fresh PID and logs that rejoin fencing is disabled — graceful degradation, not silent.

How it's verified

Handler tests in init_producer_id.rs: transactional_rejoin_bumps_epoch (same PID, epochs 0 → 1 → 2) and rejoin_appends_to_fence_log_for_broadcast (epoch 0 does not broadcast; each bump overwrites the outbound entry). Store tests in txn_state.rs: epoch_overflow_rotates_to_fresh_pid, epoch_mismatch_fences, reaper_aborts_overdue_bumps_epoch_fires_hook. Fence propagation: append_lower_or_equal_epoch_is_noop and append_higher_epoch_overwrites in fence_log.rs; applies_peer_fences, skips_self_file, dedupe_across_ticks, and higher_epoch_after_first_apply_fires_once_more in fence_watcher.rs; fence_bumps_epoch_and_clears_window in crates/kaas-storage/src/idempotence.rs. Shell suite: scripts/kafka-txn-coordinator.sh probes the rejoin-epoch wire contract against a live broker.