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::init_producer_id, a port of Apache's prepareInitProducerIdTransit: the first InitProducerId for a transactional.id allocates a fresh PID at epoch 0; every subsequent call returns the same PID with epoch + 1. Once the epoch reaches i16::MAX - 1 (Apache's isEpochExhausted — the last epoch is kept back for fencing) the entry rotates to a fresh PID at epoch 0. 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 v3+ request's producer_id / producer_epoch are the caller's claimed identity (gh #201), validated exactly as Apache 3.7 does:

• Exactly one of the two set to -1 is INVALID_REQUEST, on both the idempotent and the transactional path. Beyond that the idempotent path ignores the claim and hands out a fresh PID, as Apache does.
• The claimed PID must be the entry's current one — or its previous one with an exhausted epoch, which is a retry of a rotation whose response was lost. Anything else is PRODUCER_FENCED.
• A claimed epoch equal to the current one bumps it and remembers the old epoch (lastEpoch on the slot entry, Apache's lastProducerEpoch); a claim equal to that remembered epoch is a retry of the bump and gets the current epoch back unchanged. Any other claimed epoch is PRODUCER_FENCED — so a zombie that re-calls InitProducerId after a newer instance took over is fenced, rather than handed a newer epoch that fences the live session.
• A transactional.id with no entry accepts any claim (Apache's "metadata just created" case: a producer recovering from UNKNOWN_PRODUCER_ID).

A rejoin that finds the entry Ongoing aborts the in-flight transaction before anything else: the entry moves to PrepareAbort at epoch + 1 with its partition list intact, the handler fences that epoch, writes the ABORT markers and completes (EndTxn's gh #225 order, with the marker reconcile as the backstop), and answers CONCURRENT_TRANSACTIONS. The producer's retry then bumps a second time onto a clean entry, so the fence epoch is never handed out. A rejoin while the entry is Prepare* answers CONCURRENT_TRANSACTIONS until its markers land.

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: 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. Slot entries written before gh #201 carry no remembered epoch, so a retried claimed bump that spans the upgrade is fenced rather than recognised (Apache has the same gap for metadata written before IBP 2.4).

How it's verified

Handler tests in init_producer_id.rs: transactional_rejoin_bumps_epoch (same PID, epochs 0 → 1 → 2), rejoin_appends_to_fence_log_for_broadcast (epoch 0 does not broadcast; each bump overwrites the outbound entry), stale_claim_is_producer_fenced, current_claim_bumps_and_its_retry_is_stable, half_set_claim_is_invalid_request, idempotent_path_ignores_a_claim, and rejoin_mid_ongoing_aborts_it_and_answers_concurrent (fence broadcast, ABORT marker on the log, CompleteAbort, clean retry). Store tests in txn_state.rs: epoch_overflow_rotates_to_fresh_pid, epoch_exhaustion_rotates_one_short_of_max, a_claimed_bump_is_retry_safe, a_stale_claim_is_fenced_not_bumped, a_retried_rotation_gets_the_same_answer, the_pre_rotation_pid_is_stale_once_the_new_session_moves, a_rejoin_mid_ongoing_prepares_an_abort_at_a_bumped_epoch, 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.