Conversation
`persistPeriodicallyAndUnloadOrphaned` removed a key whenever the unload threshold was crossed, whether or not the persist that preceded it went through. With `ignorePersistErrors` on, a failed persist is logged and the key is unloaded anyway; `KeyContext.remove` also drops the offset it held, so the partition can commit past state that never reached the store and the next recovery skips those events. Gate the unload on the persist outcome: an unpersisted key stays loaded and keeps holding its offset until a later tick persists it.
|
Important Draft PR not reviewedDraft PRs are not automatically reviewed by default.
To automatically review draft PRs, update your CodeRabbit configuration: reviews:
auto_review:
drafts: trueThanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
tobiajo
force-pushed
the
tj/tolerate-generation-fence
branch
from
September 8, 2026 08:17
75d566a to
c40f8a8
Compare
tobiajo
changed the title
tobiajo
marked this pull request as draft
September 8, 2026 09:45
Contributor
Author
|
Discovered an issue with this change. Will iterate on it with better testing before undrafting. |
Contributor
Author
|
An alternative to this change is to simply discourage |
In transactional snapshot mode the broker rejects a transactional offset commit whose consumer generation is stale (ILLEGAL_GENERATION, raised by kafka-clients as CommitFailedException), which aborts the transaction carrying the snapshot write. The flow failed on it, on both the persist path and the periodic commit path, and the code called that failure the fence. The rejection already is the fence: the transaction aborted, nothing landed, the producer stays usable after the abort, and the consumer refreshes its generation once it completes the rebalance. Failing the flow adds a group leave and a rejoin, each bumping the generation and fencing every peer with a transaction in flight. Under the cooperative assignor a still-valid owner is fenced routinely, because retained partitions keep committing while a rebalance completes, so a rolling deploy turns into a restart cascade across the group. `GroupCommit.commitBatch` classifies the rejection once, as `GenerationFencedError`. The persist path warns and keeps the key dirty and holding its offset; the periodic commit path warns and leaves the offset uncommitted, advancing `committedOffset` only after a successful schedule so the same offset is retried on the next tick. Every other error is handled as before. The docs and the integration test that described the crash as the fence now describe the retry.
evolution-gaming#935 tolerates the stale-generation fence on the two waiters that write state - the periodic persist and the periodic offset commit - but not on the third: the tombstone `TickToState` and `FoldToState` write when a tick or a fold empties a key's state. In preprod that waiter took 3661 of the 14945 fences tolerated across six rollouts on 2026-09-08, 30-57% of each rollout's; every one of them would otherwise have failed the flow and fenced the peers in turn. `persistence.delete` is a transaction like any other write here: the rejection aborted it, so no tombstone landed and the key has to stay. Removing it would drop its held offset too, and let the partition commit past a snapshot that is still in the store. The key is kept, with the empty state the tick or fold gave it, and deleted again on the next tick. Every other delete error still fails the flow. Reaching the error channel needs `MonadThrow`, so the tolerant path comes as new overloads taking `remove` - the effect that takes the key out of the partition once the tombstone lands - and the existing constructors keep their `Monad` bound and their behaviour. `KeyFlow.of` and `KeyFlowOf.apply` do widen, since they build that path: source-compatible (every effect they are used with satisfies `MonadThrow` already), binary- incompatible for a caller that does not recompile, hence the two MiMa filters. Drop those if this ships as a major instead. A key left with an empty state by a tolerated fence must keep its timers, or the next tick never comes to retry the delete; that is the next commit. Issue evolution-gaming#938.
A key whose tombstone the previous commit keeps through a fence is left with an empty state, and `KeyFlow` cancelled a key's timers on exactly that. The key then held its offset with no timer to ever retry the delete: in preprod on 2026-09-08 the committed offset froze on 21 of 32 partitions of one service, precisely those whose delete had been fenced, and lag grew at the input rate until rollback. The rule the cancellation wanted is "the key is gone", so track that: `KeyFlow` owns a `removed` flag, set by the remove effect it hands to `FoldToState` and `TickToState`, and cancels on it. The protection is intact - once removed, the timer flow is skipped, so a dropped key is never flushed - and an emptied key keeps ticking until its tombstone lands. Two orderings the retry rests on, now written down: - a `TickOption` maps an empty state to an empty state, so the next tick of a key waiting for its tombstone asks for the delete again instead of resurrecting it; - within one `onTimer` the tick runs before the timer flow, so the re-attempted delete marks the key persisted before the periodic persist could flush the emptied buffer and hold an offset ahead of the tombstone. `PartitionFlowSpec` drives the real chain for this - `PartitionFlow`, the periodic timer flow, memory timers, an eviction tick, a snapshot database that fences the first tombstone - because the unit tests cannot see the pin: `Timers.trigger` runs `KeyFlow.onTimer` only for a registered timer, and the timer flow is what registers the next one, which is the link the old rule cut. It asserts the tombstone did not land, that the next tick deletes again, and that the partition then commits past the key; under the old rule the second delete never comes. Issue evolution-gaming#938.
evolution-gaming#935 matched `CommitFailedException` anywhere in the cause chain, which also catches `KafkaException("Cannot execute transactional method because we are in an error state", cause = CommitFailedException)`. That is not this transaction being fenced. Read against kafka-clients 4.3.1: the fence is raised unwrapped by `sendOffsetsToTransaction`, from the single site mapping ILLEGAL_GENERATION and UNKNOWN_MEMBER_ID, and as an abortable error, so `commitBatch`'s abort clears it and the next transaction opens on the same producer - exactly what tolerating it assumes. The wrapped shape instead says an earlier fence is still recorded because nothing aborted it, and one reachable variant of that - a `commitTransaction` that timed out, leaving an unacked pending transition the abort can no longer get past - fails every future transaction identically. Tolerated without bound, that is a stall logged as a fence; failing the flow re-creates the producer, which is the only recovery. So the classification is now the top-level type, and the chain walk goes. All 47 fences traced in preprod arrived bare, and the broker-level tests in `TransactionalKafkaPersistenceSpec` see the same shape. `GroupCommitSpec` pins both: the bare fence classified, the wrapped one surfacing as it is. Issue evolution-gaming#938.
Tolerating the fence removes the only signal there was: before, a fenced
flow failed and restarted, which every monitor sees. Now it warns and
retries, and the WARN lines - about 580 per rolling deploy of one preprod
service - are not something to alert on. So count them:
`snapshot_write_fenced_total{topic,partition}`, incremented in
`commitBatch`, where the fence is the shared outcome of the transaction.
Counting at the write instead would multiply each fence by the batch size,
which is why this is not a `SnapshotDatabase` wrapper like the other
snapshot metrics but a small `SnapshotWriteMetrics` the writer holds.
`FlowMetrics.snapshotWriteMetrics` comes with a default body returning the
empty instance, and `transactional` takes the metrics through a new
overload, so neither addition breaks an implementation or a caller. The
partition label is the input partition, as in the other snapshot metrics.
A rate above zero outside a rebalance window, or one that does not
subside, is the alerting signal: the fence is only harmless while the
consumer keeps completing rebalances.
Issue evolution-gaming#938.
evolution-gaming#935 tolerates a fence for as long as it keeps arriving, which invites the question of a bound - the variant this PR did not take carried a one-minute `fenceTolerance`. Measured instead of guessed: six rolling deploys of two services against one preprod cluster (classic protocol, cooperative-sticky) tolerated 14945 fences, arriving in 42 per-member bursts across 28 members (a gap over 10 s starting a new burst) with a median of 1.1 s, a p90 of 3.9 s and a maximum of 20.6 s; none reached 30 s. A local harness with a third member joining and leaving in a loop put 793 of its 825 fences inside one 3.5 s window. Nothing there wants a bound. The consumer's own rejoin ends every run - a lagging member on completing its in-flight round, an evicted one via `onPartitionsLost` - so a bound could only fire on runs the consumer is about to end anyway; and a plausible-looking one (a minute) would fire during the longest rebalances, where a peer is in eager recovery, which is exactly when restarting the storm is worst. The case a bound is imagined for - polls succeeding while the generation never becomes valid - is better detected by the committed offset not advancing while the input moves, an alert that also covers a pinned key, a stalled fold and a wedged producer, with `snapshot_write_fenced_total` to attribute it. Written up under Tolerating the fence, with the bound listed among the rejected alternatives. Issue evolution-gaming#938.
Tolerating the fenced delete opened a hole the crash used to close. `Snapshots.delete` emptied the buffer before calling the database, so with the tombstone still owed any `flush` of that key succeeded on an empty buffer - and `attemptToPersist` reads that as persisted and holds the key's offset. From there `flushOnRevoke = true`, `unloadOrphaned` or a `canUnload` tick removes the key outright, releasing the held offset, and the partition commits past a snapshot that is still in the store: on the tick path the eviction is merely deferred, but on the fold path the recovered state is one the fold had purged. So the buffer is emptied only once the delete lands, and while it is owed the key is marked tombstone-pending: `flush` then retries the delete instead of reporting success. A fenced retry surfaces as the same `GenerationFencedError` the persist path already tolerates, so the key stays loaded, holds its offset and is not unloaded. A state appended before the tombstone lands drops the mark - the key is not going away after all, and the next flush writes the new snapshot over the one the delete did not remove; without that the retry would delete a key that had come back. Nothing else changes: the mark is only ever set by a delete that did not go through, and every non-fence delete error still fails the flow. This also makes the tick-before-timer-flow ordering belt-and-braces rather than load-bearing: a periodic persist that now runs on an emptied buffer retries the tombstone instead of vacuously advancing the offset. Issue evolution-gaming#938.
The transactional suites so far inject a stale generation into a single flow. That covers the coordinator's validation, not what the tolerance is for: several members alive at once, made stale by rebalances they did not choose. Two suites now do that, and one gap in the injected set is closed. `FenceStormSpec` runs two instances in one group under cooperative-sticky with transactional snapshots, continuous input and a third member joining and leaving in a loop. It asserts no flow failure, no give-up, no restart, at least one fence actually provoked (a run that provoked none is inconclusive, not a pass), a tombstone written, every partition's committed offset still advancing after the churn, and the flows draining to the end offsets - the oracle the pinned partitions of 2026-09-08 would have tripped. Correctness is then checked against the store, not the flows: the input is replayed from the committed offsets on top of the `read_committed` snapshots and must fold to the same state per key, with keys that never close so a lost record cannot hide behind a tombstone. Locally: 183 s, 747 fences across the persist, delete and offset-commit waiters, 3918 tombstones, zero retries. `EvictionFenceSpec` covers the other rejection, `UNKNOWN_MEMBER_ID`. A member stalls inside its fold, which runs on the poll loop's thread of control, so the coordinator drops it past `max.poll.interval.ms` while it keeps its flows, its state and its producer; a second member takes the partition over for real and persists. The evicted member then tolerates the rejection rather than failing, nothing of its write lands, and its next poll completes the rejoin that reports the partition lost and tears its flows down - it ends up in the group owning nothing while the new owner keeps committing. The two use separate transactional ids, or the takeover's `initTransactions` would epoch-fence the stale producer before its write ever reached the offset commit. `TransactionalKafkaPersistenceSpec` gains the fenced **tombstone** the injected set was missing: the key keeps its snapshot through the fence and the next tick deletes it for real. Both new suites are paced by real coordinator timeouts and each brings up its own broker, so the module now runs its tests one at a time: beside four peers the churn suite took 38 minutes, alone it takes 3, and the module as a whole 6.
Two gaps the tolerance leaves in the docs. Which assignor pays the spurious fence at all: classic **eager** never sees one - it revokes the whole assignment in `onJoinPrepare`, before the generation bumps, so no flow is alive to carry a stale token - and pays for that with a full re-recovery on every rebalance. Classic **cooperative** keeps its retained partitions folding, flushing and committing straight through the round, which is where the fence comes from, and the classic protocol has no broker-side absorption of it: KIP-1251 covers the consumer protocol only, so `group.protocol=consumer` on brokers that carry it is the protocol-level exit and tolerance is what makes cooperative-sticky stable until then. And what tolerance costs, next to what it buys: an aborted transaction per fence, a WARN per fenced waiter (some 580 per rolling deploy of one preprod service - log volume, not an alerting signal, which is what `snapshot_write_fenced_total` is for), a key held loaded with its offset until its tombstone lands, and a committed offset that lags by the ticks the generation takes to become current. The alert to keep is the committed offset not advancing while the input moves - the one symptom a fence that never clears shares with a pinned key and a stalled fold.
The third timer flow was left out of the tolerance. `unloadOrphaned` called `persistence.flush` directly and removed the key straight after, so any error - the stale-generation fence included - failed the flow, and the fence was reachable there: the previous commit turned a `flush` that used to succeed vacuously on an already-emptied buffer into a real delete retry, which is exactly the write the broker rejects. It now goes through the same `attemptToPersist` as the other two flows, with `ignorePersistErrors = false` so every non-fence error still fails the flow as before. A fenced persist keeps the key loaded and holding its offset, and re-registers the timer so the next tick retries the delete; without that re-registration the key is pinned with no timer to ever come back to it, which is the freeze the previous commit fixed on the `KeyFlow` side. `TimerFlowOfSpec` covers both halves against a flush fenced once: the first tick does not fail and does not remove the key, and the second `trigger` fires at all - `Timers.trigger` runs `onTimer` only for a registered timer - so the retry lands and only then is the key unloaded and its offset released. The revoke-time flush is now the only path that does not tolerate the fence, which the docs said of `unloadOrphaned` too. Issue evolution-gaming#938.
`persistence-kafka-it-tests` is in the root aggregate and the shared Evolution CI workflow runs `sbt test` at the root, so the churn suite would run on every pull request, on a hosted runner. It is not shaped for that: it spends minutes racing real rebalances against a broker it brings up itself, and its `assert(fences > 0)` fails a run that provoked none - correct when the point of the run is to provoke them, a flake when it is one suite among many on a machine that may not go fast enough. Locally four runs provoked 747, 26, 23 and 16. So it takes the gate `TransactionalWriteThroughputSpec` already uses, `munitIgnore` behind an env var, with the command in the comment. The assertions are unchanged: run it and it still asserts everything it did. `EvictionFenceSpec` stays in the default run. It also asserts a fence was provoked, but it does not race for one: it stalls the fold until the coordinator drops the member, and every step waits on the broker's own verdict through `describeConsumerGroups`, so the fence is forced rather than hoped for and a run that cannot get there fails on a named timeout instead of a bare count. Issue evolution-gaming#938.
The suite carried a downstream application's retry policy - exponential with jitter, a backoff budget, an attempt limit, and a `resetOnQuiet` wrapper reimplementing the budget reset - to run the flows under. None of it was exercised: the spec asserts the retry list is empty, so with the tolerance in place no attempt ever fires, and the whole thing was 40 lines of untested reimplementation standing between the fence and the assertion. It now runs on `Retry.empty`, as `EvictionFenceSpec` already does, and the supervisor that was there to restart a flow that gave up records the failure that killed it. The assertion that no fence failed a flow is unchanged in what it catches - a failure now reaches the supervisor directly instead of a retry's `onError` first, and it still carries the cause chain into the clue - and one assertion covers what three did. Nothing is lost with it: no retry, no give-up and no restart were three readings of the same fact, that the flow failed at all. Issue evolution-gaming#938.
The "Monitoring the fence" bullet went in in the middle of the previous one, so the recovery deadline's sizing advice ended up under it, where "the value" has no referent - the fence bullet names no value to size. Move it back to the `recoveryStallTimeout` bullet it was written for; the fence bullet ends at the symptom to alert on.
`versionPolicyIntention` still said `BinaryCompatible` while two MiMa filters suppressed genuine public breaks, which is the check asserting the opposite of what the release does. `KeyFlow.of` and `KeyFlowOf.apply` did change signature: 3bd12ad widened `Monad` to `MonadThrow` on all of them, and 1bc9fc6 added `Ref.Make` to the two `storage`-taking `of` overloads, which had neither before - so the build.sbt comment's "the constraint is the only change" understated it by a whole constraint on two of the five methods. Source-compatible either way, since every effect these are used with satisfies both, but not binary-compatible, and that is what the check is for. So the intention is `Compatibility.None` for this release and the two filters go. The repo's precedent is to set it in build.sbt for the release and revert right after (8081ea0, then ea43339 nine minutes later), which the comment now records. The two remaining filters stay: `Snapshots.apply` and `GroupCommit.this` are package-private and a nested class' constructor, members no caller can reach, and neither is a break to declare - they are noise MiMa emits from the bytecode, not from the API.
tobiajo
force-pushed
the
tj/tolerate-generation-fence
branch
from
September 9, 2026 19:40
c40f8a8 to
79ab793
Compare
tobiajo
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
In transactional snapshot mode (
cachingTransactional) a stale-generation rejection of the offset commit (ILLEGAL_GENERATION, surfaced asCommitFailedException) failed the flow, on both the persist path and the periodic commit path. The rejection already is the fence: the transaction aborted, nothing landed, the producer stays usable after the abort, and the consumer refreshes its generation once it completes the rebalance. Failing the flow only adds a leave and a rejoin, each of which bumps the generation and fences the peers' in-flight transactions; underCooperativeStickyAssignorretained partitions keep committing while a rebalance completes, so a rolling deploy becomes a restart cascade.KafkaSnapshotWriteDatabase.GroupCommit.commitBatchclassifies the rejection once, as the newGenerationFencedError(acoretype;corestill imports nothing from kafka-clients).persistPeriodically,persistPeriodicallyAndUnloadOrphaned,AdditionalStatePersist): warn, the key stays dirty and keeps holding its offset, regardless ofignorePersistErrors. The timer flow retries on its next tick; an additional persist leaves the state to the next persist.unloadOrphanedand the revoke-time flush are unchanged.persistPeriodicallyAndUnloadOrphanedmust not unload a key whose persist was tolerated, so this PR also gates the unload on the persist outcome; that is the same change as Unload a key only after its state was persisted #934, which fixes it forignorePersistErrorson its own. Whichever merges second rebases trivially.PartitionFlow):committedOffsetadvances only after a successful schedule; a fenced commit is warned and the offset is scheduled again on the next tick, or committed sooner by the next snapshot write.UNKNOWN_MEMBER_ID) surfaces as the same exception and is tolerated too: nothing lands, and the consumer's next rejoin reports it asonPartitionsLost, which tears the flows down.Fixes #938.