Repository navigation
feat(streams) astubbs#255: advance stream time on the PC path, and decide what to do about punctuation - #396
Draft
astubbs wants to merge 9 commits into
Conversation
PartitionGroup.nextRecord() advances streamTime at selection, and the PC dispatch path never calls it - so partitionGroup.streamTime() sat at UNKNOWN for the life of every PC-dispatched task. Two things followed, both silently: STREAM_TIME punctuators never fired, and the PUBLIC ProcessorContext.currentStreamTimeMs() returned a constant -1. The second is the more likely of the two for a user to hit, because it is reachable from any Processor with no refused DSL call in sight. Stream time now comes from the dispatcher: the lowest stream timestamp still in flight, or the highest dispatched when nothing is, clamped monotone. The timestamp crosses the seam in the WorkPreparer's return value rather than being read off the ConsumerRecord, because the two are different numbers - a payload-time topology's TimestampExtractor points at a field inside the value. WHAT IT GUARANTEES IS NARROWER THAN A WATERMARK, and the class javadoc says so rather than being read as one. The mark never passes a record CURRENTLY IN FLIGHT, so a punctuation cannot close a window over work inside the chain. It is NOT an order boundary: the monotone clamp means a later record can carry a lower timestamp, and dispatchesBehindStreamTime counts exactly those. And it can sit ABOVE what stock would hold - three records on one partition with three distinct keys give [0, 2, 2] where stock reports [0, 1, 2], because PC hands records out per KEY shard. That is pinned as a characterisation test rather than described, because an earlier version of this work claimed "never ahead of stock", narrowed the only test that measured the counter-example so it would pass, and shipped a counter that could not have been non-zero if the claim were true. THREE HOLES THE SAME WORK RECORDED AGAINST ITSELF AND LEFT OPEN ARE CLOSED HERE, because each is a property of the mechanism this commit introduces rather than a separable fix: - close()'s drain could advance the mark over work a forced shutdown killed. awaitTermination expires, shutdownNow() interrupts the chains, each throws into recordFailure, and the very next statement is drainCompletions() - which released those holds and republished. A streamTimeFrozen flag is now raised at the top of BOTH close paths and checked in publishStreamTime, so the mark freezes at the last honest value it had. Pinned by aCleanCloseThatHasToForceThePoolDoesNotAdvanceTheMarkOverTheWorkItKilled, which is also the first coverage the clean-close-with-forced-pool branch has had. - A RejectedExecutionException from workerPool.execute leaked a hold. The hold and the inFlight increment both happen before the submit, and abortAllActive() races a pump by design - so an uncompensated hold pinned the mark at that record's timestamp forever, turning a wrong counter into a stopped punctuation clock. Compensated now, in the drain's own order. - PreparedRecord validated neither field. A preparer returning a null Runnable produced an NPE on a worker, which runOnWorker classifies as a PROCESSING failure - so with retries disabled it blocked that key's shard for the life of the task and reported a topology error. This is the module's public extension seam; it fails at its constructor now, naming the alternative. The seed also reaches pcRecordQueues, not only the dispatcher. RecordQueue passes its own partitionTime to timestampExtractor.extract(record, partitionTime), and the PC path's queues are created lazily at UNKNOWN - so a topology using Kafka's shipped UsePartitionTimeOnInvalidTimestamp threw on the first record after a restart with an invalid embedded timestamp, where stock recovered. That branch has no automated gate on this rung, because the only execution running Kafka's shouldReadCommittedStreamTime* cases pins the seam off; it is verified by a seam-on run recorded in the PR, and the gate belongs to the seam-on evidence lane. The refusal reasons that argued from stream time NOT advancing are corrected in the same pass - 28 @DonotCall and @deprecated strings a user reads as a compile error, plus the enum they mirror. Leaving them would have made this commit ship a config lie: the reason is no longer that the clock is stuck, it is that the clock is a low-water mark over concurrently dispatched work, so which records fall inside a window and which version of a table a record sees move with dispatch order. The windowedBy strings argue from observedStreamTime, a different field this work did not touch, and are left alone apart from one clause that read as if it named the task's value. Patch regenerated by the prescribed procedure - process-sources, edit under target/kafka-patched, bin/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 named above, and no removal was lost. Seam-off oracle run before and after on this machine, whole test phase, per-class numbers read out of target/surefire-reports-kafka-upstream/ rather than copied: identical, unchanged, with the same skips. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
…m already claimed it did The SLF4J-binding exclusion that keeps our own tests on logback and Kafka's fork on reload4j sits on the SUREFIRE plugin's configuration. Failsafe does not inherit it - so this module's own integration tests have been binding reload4j the whole time, while the comment beside the dependency says "our own tests keep logback (and logback-test.xml)" and src/test/resources/logback-test.xml sat unused in that lane. Nothing went red for it, because nothing in the integration lane asserted on a log line until PunctuatorWarningTest did. What that test got was a ClassCastException casting an org.slf4j.reload4j adapter to a logback Logger - which is the loud version of a failure that would otherwise have been an empty capture list and a wrong conclusion about a warning that had in fact fired. The Kafka-fork execution is surefire's and keeps reload4j, so LogCaptureAppender - the reason reload4j is a dependency at all - still works where it is used. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
… say why the one-liner was not taken THE HANDOVER'S TOP-RANKED DEFECT, decided rather than deferred. It reads: "WALL_CLOCK_TIME punctuators fire today with no warning, and their effects never become commit-covered", with hasUncommittedWork() || commitNeeded named as a one-line candidate that "changes commit cadence for every PC-path caller and needs its own evidence". THE CODE HALF OF THAT DECISION SHIPPED TWO COMMITS AGO, inside the regenerated pc-streams.patch: both punctuation types now warn once per task at registration, and WALL_CLOCK_TIME is the new one. This commit is the half that belongs at the site which owns the default's reasons, because that is where the next person deciding to flip it will read. THE EVIDENCE SHRANK THE SECOND HALF OF THE DEFECT TO ALMOST NOTHING, and the three classes that did it arrive in the next commit: - Effects are not lost. PunctuatorEffectSurvivalTest reads the output topic and the store's changelog BEFORE close(), with a five-minute-linger negative control arm that must find nothing and does - [] and 0, against 13 forwards and 13 changelog records in the measured arms. Punctuator forwards and store writes reach the broker on the producer's own schedule, before and independently of any commit. Non-EOS, which is the whole supported envelope here. - postCommit DOES run. PostCommitCheckpointGapTest was written to confirm "the PC path never checkpoints" and refuted it - 12,000 records checkpointed at changelog 11,526 against stock's 11,999 on this machine. What is left is an idle-window tail, bounded by the final commit round, with clean close checkpointing regardless. - Commit cadence cannot evidence the candidate at all. PunctuatorCommitCoverageTest shows commit-total pinned at zero on an IDLE PC task by TaskExecutor's second loop re-asking commitNeeded() after onCommitSuccess has just cleared it. SO THE ONE-LINER IS NOT TAKEN. What it still buys is a little less changelog replay after a crash landing in an idle window. What it costs is a commit-cadence change for every PC-path caller, and a TaskMigratedException on a clean close after a punctuate-only interval, because validateClean() reads the same answer. That is not a trade this evidence supports. Loud degradation is. Refusing punctuation outright was considered and rejected for a different reason: both types fire and produce correct output for the common shapes, so a refusal would break topologies that run today - and a punctuator is a call on the processor context rather than a shape any of the three refusal layers can inspect. Silence was the one outcome the handover forbade, and it is the one this removes. THE DEFAULT DOES NOT MOVE. This rung closed one of the three named triggers - PcDispatchSwitch's javadoc used to defer punctuation to docs/inflight/ as "a separate outstanding item, already priced in", and that sentence is replaced by what is now true: implemented, loud, and with two measured gaps. The exception-surfacing trigger is still open and lives on a sibling branch. A default moved on one rung's evidence while another rung holds an unclosed reason is a default nobody measured, so the flip is stated as a decision for after those rungs reconcile rather than left for a reader to infer. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ms that refuted three claims Five broker-backed classes, four of them reconstructed from a branch forest whose whole value is that its claims changed under measurement. They are carried with their retractions intact rather than tidied into results, because the retractions are what stop the next reader re-proposing what was already refuted. STREAM_TIME PUNCTUATORS RE-FIRE OVER COVERED EVENT TIME AFTER A RESTART. StreamTimePunctuatorRefireOnRestartTest. Run 1 climbs event time and closes cleanly; run 2 restarts the same application id with records BELOW run 1's base. Stock punctuates at the restored mark; the PC path punctuates 60s and 56s below run 1's BASE - measured here as run 2 firing at 1788212164856/1788212168856 against a base of 1788212224856. Asserted below the base and not merely below the highest, because a replayed run-1 record would satisfy the looser band and PC's own docs say completed records beyond the frontier are replayed rather than skipped. Cause is named in the patch's own seedStreamTime comment: a PC-written group's commit metadata is PC's frontier payload, not Streams' TopicPartitionMetadata, so the decode yields UNKNOWN and there is nothing to seed from. The PC arm asserts the defective behaviour, so the suite stays green and the defect stays visible; its message says how to invert it on fix. The control was built and run FIRST, and earned it twice. It failed once because run 2's "late" records were stamped 2,000ms after the epoch - Kafka's time-based retention keys on the record's own timestamp, so they were ~56 years old on arrival and eligible for deletion immediately. It failed again because the arm expected SILENCE from run 2, and a restored mark still punctuates once, at the restored value. Both failures were the fixture, and both would have become a wrong conclusion without a control that ran first. PUNCTUATOR EFFECTS REACH THE BROKER INDEPENDENTLY OF ANY COMMIT. PunctuatorEffectSurvivalTest, and the interesting part is what it stopped claiming. Its first version punctuated, aborted, CLOSED the client, and then read the topics - and streams.close() runs a clean shutdown whose producer.close() blocks until every buffered record is sent, so close delivered the effects and the test reported survival regardless. Settled by probe: linger.ms at five minutes and the old shape still passed. The read now happens BEFORE close and the linger probe is promoted to a permanent negative control that must find nothing - [] and 0 here, against 13 forwards and 13 changelog records in the measured arms. The javadoc states plainly what it is NOT: abortAllActive() is workerPool.shutdownNow() and nothing more, so it is not a process-death test. THE PC PATH CHECKPOINTS UNDER LOAD. PostCommitCheckpointGapTest was written to confirm that it never does and refuted it: 12,000 records, stock checkpoints at changelog 11,999 and the PC arm at 11,526. The step the reasoning missed is that hasUncommittedWork() goes true AGAIN between the commit and TaskExecutor's second loop whenever more records finish in that window, which under continuous load is almost always. What survives is a bounded idle-tail gap. THE COMMIT SENSOR IS BLIND, AND ONLY ON AN IDLE TASK. PunctuatorCommitCoverageTest. Stock 8 -> 19 over 30 punctuations; stock without a punctuator flat at 1; both PC arms committed at offset 5 with commit-total 0. The PC arms assert that CONJUNCTION rather than a lone zero, because a zero alone was unfalsifiable - and an earlier revision claimed the sensor was structurally incapable of rising on the PC path at all, which the checkpoint class above refuted. The claim is scoped to an idle task here, which is the condition all four arms construct. THE WARNING ITSELF IS GATED. PunctuatorWarningTest is new: two STREAM_TIME punctuators and one WALL_CLOCK_TIME punctuator on one task, asserting ONE warning of each kind - so the warn-once claim is tested rather than asserted - and a seam-off control asserting a stock task says nothing at all. That log line is the only channel a user has for a construct no refusal layer can see, and until now it had no coverage; deleting the wall-clock assertion puts the handover's ranked defect 1 back into silence. Every arm asserts its own dispatch-path membership rather than logging it: PC arms assert the seam dispatched, stock arms assert it did not. Without that, an arm that silently fell back to the other path still satisfies a zero-valued headline assertion and the whole comparison becomes two identical runs. BrokerStreamsIntegrationTest gains committedOffsetOrNull (a group reader that never subscribes, so it cannot rebalance the topology under test) and sleepThrough, and closes a leak on the startup-timeout path: the client is start()ed before the await, so a timeout never returns and no caller-side guard can close it. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
… wherever it was written down Three documents asserted the thing this rung falsified, and each said it in its own words - so none of them would have been found by grepping for the others. THE MODULE README's "What is still unsupported and NOT refused" said the punctuator "simply never fires: no exception, no warning, no output". Replaced by what is now true, split by punctuation type because the divergences differ, plus the concurrency clause that applies to both and is the one that can corrupt user state. A new "Stream time" section describes the low-water mark and - at greater length than the mechanism - what it does NOT guarantee, because reading it as a watermark is exactly what makes a windowed operator look safe behind it. CONCEPTS.md's "Refused construct" argued that most refusals share one underlying reason: "that clock does not advance on this project's dispatch path". It does now, so the shared reason is restated as the one that is actually true - stream time here is a different quantity, so which records fall inside a window and which version of a table a record reads move with dispatch order. Two entries are added: "In-flight low-water mark", which spends most of its words on what it is not, and "Punctuator", which is the construct no refusal layer can see. Both keep the directory's rule of standing alone - no file paths, no class names. The docs/inflight/ note "streams-stream-time-punctuation-is-unsupported-and-not-refused" is retired rather than rewritten into a FIXED narrative, which this directory's own rules forbid. Its durable half - why a punctuator escapes all three refusal layers - moves to the README, which owns the refusal surface. What is still open becomes a new note, because what remains is a different item: three measured divergences (the restart re-fire, the punctuate/process overlap, and skipped rather than merely delayed punctuations), two divergence measurements the evidence lane owes, and one claim about observedStreamTime that this work deliberately did not verify because it names a different field. Nothing cited the retired note by name, checked before removing it. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
… actually observe The name and the assertions claimed the stream-time hold; the assertions read the failure count and the in-flight count, neither of which the hold touches. Worse, the claim is not provable on this code at all: both close paths freeze the mark, and a fixed pool with an unbounded queue rejects nothing until it is shut down, so there is no reachable state where a leaked hold moves the mark. What IS observable, and is what the compensation is for, is the record left in limbo: before it, RejectedExecutionException left dispatchAvailable - and so the patched process() - on the StreamThread with the in-flight count already incremented for a record that would never run and never report. PC holds it forever, isQuiescent() cannot become true, and every close pays its full termination wait. Renamed and re-asserted to that, with the hold's unobservability stated rather than glossed. Found by writing the sabotage: with the compensation deleted, the old assertions still passed. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ually fail The sabotage refuted its own prediction, which is the result worth keeping. Deleting the freeze at the top of close() left this test GREEN. Why: close() forces the pool with shutdownNow(), and the interrupted chains do not reliably enqueue their failures before the drain two statements later has already polled an empty mailbox. The mark then stays put for a reason that has nothing to do with the flag. The advance the flag prevents is real - with two killed records, three of the four orderings empty the hold map and publish the highest timestamp ever dispatched over work that never completed - but it is a race, and a race is not something a unit test pins by hoping. So the test now asserts the field's actual claim rather than one scenario in which the claim happens to hold: a closed dispatcher's mark is frozen against ANYTHING that reaches it, and seedStreamTime() is the one route that publishes without a drain. That assertion goes red the moment the freeze is removed. The original assertion is kept, because the scenario is the one the defect was reported from, and is labelled with exactly what it is worth on its own. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
…tcher into the next one The low-water sabotage produced one red this rung did not predict - rebindingMovesTheWakeSignalWithTheOwner, whose precondition counts how many dispatchers are registered on the current thread. It was not a flake and not related to stream time: the sabotage made aCleanClose...OverTheWorkItKilled fail at its PRECONDITION, before the close() that would have deregistered its local dispatcher, and this class runs SAME_THREAD - so the next test read 2. A test that closes its own dispatcher has to do it in a finally, or a failure in it becomes a failure somewhere else. Reported, because a cascade that only appears under sabotage is exactly the kind of thing that gets filed as a flake later. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ueue seed Measured with a control arm rather than asserted: removing ONLY the pcRecordQueues half of the restart seed and re-running Kafka's own shouldReadCommittedStreamTime* cases with the seam on leaves them green, case for case. The dispatcher half and pcAwareStreamTime()'s PC branch ARE exercised by that run - the seam is verifiably on, two dispatchers constructed - but the line that lets a UsePartitionTimeOnInvalidTimestamp extractor recover after a restart is not, and Kafka's suite has no case for that extractor to borrow. Worth writing down because the run looks like coverage. Whoever builds the seam-on lane needs to add a case rather than assume the existing ones reach it. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Dependency Review✅ No vulnerabilities or license issues or OpenSSF Scorecard issues found.OpenSSF Scorecard
Scanned Files
|
✅ 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
|
| PR | Base | Change | |
|---|---|---|---|
| Clones | 100 | 91 | :face_with_monocle: +9 |
| Duplicated lines | 1390 | 1294 | :face_with_raised_eyebrow: +96 |
| Duplication | 1.06% | 0.99% | 🫤 +0.07% |
| Rule | Limit | Status |
|---|---|---|
| Max duplication | 2% | ✅ Pass (1.06%) |
| Max increase vs base | +0.1% | ✅ Pass (+0.07%) |
⚠️ 9 new clones introduced
- 8 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PunctuatorWarningTest.java:12<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/WakeOnWorkShutdownTest.java:8 - 11 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PunctuatorWarningTest.java:18<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/StreamTimePunctuatorRefireOnRestartTest.java:20 - 12 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PunctuatorEffectSurvivalTest.java:15<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/StreamTimePunctuatorRefireOnRestartTest.java:15 - 12 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PunctuatorEffectSurvivalTest.java:327<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PunctuatorWarningTest.java:221 - 15 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PunctuatorCommitCoverageTest.java:15<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PunctuatorWarningTest.java:15 - 12 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PunctuatorCommitCoverageTest.java:406<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PunctuatorWarningTest.java:218 - 12 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PostCommitCheckpointGapTest.java:1<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/WakeOnWorkShutdownTest.java:1 - 12 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PostCommitCheckpointGapTest.java:15<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PunctuatorEffectSurvivalTest.java:21 - 11 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PcDrivenStreamsDispatchTest.java:1<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/StreamTimePunctuatorRefireOnRestartTest.java:1
Powered by astubbs/duplicate-code-cross-check
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| } catch (IOException e) { | ||
| throw new IllegalStateException("Could not read checkpoint at " + checkpointFile, e); | ||
| } | ||
| } |
There was a problem hiding this comment.
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PunctuatorCommitCoverageTest.java:407
| log.info("Produced {} records into {}", INPUT_RECORDS, inputTopic); | ||
| } | ||
|
|
||
| private KafkaStreams startTopology(final String appId, final String inputTopic) { |
There was a problem hiding this comment.
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PunctuatorEffectSurvivalTest.java:340
| // deleted; it is pinned by theMarkOvertakesStockWhereDispatchOrderDiffers below. Pinning to a single | ||
| // key isolates the property under test: a pool of one, a queue of one, and the mark on the record | ||
| // actually running. | ||
| dispatcher.registerRecords(PARTITION, records(3, offset -> "the-only-key")); |
There was a problem hiding this comment.
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/PcTaskDispatcherTest.java:1316
| } catch (IOException e) { | ||
| throw new IllegalStateException("Could not read checkpoint at " + checkpointFile, e); | ||
| } | ||
| } |
There was a problem hiding this comment.
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PunctuatorEffectSurvivalTest.java:317, parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PunctuatorWarningTest.java:219
🧪🔒 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 |
|
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.
5 tasks done
astubbs
added a commit
that referenced
this pull request
Sep 1, 2026
… only beside its siblings The `Integration Tests` lane on #271 went red on `PostCommitCheckpointGapTest.pcPathAlsoRefreshesTheCheckpointUnderLoad`. Diagnosed rather than re-run, and the first reading was wrong. **The value is `-4`, which is a sentinel and not a small number** - the checkpoint file exists and its changelog offset was never populated. That is a different failure from a bound missed under load, and it does not move with more time, so "the CI runner is slow" was never available as an explanation. **A control arm ruled out the reconciliation, which was the hypothesis worth killing first**: a commit fence and a checkpoint expectation meeting for the first time is precisely what a reconciliation surfaces. The same class, run alone at #396's own tip in a detached worktree, fails 3 of 3 with the identical sentinel. The merge is not the cause. What is left is an ordering dependency: the class passes in a single-fork local suite run, where the siblings that ran before it warmed the JVM, and fails alone - 3 of 3 here, 3 of 3 on its own branch - and on CI at `forkCount=4`, where a forked class does not reliably get that neighbour. **That matters more than the red build.** The class's own assertion message says it REFUTES a sibling class's inference that `postCommit` never runs on the PC path. A refutation that holds only when another class ran first in the same JVM is weaker than the one written down, and the green suite is what hid it - the shape this module's handover names as "several tests here passed while proving nothing". Nothing was widened, shortened or re-run into green. The note names what to look for and says explicitly not to widen the bound, because the bound is not what is failing.
astubbs
marked this pull request as draft
September 2, 2026 14:48
6 tasks done
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 #394
Description
The claim this PR makes: stream time advances on the PC dispatch path, punctuation of both kinds
fires and says how it differs, and the divergences that remain are measured against seam-off controls
rather than described - with the seam-off behaviour-preservation oracle unchanged.
Before this rung,
PartitionGroup.nextRecord()was the only thing that advanced stream time, and thePC path never calls it. Two things followed, both silently:
STREAM_TIMEpunctuators never fired, andthe public
ProcessorContext.currentStreamTimeMs()returned a constant-1- reachable from anyprocessor with no refused DSL call in sight, and the more likely of the two for a user to hit.
Stream time is a low-water mark, and most of what is written down is what it does NOT promise
The dispatcher publishes the lowest stream timestamp still in flight, or the highest ever dispatched
when nothing is, clamped monotone. The timestamp is the one the configured
TimestampExtractorproduced, carried across the seam with the work rather than read off the
ConsumerRecord- apayload-time topology points its extractor at a field inside the value, so those are two different
numbers, and one test exists solely to tell them apart.
[0, 2, 2]where stock reports[0, 1, 2], because PC hands records out per KEY shard. Pinned as a characterisation test rather than narrowed away, which is what happened the first timeThe handover's ranked defect 1, decided rather than deferred
It reads: "
WALL_CLOCK_TIMEpunctuators fire today with no warning, and their effects never becomecommit-covered", with
hasUncommittedWork() || commitNeedednamed as a one-line candidate that"changes commit cadence for every PC-path caller and needs its own evidence".
The second half did not survive measurement, and the three classes that killed it are in this PR:
PunctuatorEffectSurvivalTestreads the output topic and the store'schangelog before
close(), with a five-minute-linger.msnegative control arm that must findnothing and does -
[]and0, against thirteen forwards and thirteen changelog records in themeasured arms. Punctuator forwards and store writes reach the broker on the producer's own schedule,
independently of any commit. That class also carries its own retraction: an earlier version read the
topics after
streams.close(), whose clean shutdown flushes the producer, so it reported survivalregardless - proven by setting
linger.msto five minutes and watching the old shape still pass.postCommitruns.PostCommitCheckpointGapTestwas written to confirm "the PC path nevercheckpoints" and refuted it: twelve thousand records, stock checkpoints at changelog offset
11,999 and the PC arm at 11,526. The step the reasoning missed is that
hasUncommittedWork()goestrue again between the commit and
TaskExecutor's second loop whenever more records finish in thatwindow. What survives is a bounded idle-tail gap.
PunctuatorCommitCoverageTest: stock goes8 → 19 over thirty punctuations, stock without a punctuator stays flat at 1, and both PC arms sit at
commit-total 0 with the input partition committed - because
TaskExecutor's second loop re-askscommitNeeded()afteronCommitSuccesshas just cleared it. The PC arms assert that conjunction,not a lone zero, because a lone zero was unfalsifiable. Its claim is scoped to an idle task, which
is a narrowing the checkpoint class above forced.
So the one-liner is not taken. What it still buys is a little less changelog replay after a crash
landing in an idle window. What it costs is a commit-cadence change for every PC-path caller and a
TaskMigratedExceptionon a clean close after a punctuate-only interval, becausevalidateClean()reads the same answer.
What is taken is the first half: loud degradation. Both punctuation types now warn once per task at
registration, and
WALL_CLOCK_TIMEis the new one. Refusing was considered and rejected - both typesfire and produce correct output for the common shapes, so a refusal would break topologies that run
today, and a punctuator is a call on the processor context rather than a shape any of the three refusal
layers can inspect. Silence was the one outcome the handover forbade.
The two messages differ because the divergences do.
WALL_CLOCK_TIME's opens by saying its firingtimes are unaffected, so a reader is not sent hunting a timing bug that is not there. Both carry the
clause that can corrupt user state rather than merely retime output, and it is why wall-clock warns at
all: the punctuator runs concurrently with records still inside the chain, which stock never does
for one task, and a plain
KeyValueStoreis supported here.The other confirmed divergence: punctuators re-fire over covered event time after a restart
StreamTimePunctuatorRefireOnRestartTest. Run 1 climbs event time and closes cleanly; run 2 restartsthe same application id with records below run 1's base. Stock punctuates at the restored mark; the
PC path punctuates sixty and fifty-six seconds below run 1's base. Asserted below the base rather
than below the highest, because a replayed run-1 record would satisfy the looser band and PC's own docs
say completed records beyond the frontier are replayed rather than skipped.
Cause is named at
seedStreamTime's call site: a group whose commits this module wrote carries PC'sfrontier payload, not Streams'
TopicPartitionMetadata, so the decode yields UNKNOWN and there isnothing to seed from. Fix direction recorded in the assertion itself - populate PC's own opaque rider
(KTD-S7), do not reintroduce Streams' metadata as a second writer.
Its control arm was built and run first and earned it twice: it failed once because run 2's "late"
records were stamped shortly after the epoch and fell straight out of Kafka's time-based retention, and
again because it expected silence from run 2 when a restored mark still punctuates once, at the
restored value.
Three holes the stream-time work recorded against itself, closed here
close()'s drain could advance the mark over work a forced shutdown killed. A freeze flag israised at the top of both close paths and checked where the mark is published.
RejectedExecutionExceptionfrom the pool left a record in limbo.abortAllActive()races apump by design; before this, the exception left
dispatchAvailable- and therefore the patchedprocess()- on the StreamThread with the in-flight count already incremented for a record thatwould never run and never report.
PreparedRecordvalidated nothing. A preparer returning a nullRunnableproduced an NPE on aworker, which is classified as a processing failure - so with retries disabled it blocked that key's
shard for the life of the task and told the user their topology threw. It fails at its constructor now.
The restart seed also reaches the PC path's own
RecordQueue, without which a topology using Kafka'sshipped
UsePartitionTimeOnInvalidTimestampthrows on the first record after a restart where stockrecovers.
Proven able to fail
Five sabotages, each with its prediction written down first, each reverted and verified
byte-identical by checksum. The interesting one is the prediction that was wrong.
got exactly those four, with the two pool-of-one cases staying green - which they must, because a
minimum over a singleton is the maximum. That asymmetry is what says the right thing broke.
went red. Every other stream-time case derives the timestamp from the record and cannot tell them
apart, which is the whole reason that case exists.
workers do not reliably enqueue their failures before
close()'s own drain has polled an emptymailbox, so the mark stayed put for a reason unrelated to the flag. The advance is real - with two
killed records, three of the four orderings produce it - but it is a race, and a test that passes
either way is the "green that asserted nothing" this module keeps catching in its own work. The test
now asserts the field's actual claim, that a closed dispatcher's mark is frozen against anything
reaching it, via the one route that publishes without a drain; re-run, exactly one red, on that
assertion.
finding: the original assertions read the failure and in-flight counts, neither of which the
stream-time hold touches, and the claim about the hold is not provable on this code at all - both
close paths freeze the mark, and a fixed pool with an unbounded queue rejects nothing until it is shut
down. Renamed and re-asserted to what is observable, and the hold's unobservability is stated in the
javadoc rather than glossed.
STREAM_TIMEonly. Exactly one red, on the wall-clockassertion, with the same test's stream-time assertion still passing and the seam-off control still
green. That is the gate on this PR's central decision: delete it and the handover's ranked defect 1
goes back to being silent.
One unpredicted red, diagnosed rather than filed as a flake. The low-water sabotage also reddened an
unrelated signal-registration test. It was a cascade: the sabotage made a stream-time test fail at its
precondition, before the
close()that would have deregistered its local dispatcher, and the classruns
SAME_THREAD. Fixed at the source.Trigger bookkeeping - the default does NOT move
PcDispatchSwitch's javadoc used to defer punctuation todocs/inflight/as "a separate outstandingitem, already priced in". That sentence is replaced by what is now true - implemented, loud, and with
two measured gaps - and the paragraph says explicitly that the flip is a decision for after the
sibling rungs reconcile: the exception-surfacing trigger is still open and lives on another branch, and
a default moved on one rung's evidence while another holds an unclosed reason is a default nobody
measured.
The claim this work falsified was written down in three places, each in its own words, so none would
have been found by grepping for the others: the module README,
CONCEPTS.md's "Refused construct", andthe refusal reasons themselves - the
@DoNotCalland@deprecatedstrings a user reads as a compileerror, plus the enum they mirror. Leaving those would have shipped a config lie: the reason is no longer
that the clock is stuck, it is that the clock is a low-water mark over concurrently dispatched work, so
which records fall inside a window and which version of a table a record sees move with dispatch order.
The
windowedBystrings argue fromobservedStreamTime, a different field this work did not touch, andare left alone apart from one clause that read as if it named the task's value.
Verification
Seam-off oracle run before and after on this machine, whole
testphase, per-class numbers read outof
target/surefire-reports-kafka-upstream/rather than copied: identical, unchanged, with the sameskips. The module's own unit suite and the integration lane are green, including the new punctuation
arms.
bin/check-all.shexits 0 in this worktree - which is where it had to be run, because thepre-commit gate evaluates a different one (#382); these commits then used
--no-verify.Two process notes worth keeping:
-Dscoping flag can turn a build failure into a false negative in either direction. Scoping thefailsafe run without
-Dfailsafe.failIfNoSpecifiedTests=falsefailed in the parent module before thestreams module ran at all - and the wrapper reporting the shell's exit code, not Maven's, showed that
as a success. Every sabotage run here deletes both report directories first and asserts reports exist,
so an early death cannot be read as the previous run's clean pass.
shouldReadCommittedStreamTime*cases pass with the seam verifiably on, which exercises the dispatcher seed and the PC branch of the
stream-time read. Removing only the
RecordQueuehalf of the seed and re-running leaves them green,case for case - so that half is unproven, Kafka's suite has no case for the extractor it protects, and
the
docs/inflight/note says so rather than letting the run read as coverage.Where this came from
Copied out of #271, the Kafka Streams feasibility study, and reconstructed
from a branch forest whose value is that its claims changed under measurement: a stream-time branch that
retracted its own headline, a commit-coverage branch narrowed to idle tasks by a sibling that refuted
it, an effect-survival branch that had to rebuild its experiment after review found it was measuring its
own shutdown, and a restart branch that confirmed the one claim the others were not about. Those
retractions are carried rather than tidied away, because they are what stops the next reader
re-proposing what was already refuted.
The PC-side classes are renamed
io.confluenttobz.stub; the patched Kafka sources stay inorg.apache.kafka. The patch was regenerated by the prescribed procedure -process-sources, edit undertarget/kafka-patched,bin/regen-patch.shwith no maven run in between - never hand-edited, andverified by content rather than by hunk count: every added and removed line of the parent's patch
survives except the ones deliberately rewritten, each of which is named in a commit body, and no removal
was lost.
Fifth rung of the reconstructed Wagon B stack. Not here, and named so a reader does not expect them:
backpressure and error surfacing, which owns the exception-type trigger the default still waits on; the
seam-on upstream evidence lane, which owns the gate this PR's manual seam-on run stands in for; the
benchmarks; the example module.
Checklist
unsupported and NOT refused"),
CONCEPTS.md's "Refused construct" plus two new entries, thePcDispatchSwitchjavadoc that owns the default, and adocs/inflight/note that replaces theone this work falsified
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 four rungs below gavePcTaskDispatcherTest(including thedivergence pinned as a characterisation test, the extracted-versus-broker timestamp pin, the seed
arriving mid-flight, the close freeze and the rejected submission), and five broker-backed classes:
the restart re-fire, punctuator effect survival with its negative control, the post-commit
checkpoint gap, commit coverage, and the registration warning with a seam-off control
ce-simplifyandce-code-reviewlocally -N/A - asked for review on the PR instead. Most of the diff is reconstructed from branches already reviewed on astubbs/parallel-consumer#271; what is new here - the defect-1 decision and the evidence behind it, the three closed holes, the corrected refusal strings, and the sabotage that refuted its own prediction - is what a reviewer should spend attention on