feat(streams) astubbs#255: carry PC task ownership through Streams' task lifecycle and rebalance - #394
Draft
astubbs wants to merge 6 commits into
Draft
Conversation
…ebalance and revive The dispatcher was built to live exactly as long as one StreamTask constructor call. Kafka Streams does not treat a task that way: it suspends it, recycles it into a standby, gains and loses its partitions in a cooperative rebalance, closes it dirty and revives the same object. Five holes, each of them silent. - updateInputPartitions never reached PC, so a partition gained by a cooperative rebalance had no assignment epoch and EpochAndRecordsMap dropped every one of its records. Zero registered, no exception, a topology that just looks idle. Revocation is equally load-bearing: it is the epoch bump that lets an outcome arriving for a partition somebody else now owns be recognised as stale and dropped rather than advancing that owner's frontier. In-flight work on a revoked partition is abandoned behind that fence rather than drained or waited on - which is the at-least-once trade parallel-consumer-core already makes, adopted rather than a third policy invented here. - prepareRecycle tore the task down without routing through close(boolean), leaking four things per recycle: the static ACTIVE registry entry, the worker pool, the PcWorkSignal registration, and the WorkManager's partition state. Dormant only because no test configures standby replicas. - revive() threw rather than rebuilding. closeDirtyAndRevive resurrects the same instance, whose dispatcher was closed on the way down, so the rung below this one refused revival outright as a loud-failure floor. That floor is now gone: a revived task builds a new dispatcher over the partitions it holds at that moment, on the thread doing the reviving. Nothing is stranded by the swap - the closed dispatcher had already drained and revoked, and no offset it never committed was ever reported to Kafka, so the revived task re-reads exactly what a stock revived task would. Null-ness may still not change: a mid-run flip of the switch would silently move a live task between record paths, and that throws. - The owner-thread guard bound at construction, so a task handed to another thread would have thrown IllegalStateException on a legitimate call. It binds at hand-off now, and the wake signal moves with it - moving the guard alone would trade a loud exception for a silent stall. NOTHING IN PRODUCTION CALLS THE REBIND: Kafka 3.9.2 closes and rebuilds a reassigned task rather than handing it across threads. This is capability with an unstated assumption removed, not a fix, and the javadoc and the inflight note both say so, because the first draft of this work claimed otherwise and four independent reviewers caught it. - A clean close could discard live work. pcAwareCommitNeeded asked "is there COMPLETED work no commit has covered". On the stock path that is the same question as "is there uncommitted work", because processing is synchronous and there is no third state. Asynchronous dispatch creates one, and validateClean() is the caller that must see it: a clean close while records were still inside the processor chain succeeded silently, where Kafka's contract is to throw TaskMigratedException so the TaskManager closes dirty instead. hasUncommittedWork is that second question, and it stays a genuine query - counters and an AtomicInteger, touching neither the WorkManager nor the completion mailbox - because the state updater reaches it from its own thread through maybeCheckpoint. A query that drained the mailbox "just to be accurate" is how a plain field read became a cross-thread write; the rule that a question may not mutate is what keeps the two surfaces apart, and the class javadoc now splits them into guard-enforced and convention-only rather than claiming enforcement that three of the methods do not have. Deliberately NOT counted as uncommitted work: records PC is still holding. With retries disabled a failed record blocks its KEY shard forever and the records behind it stay available in PC's accounting, so that definition would make one poison pill enough to keep every future clean close throwing. close() now also marks everything it published as covered, because a closed dispatcher owns nothing and will never commit again. Kafka's own shouldClearCommitStatusesInCloseDirty is what catches the omission. Seam ON, against Apache Kafka's own suite, before and after this change and nothing else varied: nine cases go green and none regress. Three of them are shouldRecoverFromInvalidOffsetExceptionOnRestoreAndFinishRestore, which is the revival path, and no exception leaves a StreamThread uncaught any more where three did. The other five are StreamTaskTest close and checkpoint cases, which went green with validateClean learning to see running work. Seam OFF is unchanged, per class. The new integration arm is the module's first rebalance coverage: two KafkaStreams instances in one application id over a multi-partition topic, the cooperative assignor, the second joining mid-run, both dispatching through PC. Its four assertions were chosen before the harness - no loss, duplicates bounded by CAPACITY rather than by a fraction of throughput, a handover that demonstrably happened, and ownership that moved rather than being shared. The handover reader is assign+seek'd past a position captured at the instant the second instance starts, so it is structurally incapable of being satisfied by the first instance's earlier output rather than relying on the assertion staying careful. Both claims were sabotaged with the prediction stated first, and both reverts verified byte-identical by checksum. - The mailbox handoff to the WorkManager removed from drainCompletions. Predicted: the rebalance arm's no-loss assertion still passes, because output reaches the topic from the worker regardless of PC's accounting, while the capacity bound fails, because the frontier never advances and the joining instance re-reads what the first already produced. Exactly that - 120 duplicates against a bound of 76 - and at unit level only the four completion-dependent dispatcher cases went red out of thirty-one. - The close-time commit-state publication removed. Predicted: exactly one unit case red, on its last assertion, with its four resource-leak assertions still passing; and seam on, exactly one new upstream red, shouldClearCommitStatusesInCloseDirty. Exactly that, and nothing else moved in either direction. The patch was regenerated by the prescribed procedure - process-sources, edit under target/kafka-patched, regen-patch.sh with no maven run in between - never hand-edited, and verified by CONTENT rather than by hunk count: every added and removed line of the parent's patch survives except the ones this commit deliberately rewrites, which are the revive body and javadoc, the pcDispatcher field declaration, and pcAwareCommitNeeded.
…surface still uncovered CONCEPTS.md gains owner thread and completion mailbox. Both are now load-bearing across a class javadoc, a README section and a solutions write-up already on master, and nothing said what they mean. Owner thread is written as the general case with the control loop as its standalone instance, because the whole point of the Streams work is that the owner is somebody else's thread; and the entry carries the rule that follows from it - ownership is enforced by refusing foreign calls, so anything a foreign thread must legitimately ask has to be answerable without touching owned state, which is why a question may never mutate. The completion mailbox entry records the distinction that is easy to read as an implementation detail and is not: counting outstanding work at the drain rather than at publication is thread-safe and wrong, because it reports nothing outstanding for work that has genuinely finished and is merely waiting to be drained. The inflight note is about the rebalance surface, which is larger than the part now covered. Three things in it are worth more than the list: - The zero-duplicate result is not yet evidence. The test permits a capacity-derived bound and has measured zero every run. The likelier reading is that the rebalance is not landing where duplicates arise - the topology has no per-record cost, so there may be nothing in flight when partitions are revoked. A test that never approaches its own bound is not exercising the thing it bounds: the bound is unfalsified, not satisfied. The discriminator is a control arm with real per-record cost. - shouldThrowExceptionOnCloseCleanError fails on purpose and someone will eventually try to close it. Making it green requires acknowledging the commit in postCommit, which is a silent-data-loss defect rather than a fix: Kafka reaches postCommit after a swallowed commit failure and with no commit attempted at all. - Two write-ups from the branch forest are named with the command to read them rather than carried. Porting them would mean repairing their own citations into plan documents that are equally unmerged, and #312 recovered part of that learnings set without them. Naming them keeps the pointer rather than losing it.
…nt names the third reason Three rungs, three reasons, each found by measuring rather than by review, and each uncovered by the measurement that closed the one before it. The chain is recorded at the same three sites the last rung used - PcDispatchSwitch, which owns the decision; the patched StreamTask, where a reader arrives from a stack trace; and the pom's module description - plus the README section a user reads first. Reason one was a missing refusal, closed by #389. Reason two was revival, closed by this branch: revive() rebuilt rather than threw, and the seam-on before/after shows the three shouldRecoverFromInvalidOffsetExceptionOnRestoreAndFinishRestore cases green, nothing leaving a StreamThread uncaught, and no regression. Reason three is why the default does not move. A TaskCorruptedException or TaskMigratedException raised INSIDE a processor is caught by the worker, delivered one or more pump cycles later, and wrapped in a StreamsException - so Kafka's TaskManager never sees the type it dispatches recovery on, and an application stock Streams would have recovered shuts down instead. StreamThreadTest.shouldReinitializeRevivedTasksInAnyState fails for exactly that, on every parameter, IDENTICALLY BEFORE AND AFTER this branch. That control arm is what separates it from the revival defect the same test class was also reporting: the lifecycle work neither caused it nor fixed it. It is the same shape as the reason just closed - a recoverable event turned fatal - arriving by the other route, and it cannot be refused, because it is a property of the exception rather than of the topology shape. There is nothing to inspect at build time or at task construction, so extending the refusal envelope cannot reach it however far it goes. Two candidate fixes are recorded with what each does not buy. Rethrowing the control-flow types unwrapped is cheap and right on its own terms but does not fix the timing, so no test can currently prove it; surfacing a worker failure before the pump returns fixes both halves and changes the dispatch loop's shape. Both belong to the error-surfacing work, alongside #271's open review thread about a worker's failure being committed past - the same asynchrony seen from the commit side rather than the recovery side. Stream-time punctuation is a separate outstanding item and is not what this decision turns on. It was already priced in when the refusal reason was closed; the README and the note both say so, so that a later reader does not merge the two.
Dependency Review✅ No vulnerabilities or license issues or OpenSSF Scorecard issues found.Scanned FilesNone |
5 tasks
✅ Duplicate Code ReportTwo engines run in parallel for cross-validation. Each has its own thresholds tuned to its baseline - the real safety net is the per-engine "max increase vs base" check. ✅ PMD CPD
No new clones introduced by this PR. ✅ jscpd (language-agnostic)
|
🧪🔒 Quarantine Lane Report
🔴 expected while the owner PR is open · 🟡🎲 flapper, pass proves nothing · 🚨 a deterministic quarantined test passing means its fix landed: delete its |
…ngs already use The area prefix names an area, and this module's existing notes - streams-dispatch-switch-is-jvm-wide-not-per-instance.md and streams-stream-time-punctuation-is-unsupported-and-not-refused.md - already establish "streams-" for it. The new note is the same shape as the second of those: a construct that is unsupported on the dispatch path and cannot be refused, only recorded. "core-" names the engine, which this is not. Nothing cites it yet, so the rename costs no citations.
|
…pe the cache-warming job The Unit lane went red on this branch with the streams module FAILURE and every other module SUCCESS - and no test had run. It died in generate-sources, at dependency:unpack, with a Maven Central read timeout on kafka-streams:jar:sources:3.9.2. That is the misdirection worth recording: the lane is called "Unit Tests", so a reader goes looking for a failing test and finds none. The build never reached one. prepare-deps warms the cache with dependency:go-offline, which resolves the declared dependency graph. This module additionally fetches Apache Kafka's sources and test-sources classifiers by explicit artifactItems in two unpack executions, and those are not in that graph - so they are downloaded live from Central on every run of this module, in exactly the phase where the region-dependent timeout class already written up in docs/solutions/ bites. That write-up's own finding is why this needs a fix rather than a re-run: the runner is often reassigned to the same region, so re-running is unreliable, and what actually worked there was pre-warming so nothing is fetched from Central during the build. This module is the one place left outside that fix. Recorded rather than fixed here: warming a classifier artifact is a workflow change with its own verification (the unpack executions must log a cache hit rather than a download), and it is not this branch's subject.
astubbs
added a commit
that referenced
this pull request
Aug 31, 2026
The streams module gets Kafka's `sources` and `test-sources` classifier jars through two `dependency:unpack` executions that name them as `<artifactItems>`. `prepare-deps` warms the cache with `dependency:go-offline`, which resolves the declared dependency graph - and an artifactItem is not in that graph, nor is either jar declared anywhere else. So both stayed cold, and every lane that built this module fetched them from Maven Central inside `generate-sources`. That put them on the CDN route lottery recorded in docs/solutions/build-errors/maven-central-timeout-azure-west-regions-2026-04-21.md: an exactly-240s read timeout, and re-running does not reliably help because the runner is often reassigned to the same region. It struck #394 and #395 - Unit and Integration both red at `unpack (unpack-kafka-streams-sources)` with zero tests run, so the lane named "Unit Tests" pointed at a subsystem that never reached compilation. Same plugin-resolves-for-itself class the proxy stack already warms three times over (protoc, scala-maven-plugin, the SpotBugs detector plugins), reached from a new direction: here the plugin resolving for itself is maven-dependency-plugin. The step follows that established pattern - version derived from the pom property rather than a literal, a non-version help:evaluate answer is a hard error naming the property, and the jars are asserted on disk afterwards. Rejected alternative: warming by building the module, which is what the Connect spike branch does. It works, but costs a full build of core inside this job, and that branch records why it cannot be shortened to a cheap phase-only walk - the module's test-scope resolution wants core's tests-classifier jar, which does not exist until core reaches `package`. Two jars are the whole gap and `kafka.version` is a real root-pom property, so naming them is cheaper and no more prone to rot. Verified by extracting the step's script from the workflow and running it: - Empty local repository, script otherwise verbatim: exit 0, and both jars arrive at exactly the paths the assertions name, byte-identical to the populated repository's copies. - Property renamed to `kafka.verzion`, one term changed and nothing else: exit 1, `::error::kafka.verzion did not resolve to a version from the root pom (got 'null object or invalid expression')`, and nothing fetched. The guard fires and names the property. - On-disk assertion is independently load-bearing: a run where the `get` goal succeeded but the asserted path was elsewhere exited 1 on the `test -s`, which is the case the exit code alone would have called a pass. - Sufficiency: after the warm, `-pl parallel-consumer-streams -am generate-test-sources` runs OFFLINE from a cleaned target and both unpack executions succeed, producing the unpacked sources. Offline makes any Central fetch a hard failure, so this shows the warmed jars satisfy the artifactItems resolve rather than merely sitting near it. `kafka.version` is the default every live lane builds at. The one job that overrides it, `test-kafka-compat`, is `if: false`; the comment says so, so re-enabling it comes with the note that its Kafka version falls outside this warm.
5 tasks
astubbs
added a commit
that referenced
this pull request
Sep 1, 2026
…chinery rung Merges `feats/ks-streams-fork-machinery` (#379) into this rung, so the whole stack above it stops paying the Maven Central coin flip. Two commits arrive, and only one of them is code: - `ec432b439` teaches `prepare-deps` to warm Kafka's `sources` and `test-sources` classifier jars. This module fetches them through explicit `<artifactItems>`, which `dependency:go-offline` does not resolve, so every lane that built the module downloaded them live from Central inside `generate-sources` - the exact phase the region-dependent 240s read timeout bites. It struck #394 and #395 with Unit and Integration both red at `unpack (unpack-kafka-streams-sources)` and zero tests run. - `311f0b797` moves the ledger note for that gap onto the branch that owns the mechanism. Ordinary merge, no history rewritten. The merge was clean - the workflow hunk and the note are both additions this branch did not have. `bin/check-all.sh` exits 0 in this worktree before and after, all 15 runnable gates passing, so the merge is not-worse by measurement rather than by assumption. The repo's pre-commit gate evaluates a different worktree (#382), so this commit uses `--no-verify`.
…cycle rung Merges `feats/ks-streams-refusal-envelope` (#389), which has just taken the two rungs below it, into this one. This closes the whole spine's Maven Central coin flip: every lane that builds `parallel-consumer-streams` from here up now finds Kafka's `sources` and `test-sources` classifier jars in the warmed cache instead of fetching them live inside `generate-sources`. One real conflict, and the incoming file told this branch how to resolve it. `docs/inflight/ci-streams-classifier-artifacts-escape-the-cache-warming-job.md` exists on both sides: this branch wrote it when the defect was open and it still calls the warm a *candidate* fix, while #379 rewrote it once the fix landed - and left an instruction in its own `post-merge` block naming this branch, this path, and which version to take. The post-fix version wins, as instructed. Nothing is lost by taking a side here: the superseded half describes the same defect as unfixed, and the workflow step's own comment is now the durable owner of the reasoning. `docs/inflight/test-streamthreadtest-invalid-timestamps-flake.md` auto-merged, gaining the example rung's two CI sightings; the diff against this branch's pre-merge version is that paragraph and nothing else. `bin/check-all.sh` exits 0 in this worktree before and after, 15 runnable gates passing. The repo's pre-commit gate evaluates a different worktree (#382), so this commit uses `--no-verify`.
astubbs
added a commit
that referenced
this pull request
Sep 1, 2026
Merges `feats/ks-streams-reconciled` - the reconciliation of #395, #396, #398 and #391 on top of the refreshed #394 - into this branch, so that #271 can be retargeted onto that rung and its displayed diff collapse to the residue. This is the plan's "retarget move", `docs/plans/2026-08-31-001-process-god-branch-decomposition-plan.md`, and it is an ordinary merge: no history was rewritten and nothing was force-pushed. The pre-merge tip is preserved as `origin/backup/pre-stack-merge-271`. The merge also brings `master` forward, because the stack is current with it and this branch was not. That is most of the volume and all but a handful of the conflicts. ## Which side won, and the check that made it safe to say so The rule was: the rungs' refined versions win wherever both sides have a file; the spike's residue is kept. **The second half of that is only safe if the spike contributed nothing to the shared files, and that was measured rather than assumed.** Both trees were exported with package paths normalised (`io.confluent.parallelconsumer` and `bz.stub.parallelconsumer` folded to one token, and `io.confluent.csid.utils` to its new home) and compared file by file against the branch's own merge base. The answer is clean: outside `parallel-consumer-streams` and the streams example, **every difference this branch carries is the package rename itself plus a copyright-header line** - the same rename `master` performed independently, which is why taking the rungs' side loses nothing. The one substantive exception is the streams example's stock-baseline fixtures, which are already on the stack because #398 depends on them. - **Files on both sides: the rungs' version, 93 of them.** They carry current master, the rename, and every fix the five rungs landed. - **Files only on the spike: kept, 32 of them.** The plan and result documents, the branch handover, the ranked next-work notes, the eight `docs/solutions/` write-ups, and two streams tests the stack deliberately left behind - `HeadOfLineBlockingBenchmarkTest`, which #398 says it is leaving on the forest, and `ProcessorContextConfinementTest`. That residue is what this PR still exists to review. - **Files master deleted: they stay deleted, 12 of them** - `.semaphore/`, `.travis-archived.yml`, `service.yml`, two `.idea` run configurations, the uppercase `docs/` names master lowercased, and the `InternalRuntimeException` pair master renamed to `PCInternalRuntimeException`. - **Two shared ledgers merged as unions, not resolved to a side**: `docs/inflight/pr-strategy-doc-merge-triggers.md` and `docs/inflight/release-0.6.0.0.md` keep this branch's paragraphs alongside master's. **The strongest check is the one that makes a mis-paired rename impossible to hide.** Rather than counting recorded renames, the merged tree was differenced against `feats/ks-streams-reconciled` directly: the only differences are the 32 additions and the two union-merged ledgers. Nothing else survived on the wrong side, in the wrong module, or under an `io/confluent` path - `git ls-files | grep io/confluent` is empty, and the prescribed `grep -rnE 'io[\./]*conflu'` sweep returns only prose in dated documents. **Both sides were already renamed, which is not what the plan predicted.** It expected the spike to be pre-rename; `origin/feats/ks-on-pc-spike` had in fact moved five commits past the worktree's copy and carries the rename. That is the good case AGENTS.md describes - the mis-pairing surfaces as rename/rename conflicts on the right files instead of silently applying one module's edit to another - and it did: three `TestConventionsArchTest.java` files were offered paired across modules, and the right file was taken into each module rather than accepting git's pairing. ## One collision the merge produced, and the reason it is a good one `HeadOfLineBlockingBenchmarkTest` carried a private `sleep(Duration)` helper. Since #395 hoisted the identical helper onto `BrokerStreamsIntegrationTest` - because two arms that simulate cost differently are not comparable - the two now collide, and the compiler said so rather than either winning silently. The local copy is deleted and the shared one inherited, which is what the hoist was for. ## The gates, and the one that is a scoping artefact `bin/check-all.sh` was run in this worktree. Four gates failed on first pass and **three of the four did not exist on this branch before the merge** - they arrived with master, and every finding was on content that predates them. - **`check-inflight-tags`** flagged sixteen notes with no `inflight-type`, all of them this branch's own and all older than the tag scheme. Tagged - a first pass a reviewer should correct where it reads wrong. `bug-core-tests-jar-junit-parallelism-leak.md` additionally carries a `closed` state, because #265 removed the file it is about. - **`check-file-refs`** flagged citations broken by the rename - the exact class [`docs/citations.md`](docs/citations.md) says must be repaired rather than left. Forty-seven paths were re-pointed to their successors, **each verified to exist before the edit was written**; the claims around them are untouched, which is the line that document draws. What genuinely has no repo path - Apache Kafka's own sources inside the published jar, a placeholder in a shell recipe, a class that was only ever proposed - carries a line-scoped `file-refs: N/A` with its reason. Two targets that moved out of the tree point at the history holding them instead. - **`check-issue-refs`** flagged two bare `#NNN` in the handover; qualified. - **`check-branch-self-reference`** is the artefact, and the gate's own header names this exact situation: mid-merge the merge base is still the old one, so master's sentences read as this branch's. Twenty-five of the twenty-eight files it flagged are on `feats/ks-streams-reconciled` too and twenty-four are byte-identical to it, so marking them would be attesting to somebody else's notes - which that header explicitly says not to do. Committing the merge is the documented fix. ## Verified JDK 17, macOS. The whole reactor compiles, main and test. The reconciled branch it merges was verified in its own right before this: seam-off oracle unchanged, module unit suite green, and all 35 broker-backed integration arms green against a real broker.
astubbs
marked this pull request as draft
September 2, 2026 14:47
6 tasks
astubbs
added a commit
that referenced
this pull request
Sep 8, 2026
…r-surfacing rung Merges `feats/ks-streams-task-lifecycle` (#394), this PR's BASE, which has just taken #379's post-cut work. Six commits arrive; the only code among them is the `prepare-deps` warm for Kafka's `sources` and `test-sources` classifier jars, which closes this rung's share of the Maven Central coin flip - the failure this branch spent four commits recording sightings of. **This merges the base rung, NOT master, and that is a correction rather than a shortcut.** The task was framed as "merge origin/master so #395 stops reading CONFLICTING", but this PR's base is `feats/ks-streams-task-lifecycle`, not master - so GitHub's CONFLICTING is computed against that rung, and master cannot clear it. The evidence is direct: the sole conflicting path, `docs/inflight/ci-streams-classifier-artifacts-escape-the-cache-warming-job.md`, **does not exist on master at all**, so a master merge leaves the collision untouched while putting this rung 75 commits ahead of its own base and inflating a stacked PR's diff with work no rung below it carries. The whole spine still sits on `b2e6c190d`; changes travel DOWN it by merging the rung below, which is what the three merges arriving here did. One real conflict, and the incoming file left the instruction for it. - `docs/inflight/ci-streams-classifier-artifacts-escape-the-cache-warming-job.md` - resolved to the incoming POST-FIX version's structure, as #379 instructed in the note's own `post-merge` block: take the version that names the warm step, not the one that still calls the fix a candidate. Taken as a straight side, though, that instruction would have dropped four commits of evidence it was never written about: it names the *lifecycle* rung, whose copy was the pre-fix one unchanged, while THIS rung had since added its own sightings to it. So the branch's findings are kept, condensed into a new section written in post-merge terms and marked `post-merge: checked` - the second independent sighting that made this master-state rather than one PR's problem, that Unit and Integration fail together because they share the runner's route to Central, that a re-run is a coin flip rather than a fix, and the markdown-only control arm that settles the attribution. All three remain true for any branch that has not yet merged the warm forward, which is exactly the population the post-fix version's "What is still open" section addresses. Nothing either side wrote and still holds was dropped; what went is the superseded framing of the defect as unfixed, whose reasoning the workflow step's own comment now owns. One sentence was moved to the past tense rather than carried verbatim: the incoming copy says `feats/ks-streams-task-lifecycle` still holds the pre-fix file and will collide add/add. Both rungs that hit that collision have now resolved it as instructed, so as written it was a claim about NOW that this merge falsifies - the exact rot `bin/check-branch-self-reference.sh` exists to catch. The instruction is kept for any rung above these two that has not yet merged forward. `.github/workflows/maven.yml` and `docs/inflight/test-streamthreadtest-invalid-timestamps-flake.md` auto-merged as pure incoming additions - the warm step, and the example rung's two CI sightings. No decision this branch recorded is reversed by the incoming side. Verification. The merge changes **no Java and no pom** - `git diff --stat HEAD` against the pre-merge tip is the workflow file and two markdown notes - so no module's compilation or test outcome can move, and none was re-run on that basis rather than on assumption. `bin/check-all.sh`: 15 ran, 13 passed, 2 failed, and **both failures were reproduced on the pre-merge tip 228cebc in a detached worktree**, so neither is this merge's: - `check-file-refs.sh` - `docs/BUG_857_INVESTIGATION.md`, cited by three `docs/solutions/` write-ups, does not resolve on this branch. Identical before and after. - `check-quarantine-owners.sh` - `ProducerManagerTest.producedRecordsCantBeInTransactionWithoutItsOffsetDirect` names owner PR #262, which has merged since this branch was cut while the quarantine stayed. Live GitHub state, not tree state. `check-branch-self-reference.sh` and `check-issue-refs.sh` both pass, including the new section's `post-merge: checked` block. The repo's pre-commit gate evaluates a different worktree (#382), so this commit uses `--no-verify`; the gates above were run in this worktree instead. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018rEzrWYFr6oEzy6porczd3
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.
Serves #255.
depends on #389
Description
The claim this PR makes: PC-driven execution now survives Kafka Streams' task lifecycle - a task
that is suspended, recycled, given and taken partitions by a cooperative rebalance, closed, or
revived after a dirty close keeps a working dispatcher and a correct commit frontier across the
change of hands - and the seam-off behaviour-preservation oracle is unchanged.
The dispatcher was built to live exactly as long as one
StreamTaskconstructor call. Kafka Streamsdoes not treat a task that way. Five holes followed, and every one of them was silent.
updateInputPartitionsnever reached PCEpochAndRecordsMapdropped every one of its records - zero registered, no exception, a topology that just looks idleprepareRecyclebypassedclose(boolean)ACTIVEregistry entry, the worker pool, thePcWorkSignalregistration, and the WorkManager's partition staterevive()threwcloseDirtyAndReviveresurrects the same instance, whose dispatcher went down with it - the rung below refused revival outright as a loud-failure floorIllegalStateExceptionon a legitimate callvalidateClean()asked about completed work only, socloseClean()over records still inside the processor chain succeeded silentlyThe one that is not a synonym
pcAwareCommitNeededasked "is there COMPLETED work no commit has covered". On the stock path that isthe same question as "is it safe to walk away", because processing is synchronous and a record is
either finished or not started - there is no third state. Asynchronous dispatch creates one, and
validateClean()is exactly the caller that must see it: Kafka's contract there is to throwTaskMigratedExceptionso theTaskManagercloses the task dirty instead.hasUncommittedWorkis that second question and stays a genuine query - counters and anAtomicInteger, touching neither theWorkManagernor the completion mailbox - because Kafka'sDefaultStateUpdaterreaches it from its own thread throughmaybeCheckpoint. A query that drainedthe mailbox "just to be accurate" is how a plain field read became a cross-thread write. The class
javadoc now splits the surface into guard-enforced and convention-only rather than claiming
enforcement that three of its methods do not have.
Deliberately not counted as uncommitted work: records PC is still holding. With retries disabled a
failed record blocks its KEY shard permanently and the records behind it stay available in PC's
accounting, so that definition would make one poison pill enough to keep every future clean close
throwing.
Revocation abandons in-flight work rather than draining it
Revoking is what bumps the partition-assignment epoch, and the epoch is the whole safety mechanism: an
outcome arriving for a partition revoked while its record was in flight is recognised as stale and
dropped instead of advancing a frontier the new owner is now responsible for. That is the at-least-once
trade
parallel-consumer-corealready makes, adopted rather than a third policy invented here.bindToCurrentThread()has NO production caller, and the docs say soThe guard can now follow a task to a new thread, and the wake signal moves with it - moving the guard
alone would trade a loud exception for a silent stall. But in Kafka 3.9.2 a reassigned task is closed
and rebuilt rather than handed across threads, so the constructor's bind is the only one that happens.
It is capability with an unstated assumption removed, not a fix. The first draft of this work claimed
otherwise and four independently dispatched reviewers all reached the same finding; the javadoc and the
docs/inflight/note now both state it plainly.The default decision, which this rung owns - and it stays OFF
#389 named exactly one trigger for this rung: with the seam on, Kafka's
ordinary task-corruption recovery reached
StreamTask.revive(), whose throw left the run loopuncaught on the StreamThread. That trigger is closed, measured with a control arm - same
experiment, this branch's lifecycle work the one changed term, seam-on run before and after on the
same machine:
shouldRecoverFromInvalidOffsetExceptionOnRestoreAndFinishRestore- the revival path, on everyparameter. The other five are
StreamTaskTestclose and checkpoint cases, green becausevalidateCleanlearned to see running work, which is a second mechanism confirming itself.revival instead, logged by name.
And the default still does not move, because the same measurement named a third reason.
StreamThreadTest.shouldReinitializeRevivedTasksInAnyStatefails on every parameter, identicallybefore and after - that control arm is what separates it from the revival defect the same test class
was also reporting. A
TaskCorruptedExceptionorTaskMigratedExceptionraised inside a processoris caught by the worker, delivered one or more pump cycles later, and wrapped in a
StreamsException,so Kafka's
TaskManagernever sees the type it dispatches recovery on and an application stock Streamswould have recovered shuts down instead.
It is the same shape as the reason just closed - a recoverable event turned fatal - arriving by the
other route, and it cannot be refused: it is a property of the exception rather than of the
topology shape, so there is nothing to inspect at build time or at task construction. It belongs to the
error-surfacing unit, alongside #271's open thread about a worker's failure
being committed past - the same asynchrony seen from the commit side rather than the recovery side.
The chain is recorded at the three sites the rung below used, re-pointed rather than left stale:
PcDispatchSwitchowns the decision; the patchedStreamTask'spcProcesscarries it where a readerarrives from a stack trace (the
revive()javadoc no longer can, because the throw is gone); and thepom's module description carries the one-line version. Stream-time punctuation is a separate
outstanding item that was already priced in when the refusal reason closed, and every site says so, so
a later reader does not merge the two.
The module's first rebalance coverage
RebalanceUnderPcDispatchTest: twoKafkaStreamsinstances in one application id over amulti-partition topic, the cooperative assignor, the second joining mid-run, both dispatching through
PC. Every other integration proof this module has is one partition, one task, one instance.
Its four assertions were chosen before the harness. No loss; duplicates bounded by CAPACITY, not by
a fraction of throughput - a percentage bound passes a run that reprocessed the lot; the handover
demonstrably happened; and ownership moved rather than being shared, asserted against the tighter
in-flight bound, because "keys processed by both
<=the duplicate bound" is arithmetically implied bythe duplicate assertion and cannot fail on its own. The handover reader is
assigned andseeked pasta position captured at the instant the second instance starts, so it is structurally incapable of being
satisfied by the first instance's earlier output.
Its zero-duplicate result is recorded as not-yet-evidence, in
docs/inflight/test-streams-rebalance-coverage-gaps.md: a test that never approaches its own bound isnot exercising the thing it bounds. The likelier reading is that the rebalance is not landing where
duplicates arise, and the discriminator named there is a control arm with real per-record cost.
Proven able to fail
Both sabotages had the prediction stated first, and both reverts were verified byte-identical by
checksum.
WorkManagerremoved fromdrainCompletions. Predicted: therebalance arm's no-loss assertion still passes, because output reaches the topic from the worker
regardless of PC's accounting, while the capacity bound fails, because the frontier never advances
and the joining instance re-reads what the first already produced. Exactly that - 120 duplicates
against a bound of 76 - and at unit level only the four completion-dependent dispatcher cases went
red out of thirty-one. The asymmetry is what says the right thing broke.
last assertion, with its four resource-leak assertions still passing; and seam on, exactly one new
upstream red,
shouldClearCommitStatusesInCloseDirty. Exactly that, and set-differencing theseam-on reds against the clean run showed one new failure and zero cases going green.
Verification
Seam-off oracle run before and after on this machine, whole
testphase, per-class numbers read out oftarget/surefire-reports-kafka-upstream/rather than copied: identical, unchanged, with the sameassumeTrue/assumeFalseskips. The parent's refusal tests and the seam tests are green, and theintegration lane is green including the new arm.
bin/check-all.shexits 0.Re-derive rather than copy - the counts move with the patch, the Kafka version and the seam - and never
scope that run with
-Dtest=, which silently overrides the execution's<includes>so the suite doesnot run at all and the build still goes green.
Three process notes worth recording.
XML from the previous run reads as a clean pass. The second sabotage was re-run with the report
directories deleted and
-Dmaven.test.failure.ignore=trueso the oracle actually executed, and thereading script asserts reports exist rather than treating an empty glob as success.
check-file-refs.shreported 25 dangling references, every one of them inbench/README.mdandbench/llingr/, paths that do not exist on this branch.bin/check-all.shwas run in this worktreefirst and passed every gate; these commits then used
--no-verify.generate-sources, atdependency:unpack, with a Maven Central read timeout onkafka-streams:jar:sources:3.9.2; everyother module passed and no test ran. Diagnosing it found something worth keeping rather than a
re-run:
prepare-depswarms the cache withdependency:go-offline, which resolves the declareddependency graph, while this module additionally fetches Kafka's
sourcesandtest-sourcesclassifiers by explicit
artifactItems- so those are downloaded live from Central on every run ofthis module, in exactly the phase the already-documented region-dependent timeout class bites.
Recorded in
docs/inflight/ci-streams-classifier-artifacts-escape-the-cache-warming-job.mdratherthan fixed here, because warming a classifier artifact is a workflow change with its own
verification.
Where this came from
Copied out of #271, the Kafka Streams feasibility study, which remains the
source god PR and keeps the semantic work, the measurements and the open review threads - including the
revive()thread this PR answers. The PC-side classes are renamedio.confluenttobz.stub; thepatched Kafka sources stay in
org.apache.kafka. The patch was regenerated by the prescribed procedureprocess-sources, edit undertarget/kafka-patched,bin/regen-patch.shwith no maven run inbetween - never hand-edited, and verified by content rather than by hunk count: every added and
removed line of feat(streams) astubbs#255: refuse the Kafka Streams surface PC dispatch cannot run safely #389's patch survives except the ones this rung deliberately
rewrites, and each of those is named in the commit body.
Two write-ups from the branch forest are named with the command to read them rather than carried:
the task-lifecycle-callbacks finding and the vacuous-restart-assertion one. Porting them would mean
repairing their own citations into plan documents that are equally unmerged, and
#312 recovered part of that learnings set without them.
Fourth rung of the reconstructed Wagon B stack. Not here, and named so a reader does not expect them:
stream time and punctuation, including
WALL_CLOCK_TIMEpunctuator effects never becomingcommit-covered; backpressure and error surfacing, which owns the third default reason above; the
seam-on upstream evidence lane; the benchmarks; the example module.
Checklist
CONCEPTS.md's owner thread and completion mailbox, the pom's module description,PcDispatchSwitch's javadoc chain, and threedocs/inflight/notes - the uncovered rebalancesurface, the exception-type gap the default now waits on, and the cache-warming gap this PR's
own CI red uncovered
docs/features/-N/A - the module is still unpublished, so there is no artifact to depend on and nothing for the ladder to point a user at. The roadmap stage is unchanged, for the same reason the three rungs below gavePcTaskDispatcherTestcases (the in-flight/commit-datadistinction, the poison-pill case that must NOT block a close, the owner rebind and its wake
signal, the foreign-thread query, the partition update and its epoch fence, the no-op rebalance,
the replacement dispatcher, and the four-way teardown contract) plus the new
RebalanceUnderPcDispatchTestintegration armce-simplifyandce-code-reviewlocally -N/A - asked for review on the PR instead. Most of the diff is a copy from a branch already reviewed on astubbs/parallel-consumer#271; what is new here - revival rebuilding the dispatcher, the default decision and its control arm, and the regenerated patch - is what a reviewer should spend attention on