Prove the transactional fence holds on the error-path teardown route (#732)

[tobiajo]

What this is

One new IT spec, tests only — no src/main and no docs changes. Split out of #25 (the
ambiguous-commit half is now #26).

It proves the transactional Kafka persistence fences a stale flush on the error-path teardown
route, through a live rebalance and the untouched production machinery (KafkaFlow +
kafkaEagerRecovery + KafkaPersistenceModuleOf.cachingTransactional).

Why this route

The existing evolution-gaming#732 prevention tests in TransactionalKafkaPersistenceSpec drive the revoke
callback, with two PartitionFlows over one partition and a fabricated stale generation. This is a
different path through kafka-flow, exercised for real:

1. Member A's poll loop stalls in a fold past max.poll.interval.ms. A real broker evicts it and
   hands the partition to member B while A still holds buffered, never-persisted state.
2. B recovers, re-folds from the committed offset, processes further records, persists a fresher
   snapshot and commits a newer input offset.
3. A's stall ends with an error escaping the poll loop. The stream tears down, and TopicFlow's
   release drops every cached PartitionFlow — not just the revoked ones, because teardown has
   no way to know a partition was already taken away.
4. Releasing A's partition flow releases its key flows, and flushOnRevoke flushes A's stale state.
5. The write path decides the outcome — here, a transaction the broker rejects.

So transactional safety on this route was previously inference. It no longer is.

What it asserts

• The fresher snapshot survives (12345), paired with the offset the new owner committed (6).
• The flush was attempted and rejected — observed through A's own snapshot writer, which is
  wrapped to record the outcome of every broker-answered transactional call. Without this, a passing
  prevention assertion could equally mean no flush ever happened.
• Which fence fires: the producer epoch, not the generation. Under the partition's shared stable
  transactional.id, the new owner's initTransactions has already bumped the epoch, so the broker
  rejects the flush (InvalidProducerEpochException, or ProducerFencedException depending on
  transaction protocol version) before it reaches the generation-gated offset commit — which would
  raise CommitFailedException instead. Asserted where it is observed, not inferred from a log line.
• The next owner's own durable outcome, which states the invariant positively: it recovers 12345 at
  offset 6 and carries it forward as 123456 at offset 7 — every record up to the offset it
  commits is in the snapshot it persists.

A's teardown offset commit does not clobber B's either (rejected as a kicked member, swallowed in
TopicFlow.commitPending), so on the unfenced write path the surviving pair would be a stale
snapshot at a newer offset: silent loss. That is what the fence prevents here.

What was dropped since the first push

The original version also ran the same scenario against caching and asserted the corrupted outcome.
That is removed: the non-transactional exposure is known, documented (docs/persistence.md) and
already reproduced by TransactionalKafkaPersistenceSpec's "issue evolution-gaming#732 reproduction", so a third
assertion of it bought nothing, cost ~100 s of the suite's 145 s, and left behind a test that fails if
the default mode is ever improved.

Dropping it removed the contrast that showed the scenario was live, which is why the
attempted-and-rejected assertion above replaced it — a direct check instead of a costly implicit one.

Determinism

The post-teardown observation is a single read behind an explicit release barrier, not an eventually
poll: the dying member is allocated, so its release — the flush under test — runs at a controlled
point, and the write path awaits the broker's ack. Once release returns, an accepted write is already
durable and visible. No settling window that could pass merely because the flush had not happened yet.

The next-owner wait is a liveness gate only (committed past record 6); the exact offset is asserted,
not gated on, so a wrong value fails instead of timing out.

Verification

• sbt -batch scalafmtCheckAll — clean.
• sbt -batch "persistence-kafka-it-tests/testOnly *TeardownFlushFencingSpec" — green against a real
  broker, 2 runs, ~45 s (down from ~145 s).
• Mutation-checked the anti-vacuity guard: with flushOnRevoke = false on the dying member, every
  snapshot and offset assertion still holds and the test fails on the vacuity check (member A never attempted a snapshot write, so the fence was never exercised) — confirming the guard catches
  exactly the case that would otherwise pass while proving nothing.

Follow-up, not attempted here

The route exposes something narrowly fixable, independently of the write path: teardown flushes every
cached partition (TopicFlow's Resource.make release) where the revocation path
(TopicFlow.remove) knows which partitions are gone. Noted in the spec's scaladoc; worth its own
issue.

Test plan

• scalafmtCheckAll clean
• Spec green against a real broker (2 runs)
• Anti-vacuity guard mutation-checked
• Confirmed zero src/main and zero docs changes in the diff

[tobiajo]

Closing: the finding is recorded in the verification corpus instead, which is the better home for it.

The reasoning, for the record. The marginal regression value over the retained pins is thin — the mechanism this asserts (epoch-fencing a stale writer's flush) is already pinned by the revoke-route test, and the teardown-specific edges (seeded first flush, the group-metadata None window) have their own tests. Break either fence and those fail too. Worse, the anti-vacuity guard added in the last push would fail the day teardown is fixed to skip reassigned partitions — the same defect as the 'must be inverted when fixed' framing this PR started with, one layer down, and not repairable without giving the guard up.

What the work actually produced was a finding, not an artifact:

• the teardown route is reachable through a real broker eviction — the only place in the suite where the handover is a real reassignment rather than a fabricated generation or epoch;
• on that route the epoch net pre-empts the generation net, measured: the evicted member's writer recorded exactly one broker-answered call, send-ack → InvalidProducerEpochException, and no sendOffsetsToTransaction, so member validation never ran. The ordering itself was already source-pinned in the corpus (ext(K5)); what is new is corroboration on the eviction route and the sharpening that, for any batch carrying a write, member validation there is unreachable rather than merely later.

Both are now in the corpus on the verification branch: a new experiment section plus claim KF17, and the eviction-path clause that named UNKNOWN_MEMBER_ID as the closer is annotated rather than left to drift. The branch tj/characterize-teardown-flush stays on the fork so the harness is re-runnable — worth doing before the transactional mode loses its EXPERIMENTAL label.

Follow-up still open and unattempted: teardown flushes every cached partition where TopicFlow.remove knows which are gone.