Repository navigation
Conversation
…ith the controls that keep it honest Adds parallel-consumer-example-streams-pc: a main() that starts its own broker in Docker and runs one workload twice - stock dispatch, then PC dispatch - varying only PcDispatchSwitch, then prints both latency distributions side by side. The topology is ordinary Kafka Streams with no Parallel Consumer API in it; taking the dependency is the whole integration. The effect it shows is head-of-line blocking: one 1500ms record at the head of a single partition with 24 fast 25ms records behind it on other keys. Under stock dispatch every fast record waits for it; under PC dispatch they do not. Measured on this machine at 13.3x on the median fast record and 1.51x on the whole-batch drain, with the median reading between 13.3x and 19.0x across five consecutive runs. No figure from the branch this was copied from is repeated: those were measured without wake-on-work and do not describe this build. The switch is turned on EXPLICITLY, in both arms. It defaults off here, so an arm that merely omitted the call would silently be a second stock arm and the comparison would be of one thing against itself. That default is also the demo's subject: unsupported topology shapes are not yet refused, so an opt-in preview is the honest configuration, and this module is exactly the artifact that opts in. WHAT THE DEMO REFUSES TO LEAVE TO TRUST - ClasspathGuard. The patched Kafka classes take effect only by winning classpath order, and when they lose nothing throws - the run would print plausible numbers meaning nothing at all. It resolves each patched class first, prints where it loaded from, and aborts loudly if the stock jar won. It then asks the loaded StreamTask for the pcDispatcher field the patch adds and prints its declared TYPE, because provenance and content are different questions with the same happy answer. Its unpatched control is now PartitionGroup: not in this branch's patch set, and the exact thing the seam routes around, so seeing it come from the stock jar says the seam replaced dispatch and left Kafka's own queueing alone. - PcDispatchCounters, printed per arm. dispatchedToPool is incremented at exactly one place in the codebase, so 0 in the stock arm and 25 in the PC arm is evidence rather than assertion. splitPollWaits and wakesOnWork are printed beside them, separately, because wake-on-work is a second mechanism that arrives on the same rung and would otherwise be credited to the seam. THE NEGATIVE CONTROL MOVED, AND SO DID THE CHECK ON IT The same workload with every record on ONE key, where PC's key ordering leaves no concurrency to exploit. On the branch this was copied from that arm LOST, at 0.69x, and the demo's prose explained the loss and said the fix was a later change. That fix - the split poll wait - is present here, and the arm now lands on parity: 0.99x, 0.99x, 0.98x, 1.01x across four runs, plus one reading 1.19x. That one reading is why the section's check was rewritten. The inherited check asked which DIRECTION the median went, which is a sound test of an arm that loses and a coin toss for an arm at parity - it fired on the 1.19x run and would fire on roughly half of all healthy runs. It now tests a band around parity, in either direction, because a PC WIN here would void the headline rather than strengthen it. A NOISE FLOOR, MEASURED INSIDE THE RUN The 1.19x run's two STOCK arms differed by 1.20x. They do byte-identical work - stock Kafka Streams ignores keys, and the per-record cost here is chosen by value - so that difference is pure variance, and the control had measured nothing. The summary now prints it as a NOISE FLOOR row and tells the reader to read every ratio against it. Without it on screen that run looks like a finding. PROVEN ABLE TO FAIL, AND THE PREDICTION THAT WAS REFUTED The dispatch switch was neutered so the PC arms ran stock, with six predictions stated first. Five held: dispatchedToPool read 0 and the per-arm warning fired, splitPollWaits read 0 and its warning fired, the summary printed its DO NOT QUOTE banner, the negative control stayed green, and ClasspathGuard still passed - localising the loss to the switch rather than to the build. The sixth was refuted, and it was mine. The headline check asked only whether PC came out ahead at all; the sabotaged run measured 1.01x, satisfied it, and printed "a fast record no longer waits for an unrelated slow one" over a run in which every fast record had waited. A check a broken run can satisfy is not a check. It now requires a margin far outside any measured noise - 2.0x, sitting in the empty space between a sabotaged 1.01x and a real 13x-19x - and re-running the same sabotage fired it, with the negative control still green. That asymmetry is what says the right thing broke. LatencyScenarioTest pins that case in the ordinary build, because a passing demo run never enters the contradiction branch and the demo itself is behind -Pdemo, so nothing in CI would otherwise reach it. Verified red by relaxing the threshold to 1.0: exactly that one case failed. TWO GATES THIS MODULE MET THAT ITS SOURCE BRANCH DID NOT Neither is a copy defect; both fire on our own code and are fixed rather than excluded, as the machinery rung said would happen. - Error Prone ReturnValueIgnored on the guard's reflective getDeclaredField. Fixed by using the result - printing the field's declared type - which is better evidence than the bare presence check it replaced. - forbiddenapis on Console's locale-dependent String.format. Fixed with Locale.ROOT, which the report wanted anyway: it is a fixed-width table of grouped numbers, and a locale that groups differently would misalign the columns a reader compares down. The demo is gated behind -Pdemo and bound to process-classes, so an ordinary build never starts a broker. parallel-consumer-example-streams is untouched: it is the provably-unpatched baseline arm, and a PC dependency there would destroy that evidence. The new module is added to both Maven Central deploy exclusion lists, so a release cannot publish a demo. Copied from #271's feats/ks-streams-pc-example branch and adapted: the Java packages renamed io.confluent to bz.stub, StreamThread moved from the guard's unpatched control into its patched set, and the cross-references to a HeadOfLineBlockingBenchmarkTest that does not exist on this stack removed rather than left pointing at nothing. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Um4ssjUsPh8ng73U7iKvdE
…ams seam The workstream signpost listed "the example module" among the things still off master. It is not, now, and the correction is worth more than accuracy: the note's whole complaint is that an agent reading it finds sideways references to the work and no way in. A runnable demo that prints its own evidence is that way in, and leaving it in the not-yet list would send the next reader to the branch forest for something master already has. Says what the demo shows rather than only that it exists - both arms, the dispatch counters as evidence, the negative control and the within-run noise floor - because "there is an example module" does not tell a reader whether it answers their question.
… dragging two CVEs onto the runtime classpath Three findings from the module's first CI run, all of them things a new module in this repo has to do and none of them visible until it existed. WIRE UP ARCHUNIT EveryModuleWiresUpArchUnitTest failed the Unit Tests lane: every module with test sources needs its own TestConventionsArchTest, because ArchUnit imports from the classpath of the module it runs in, and without one it silently does not run there at all. A sibling in parallel-consumer-example-streams names the package bz.stub.parallelconsumer.examples.streams, which looks as though it would cover this module's .pc sub-package - it does not, and the file says so, so the next reader does not delete it as a duplicate. The core test-jar comes with it, for the one class that holds the shared rules, at test scope so it cannot reach the demo's runtime classpath and disturb the ordering the whole mechanism depends on. NO SLF4J BINDING, AND THE CLAIM THAT SAID THERE HAD TO BE ONE IS WRONG deps: whole-tree CVE scan flagged logback-classic 1.6.1 (CVE-2026-19880, CVSS 6.3) in this module and nowhere else. That is not a coincidence: the scan audits the RUNTIME scope, logback is test scope everywhere else in this repo and therefore invisible to it, and this module's compile-scope declaration was the first thing in the tree to put it there. The inherited pom comment defended that declaration on the grounds that without a binding, main() warns about missing providers and Kafka's INFO logging then buries the report. The first half is true; the second half is not, and it was worth measuring rather than believing. SLF4J 2.x with no provider is a NO-OP, not a fallback to console. Measured: a run without logback emits three warning lines at startup and then ZERO log lines between the classpath check and the summary. Removing the dependency made the report cleaner, removed the finding at source, and cost the module two dependencies and a config file it did not need. The pom now records that as a decision rather than an omission, with the measurement, so re-adding a binding is a change somebody has to argue for. PIN COMMONS-COMPRESS PAST ITS TWO ADVISORIES Same gate, same cause, different dependency: testcontainers has to be compile scope here because DemoBroker lives in src/main, and it resolves commons-compress 1.24.0, which carries CVE-2024-25710 and CVE-2024-26308. Both are denial-of-service on malformed archives, both first fixed in 1.26.0, and neither is reachable - testcontainers uses this to build Docker contexts and copy files in, never to parse an archive this demo did not create. A fixed version exists, so it is pinned rather than argued about in an exclusion entry, and verified by running the demo end to end against a real broker on 1.28.0 rather than by assuming the API held. Module-local, for the reason example-streams gives for its own jackson pin: this is the only module putting testcontainers on a non-test classpath, so a root pin would change what every other module's tests run against to fix a problem only this one has. NULLAWAY ON THE TWO METHODS THIS BRANCH ADDED Both LatencyScenario.contradiction and stockMedianMillis were designed to return null, and NullAway said so. There is no nullability annotation library on this classpath; the codebase's idiom is Optional, so they return Optional<String> and OptionalLong. The remaining NullAway and Error Prone warnings on these files are inherited from the branch this module was copied from and are triaged in the PR body rather than silently left. Verified: the module's tests pass, and the demo runs end to end - 19.32x on the headline median, 1.00x on the negative control, 1.00x noise floor, no warnings. Two parallel-consumer-core tests (processInKeyOrder, JStreamParallelEoSStreamProcessorTest.testConsumeAndProduce) failed once in two local runs and passed on the second with nothing changed, both with the "expected N, got 0" shortfall that docs/inflight/bug-857-family.md already records as this box being saturated rather than a regression. This branch touches no core code.
…iff that are worth answering bin/check-pr-analysis-surfaces.sh reported seven findings on lines this PR wrote, none inherited. Two of them are right and are fixed; the other five are one detector making the same objection to the same deliberate shape, and are left with the reason recorded here rather than in five suppressions. Fixed: - PSC_PRESIZE_COLLECTIONS - DemoBroker.createTopics allocated its list unsized when the exact count is the varargs length it already has. - OCP_OVERLY_CONCRETE_PARAMETER - printSummary took List where it only iterates, twice. Iterable says what the method actually needs. Not changed, and it is the same finding five times: CE_CLASS_ENVY on printIntroduction, printClosing, printSummary, ClasspathGuard.failLoudly and verifyDispatchSeamIsPresent - each "excessively uses methods of another class", that class being Console. Every one of them is a report section, and Console is the report writer, so calling it repeatedly is what they are for. The refactor the detector is asking for would move this demo's explanatory prose inside the formatting helper, which is where a reader would never look for it. The `static: spotbugs` gate passes either way; this is the advisory surface, answered rather than left.
…rgence nobody has named The seam-off execution has always been the behaviour-preservation gate: the patch changes nothing while the switch is off. This adds its opposite - what Apache Kafka's own suite says about the path this module actually ships - and, more usefully, a verdict on it. THE PROBLEM WITH A SEAM-ON RUN, AND THE SHAPE THAT SOLVES IT Seam-on failures are expected. The refusal envelope throws where stock would have run a construct; the committed offset is Parallel Consumer's frontier rather than Streams' consumedOffsets; a record handed to a worker has not been processed by the time process() returns. A gate that goes red every build for those reasons is a gate people stop reading, which is why the branch forest's version of this run recorded its failures and gated on nothing. So the run and the verdict are SEPARATE surefire executions. kafka-upstream-tests-seam-on carries testFailureIgnore and never reddens a build; seam-on-evidence-classification reads what it wrote and fails when a divergence matches no named mechanism. That split is the only arrangement in which "these failures are expected" and "a new divergence fails the build" are both true. THE SEAM-OFF RUN IS THE CONTROL ARM, AND IT IS DIFFERENCED CASE BY CASE A divergence is a case that PASSES with the switch off and FAILS with it on. Nothing else is attributable to the dispatch path: a case failing in both arms is failing for a reason the seam had nothing to do with, and counting it would credit a change that did not cause it. The classifier therefore reports four disjoint sets - divergences, failing-in-both, passing-only-seam-on, and the control arm's own failures - rather than one number. MECHANISMS ARE DERIVED FROM THE CODE WHERE THAT IS POSSIBLE RefusedConstruct takes its marker as the longest common prefix of every message PcUnsupportedConstruct can produce, and attributes a failure to whichever constant the message names. No refusal string is copied, so the refusal envelope - which belongs to another rung and moves independently - can gain or lose a construct with no edit in this lane at all. CommitFrontierEncoding recognises an assertion that renders an OffsetAndMetadata, which is an assertion about the commit encoding whatever the case is called. Its limitation is stated rather than hidden: an offset REGRESSION inside a Kafka unit test renders identically and would be explained by this mechanism too. Nothing in the assertion text separates the two, so the lane does not pretend to - what defends offset correctness is a broker-backed law, not this. WHAT CANNOT BE DERIVED IS ATTRIBUTED IN THE LEDGER, NOT HERE "This assertion fails because Kafka's test drives the task synchronously" is a judgement about a test, not a fact recoverable from `expected: <5> but was: <0>`. Those attributions live in docs/inflight/test-streams-seam-on-divergence-triage.md, keyed by method rather than by parameterisation, and the lane reads them. Three consequences, all deliberate: the reasoning is reviewed as prose by whoever changes the semantics under it; the PR that closes a mechanism deletes its entries in the same change; and an entry that stops matching is reported as stale rather than rotting silently. THE FLAKE IS ABSORBED BY MECHANISM RATHER THAN BY A RETRY StreamThreadTest.shouldLogAndRecordSkippedRecordsForInvalidTimestamps has been failing this module's oracle intermittently for weeks, and its note says so. Because the oracle fails fast, one sighting used to abort the lane before the measured arm ran at all - a full re-run of Kafka's suite twice, for a failure already diagnosed. The lane now relaxes THAT ONE EXECUTION and makes the verdict itself: every control-arm failure must carry a flaky-case marker in docs/inflight/, or the lane stops and says there is nothing trustworthy to difference against. The ordinary build's oracle is untouched and still fatal. The flake registry is consulted FIRST on the control arm and LAST among the divergence mechanisms, and that ordering is load-bearing - see the triage note's own finding. PROVEN ABLE TO FAIL, THREE WAYS, WITH THE PREDICTION STATED FIRST Sabotage, end to end. Removing the close-time commit-state publication from PcTaskDispatcher.close() was predicted to make exactly one case, shouldClearCommitStatusesInCloseDirty, fail seam-on while staying green seam-off, with no mechanism recognising a bare commitNeeded assertion. Exactly that: one unexplained divergence, named, with its recorded failure and stack; every other divergence still explained by the same six mechanisms; the script exited 1. Reverted and verified byte-identical by checksum. That run also demonstrated the control-arm path unplanned - the invalid-timestamps flake fired seam-off, and the lane named it from its ledger note and carried on instead of aborting. A stale report directory. Running the classification execution alone against report directories from an earlier run failed on the freshness check, naming the file, when it was written and when the build started. maven.build.timestamp is passed in for exactly this, and is required rather than optional: a check that turns itself off when its input is missing is the failure mode this lane exists to avoid. A missing report directory. The same execution against deleted directories failed with "That is an ERROR and not an empty result", rather than reading an empty glob as zero failures. SeamOnDivergenceClassifierTest pins all three at unit level on synthetic reports, plus the cases the end-to-end runs cannot cheaply reach: an arm whose pin did not arrive (both arms recording dispatch=false differences the control against itself and reports a flawless zero), arms covering different classes, an attribution to an undeclared class, and a triaged attribution beating a flake marker on the same case. NO COUNT APPEARS IN THE MACHINERY OR IN THE LEDGER. The per-arm totals are printed for a reader and the verdict does not consult them; which cases diverge moves with the branch, the Kafka version and the seam, and the command that re-derives them is in the note and in the README. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ot a local-only claim A lane that only ever runs on the machine of whoever wrote it is half a deliverable: the first person to change the dispatch path's semantics learns nothing, and the ledger of attributions rots with nobody to notice. RED HERE IS NOT EXPECTED, WHICH IS WHAT MAKES IT DIFFERENT FROM THE QUARANTINE LANE That lane runs tests known to be failing, so its verdict lives in the step summary and its run step carries continue-on-error. This one asserts that every seam-on divergence is EXPLAINED, and green is the ordinary state. Red means one of exactly two things, both wanting a human rather than a re-run: a divergence nobody has attributed - something changed the dispatch path's semantics - or an attribution in docs/inflight/ that went stale because a rung fixed the thing it described. So the job gates. PATH-SCOPED, AND THE LEDGER IS IN THE TRIGGER SET The run costs Kafka's whole suite twice, and nothing outside the streams module, the lane script and docs/inflight/ can change its answer. The ledger belongs in that set on purpose: an edit to an attribution changes the verdict exactly as much as an edit to the code does, and a triage note edited without re-running is how a lane starts lying. The whole measurement goes to the step summary and is kept as an artifact with both arms' surefire reports, so a red run can be read without re-running it. A red on the dependency fetch is a download failure rather than a divergence, and the step says so: this module pulls Kafka's sources and test-sources CLASSIFIER artifacts by explicit artifactItems, which dependency:go-offline does not resolve, so the cache-warming job does not cover them. That gap is already recorded in docs/inflight/, and this workflow points at it rather than restating it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…Kafka Streams' own control paths Two things Kafka Streams does for itself stopped working the moment records went to a worker pool, and both failed silently. This rung restores them. FAILURES REACHED THE TaskManager WITHOUT THE TYPE IT DISPATCHES ON. TaskCorruptedException and TaskMigratedException are not errors; they are Kafka's control-flow signals - close this task dirty, revive it, re-initialise its state. StreamThread catches them BY TYPE. Every worker failure came back out of process() wrapped in a StreamsException, so recovery never fired and an application stock Streams would have recovered shut its client down instead. ProcessorNode.process already rethrows those two unwrapped, so mirroring stock's own catch ladder is the whole of the fix: a StreamsException passes straight through. AND IT REACHED IT TOO LATE FOR THAT TYPE TO BE USEFUL. Even unwrapped, the failure arrived at the top of a later pump - possibly in a later runOnce. Kafka's own StreamThreadTest.shouldReinitializeRevivedTasksInAnyState asserts the stock shape directly: one runOnce, one throw. A pump that has run out of work now waits, bounded, for the outcome of what it already dispatched, which is enough because Kafka's loop calls process() again while it reports progress: the pump that dispatched the record returns "progress", the loop comes back, and the second pump waits and throws. The unwrap alone was declined once and recorded rather than taken, and this is why: with the timing untouched no test can distinguish it from its absence, and a change no test holds in place is one the next refactor removes for free. NOTHING COMMITS PAST A FAILURE (#271's open thread). Kafka's loop is process, punctuate, commit - so a failure landing in that window let a scheduled commit make another key's completed offsets durable for a task about to be closed dirty and rewound. For a TaskCorruptedException that is worse than a duplicate: recovery wipes the state stores those offsets claim to cover. A pending failure therefore fences commit-data collection. It deliberately does NOT fence hasUncommittedWork(), which is the opposite question - that one is what validateClean turns into a TaskMigratedException so the task closes dirty, and fencing it would make a failed task look clean to close. Two tests pin the pair. A failure also stops further dispatch, which bounds what runs after a known failure to what was already in flight rather than to a whole poll budget. In-flight records are left to finish rather than interrupted: a worker cancelled mid-chain leaves a half-forwarded record. ONE DELIBERATE DIVERGENCE FROM STOCK, AND IT IS THE INTERESTING ONE. Stock rethrows a TimeoutException raw, which TaskExecutor reads as retriable - keep the task running, come back to it - and that is safe there only because the record is still in the queue and will be re-selected. Here retries are disabled and the failure bar has closed dispatch, so honouring it would mean process() returning false for ever, task.timeout.ms never tripping and the partition staying paused: zero throughput, no exception, nothing logged, from an ordinary broker timeout. Matching the exception TYPE would have broken the exception CONTRACT, so it is wrapped. NOTHING BOUNDED INFLOW BUT HEAP. Stock pauses a partition once its RecordQueue passes buffered.records.per.partition. On this path PC holds the records, partitionGroup.numBuffered() answers zero, and the pause never fired. It is applied now from the same threshold with the occupancy read from PC, and the resume mirrors the pause exactly - only partitions this path paused are ever handed back, because Kafka pauses partitions for reasons unrelated to buffering and a resume computed from the whole assignment would restart fetches on those. numBuffered() and the active-buffer-count metric follow. THE OCCUPANCY IS DERIVED FROM PC'S INCOMPLETE SET, NOT COUNTED ALONGSIDE IT, and this overrides a decision recorded on the branch this work reconstructs. That branch kept a standalone count raised by predicting, per record, what PC would accept - explicitly because getNumberOfIncompleteOffsets() counts a failed record for ever and would pause that partition permanently. The objection is real and is answered rather than ignored: a failed record is never subtracted back out of the handed-out tally, so it cancels itself out of the derivation. What the prediction could not answer is the defect ranked second on this workstream's handover: PC accepts an offset into its incomplete set while ProcessingShard.addWorkContainer DROPS the arriving container when a live one for that offset is resident - which is what happens to a re-delivered offset after a seek, an offset reset or corruption recovery. That record is never handed out, nothing decrements it back, and the partition is paused for good with no symptom but a topology that has gone quiet. Deriving is immune to it, because the incomplete map's insert is idempotent per offset. A test drives exactly that shape and asserts the count comes back to zero, which a permanent residue cannot do. A negative occupancy is clamped, COUNTED AND LOGGED rather than swallowed - that is the drift direction which removes the memory bound with nothing to say so, and a bare clamp is what hid a conditional-decrement bug in the engine's own intake gate. THE SETTLE WAIT IS GATED ON RUNNING OUT OF WORK RATHER THAN OUT OF CAPACITY, and a measurement forced that. Unconditional, it also fired in the saturated case - pool full, plenty of work, nowhere to put it - where the StreamThread's next act would have been to poll. With the pause switched off so nothing else bounded inflow, peak occupancy over a 600-record backlog measured 36 with the wait and 596 without it: a sixteen-fold intake throttle, and worse, a second undesigned memory bound sitting inside the CONTROL ARM of the memory-bound proof. The residual the gate leaves is stated rather than glossed: at poolSize 1 a single in-flight record fills the pool, so the same-runOnce guarantee is a guarantee about a pool with a spare slot. EVIDENCE. Seam-off oracle run before and after on this machine, whole test phase, per class out of target/surefire-reports-kafka-upstream/: identical and unchanged, same assumeTrue/assumeFalse skips. Seam on, same experiment with one term changed: shouldReinitializeRevivedTasksInAnyState goes green on both parameter combinations this module supports and shouldRecordBufferedRecords goes green with the backpressure work, with ZERO regressions. Its third combination stays red under Kafka's private processing-threads config, where DefaultTaskExecutor calls task.process from its own thread - out of scope since the seam landed and named as such in PcTaskDispatcher's threading contract, not a residue of this defect. PROVEN ABLE TO FAIL. Three sabotages, each with its prediction stated first, each reverted and verified byte-identical by checksum. - The type passthrough removed: predicted exactly those two upstream cases red and nothing else moving. Exactly that, 65 failing to 67. - The settle wait removed: same prediction, same result - and our own unit suite stayed entirely green, which is the honest reading that no unit test covers pump-level delivery timing. - The commit fence removed: predicted one unit case red, on the fence, with its control arm (the same shape with no failure, where the frontier IS offered) staying green. Exactly one failure, and it was that one. The memory bound has a control arm rather than a sabotage, which is stronger: two arms differing in exactly one term on the same broker, topology, data, cost and JVM, with occupancy sampled from a watcher thread because the interesting quantity is a peak and a peak is invisible to any assertion made after the run. Over a 600-record backlog it measured 596 held with the pause off against 30 with it on - the fixed arm landing exactly on its derived bound, which is what says the derivation is right rather than merely generous. The patch was 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 line the previous patch added is still added except the sixteen this rung deliberately rewrites, and no removed line was lost. Copied out of #271, which keeps the source, the measurements and the open review threads. PC-side classes are renamed io.confluent to bz.stub; the patched Kafka sources stay in org.apache.kafka.
… settles, and what it reserves The dispatch seam has now had three reasons for defaulting OFF, and each was closed by a measurement that named the next one. The third - a control-flow exception raised inside a processor never reaching Kafka's recovery - is closed by the commit before this one, and this time the measurement named nothing. That is a more delicate state to leave behind than an open defect, because "no reason left" reads as "flip it", and the flip is not this rung's to make: sibling rungs of #255 are in flight against the same base and every one of them moves the seam-on numbers. So the three sites that carry the decision are re-pointed rather than left stale - PcDispatchSwitch's javadoc, which owns it, the module README, and the pom's module description - and they now say the same thing: the trigger is closed, the numbers that close it, and that the flip waits on a fresh seam-on measurement taken on the reconciled module. docs/inflight/streams-dispatch-default-flip-is-reserved-until-the-rungs-reconcile.md carries the reservation itself, so a reader arriving from the inflight index finds it rather than inferring it. The README also gains what a field tester needs to have changed under them: the third switch and what turning it off costs, a backpressure section carrying the two-arm measurement rather than a claim, and an error-surfacing section covering the type, the timing, the commit fence and the one place this path deliberately diverges from stock (a TimeoutException is wrapped rather than rethrown raw, because matching stock's TYPE would have broken stock's CONTRACT). The unsupported-and-not-refused list loses the typed-exception entry and gains Kafka's private processing-threads mode - which was always out of scope but was only ever recorded in a javadoc, so a seam-on run showing that parameter red had nothing to recognise it by. Two write-ups land in docs/solutions/ rather than in the notes, because both outlive this work: - An async seam owes a control-flow exception BOTH its type and its timing, and fixing one leaves an unverifiable change rather than a partial one. It also carries the structural argument this rung got wrong: "the pump handed out nothing" conflated an idle pool with a full one, and waiting in the second case throttled intake sixteen-fold - which showed up not as a slow benchmark but as a contaminated CONTROL ARM, where every assertion still passed. - Predicting what a collaborator will accept drifts from what it accepted; derive the count from the collaborator's own state. Three defects go together rather than one at a time, and the objection that had blocked deriving for a release turned out to be an objection to one particular derivation. The typed-exception note is deleted by the commit that closes it, its content having moved to the first of those. The settle-wait note is new and is NOT a "we should benchmark this some day": it records the residual the gate leaves (at poolSize 1 a single in-flight record fills the pool, so the same-runOnce guarantee is a guarantee about a pool with a spare slot) and the one regime the sixteen-fold figure does not cover. No changelog entry, deliberately - the file is generated at release time from these bodies.
One asterisk pair in the settle wait's javadoc renders literally rather than as emphasis. Caught by reading the file back rather than by any check - there is no javadoc linter here.
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>
…fier fetch timing out The Unit lane went red on this rung with the streams module FAILURE and every other module SUCCESS, at maven-dependency-plugin:unpack, on a Maven Central read timeout for kafka-streams:jar:sources:3.9.2. No test ran - the module dies in generate-sources - which is exactly the misdirection the note describes, and exactly the failure the rung below this one diagnosed and left recorded rather than fixed. Worth one line rather than nothing, because what the note needs and did not have is a second, independent occurrence. One red lane on one branch is that branch's problem; the same signature on two branches stacked at different points is the module's, and that is the difference between "re-run it" and "the warming job does not cover this artifact". Deliberately not fixed here. The candidate fix is a change to prepare-deps in maven.yml with its own verification - the unpack executions logging a cache hit rather than a Downloading from central line - and that is a workflow unit, not a line in a feature branch.
…ry was writing a plain int off-thread Self-review finding, and it is this module's own recorded defect class rather than a tidy-up. getBufferedRecordCount() is documented and used as callable from ANY thread - the memory-bound proof's sampler reads it while the owner mutates - and it raises the underflow counter on the clamped path. That counter was a plain int documented as owner-thread-only, so the sampler wrote it unsynchronised, and the log-once guard (bufferedUnderflows++ == 0) could fire twice when two threads both read zero. Now an AtomicInteger with getAndIncrement, and the threading contract says so rather than claiming an owner-thread-only field that three call sites contradict. The rule this brushes against is the one this module already has a write-up for: a question must never mutate. It still mutates, and that is admissible only because what it touches is a diagnostic counter no answer depends on - it reaches neither the WorkManager nor the state the buffered count is computed from. Said at the field rather than left for a reader to re-derive, because the same reasoning applied loosely is how the original cross-thread drain got written. No behaviour change: the count is diagnostic, and the memory bound is unaffected. The dispatcher suite is green.
…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>
…, and it takes the integration lane too The note predicted that re-running would not reliably help, on the grounds that the runner is often reassigned to the same region. That prediction has now been tested rather than quoted: a fresh push reproduced the identical failure, at the same execution, on the same artifact - and this time the Integration lane went down with the Unit lane on the same run. Two things change with that. The exposure is not "one lane occasionally", it is every lane that builds this module; and re-running has stopped being a reasonable response, which matters because it is the response an agent reaches for by reflex when a lane goes red on an artifact download. So the note now says what a blocked PR should do instead: name this note in the PR and let the module's own lane stand as the evidence the code is healthy. A local run has the artifacts in ~/.m2 already, which is why it never reproduces there - and why "green locally" is a real signal about the code and no signal whatsoever about this. Still not fixed here. The candidate fix is prepare-deps in maven.yml learning to warm the classifier artifacts, verified by the unpack executions logging a cache hit rather than a Downloading from central line. That is a workflow unit with its own evidence, and smuggling it into a feature branch would leave it unverified in both places.
… way to state Apache Kafka's own suite is the oracle for behaviour PRESERVATION. It cannot state the properties this module exists for, because stock Kafka Streams has no arm in which two records of one partition run at once - so no stock test asserts what happens when they do. These are the arms that do. WHAT IS NOW ASSERTED AGAINST A BROKER, AND WITH WHAT CONTROL PcDrivenStreamsProofTest - the PC path's output is identical, record for record, to what stock Kafka Streams produced. Two controls, and they answer different questions: a same-harness seam-off arm, which separates "the patch changed the output" from "concurrency changed the output"; and the stock baseline itself, which is generated OUTSIDE this module. PcDrivenStatefulProofTest - the same for a stateful topology, which is where Kafka's own store wrappers read the per-task record context ambiently. The stateless arm's only ambient reader is the probe this module wrote; this one makes Kafka's own run. The module's existing envelope test already said this proof was the reason non-windowed aggregation is not refused - and the proof was not on this stack, so that sentence was pointing at nothing. CommitFrontierCrashRestartTest - nothing is lost or double-counted across a crash. The committed frontier is a point below which every record has been processed, not the offset of the last one handed out, and under concurrent dispatch those are different records. ShadowedStreamsControlTest - stock, single-threaded Kafka Streams running our generated classes. It is itself a control: a concurrency change that quietly breaks the serial path is not a fix. THE BASELINE IS GENERATED IN ANOTHER MODULE, AND THAT IS THE WHOLE POINT This module compiles patched copies of Kafka's classes into its own build output, which precedes the kafka-streams jar on the classpath, so EVERY KafkaStreams instance in this JVM runs the patch. A "stock arm" written here would share every defect the patch introduced and the comparison would prove nothing. The baselines come from parallel-consumer-example-streams, which does not depend on this module and asserts its own classpath is stock before recording a row. The fixture carries the INPUTS as well as the outputs, and the arms here replay those inputs rather than rebuilding them from a second copy of the generation rules - the two modules cannot share code in this reactor order, so replaying is what stops the arms drifting apart in what they were fed. Re-verified here rather than trusted: both fixture tests were run on this branch and the tracked fixtures still match what genuinely-stock Kafka Streams produces, so the baselines describe this Kafka version and not the branch they were recorded on. THE LANE ABOVE DEFERS TO ONE OF THESE, AND SAYS SO The seam-on evidence lane's commit-frontier-encoding mechanism cannot tell a deliberate encoding divergence from an offset regression: both render an OffsetAndMetadata in a Kafka unit test's assertion and nothing in that text separates them. Its javadoc names CommitFrontierCrashRestartTest as what actually defends offset correctness, because that arm asserts on records that arrived rather than on a base64 metadata string. An acknowledged blind spot with a named owner is worth more than a predicate pretending to cover it. WHERE THIS CAME FROM, AND WHAT WAS DELIBERATELY LEFT Copied out of #271's branch forest, package-renamed io.confluent to bz.stub, compiled and run unchanged otherwise. HeadOfLineBlockingBenchmarkTest is NOT carried across: it isolates head-of-line blocking on a fixture whose own construction bounds its multiplier, and the settled decision is to lead with backlog catch-up instead. The two files that referenced it now say so rather than pointing at a class that is not here. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
… that says when to ignore it A card issuer screens every authorisation against a fraud-scoring service. The screener has been down, authorisations have queued, and it has just come back. That is the workload: the canonical Kafka Streams enrichment shape, keyed the way a real deployment would have to key it, with a skewed keyspace and a heavy-tailed service cost - the two things that cost this seam the most. Three arms, not two, and the third is the point of this change. THE NOISE FLOOR, MEASURED INSIDE THE RUN THAT QUOTES IT The extra arm is a second STOCK arm, given the same records. Whatever it differs from the first by is this machine in this state, and that is an upper bound on how far a ratio can move with nothing changed at all. Without it on screen a run taken on a busy machine looks exactly like a finding - and that is not hypothetical, it is what #391 found on the example rung, where a run reading 1.19x on its control had two stock arms differing by 1.20x. The control had measured nothing and nothing said so. So the headline assertion no longer asks whether the ratio beat one. It asks whether the ratio beat the floor by a margin. A direction check is a sound test of an arm that loses and a coin toss for an arm at parity, and - the reason this rule exists - it is satisfied by a run in which the mechanism did nothing at all. PROVEN ABLE TO FAIL, AND WITH A DOCUMENTED FLAG RATHER THAN A CODE EDIT Prediction, stated first: with a worker pool of one, PC dispatch has no concurrency to exploit, so the seam-on arm should land inside the noise floor and the headline assertion should fail while the run stays otherwise healthy. Exactly that. `bin/streams-benchmark.sh --pool 1` failed one case - the headline - and its message carried the whole diagnosis: measured 0.99x against a floor of 1.00x, threshold 1.51x, with both stock arms and the PC arm printed beside it. The steady-state case, which asserts on sample size rather than on a ratio, stayed green. That asymmetry is what says the right thing broke. `--pool 1` reaching the harness at all is a fix in this commit. The script has always documented `--pool N`, and the harness hardcoded four - so a reader could ask for a configuration, be told in every log line that they had got it, and be measuring something else. The same class of failure the script's own "an unknown option is an error" rule was written against, arriving by the other route. The margin rule itself is pinned by NoiseFloorTest, which runs in the ordinary build because it is a pure function of two numbers and should not need a broker or a `performance` tag to be exercised. Its first case is the one that matters: a 1.01x observation does not beat even the tightest possible floor. That is the shape of a sabotaged run, and a direction check passes it. THE SCRIPT RAN NOTHING ON THIS LAYOUT, AND WOULD HAVE EXITED 0 Carried across unchanged, `bin/streams-benchmark.sh` drove FAILSAFE with `-Dit.test` and `-DskipUTs`. These benchmarks live in `src/test/java` and this module has no `src/test-integration`, so that selects nothing, runs nothing, and succeeds. It now drives surefire - and because `failIfNoSpecifiedTests` has to be off for the Kafka upstream executions' sake, maven can no longer report an empty selection, so the script asserts a surefire report exists for every class it selected and exits 2 when one does not. A skip is never a pass. NO FIGURE FROM AN EARLIER RUN SURVIVES IN THIS DIFF The forest's numbers were measured on another machine and describe it, so the headline multipliers, the wake-on-work ablation percentage and the CPU-bound refutation are all removed from the javadoc or attributed to the branch they were taken on. `DEMO.md` shows the report's SHAPE with the numbers elided, and says why: a figure written into documentation quietly stops describing the reader's machine. The run prints its own, and prints the floor beside them. Two method write-ups come with the suite because its citations were pointing at them - choosing the statistic that states the claim, and control arms varying exactly one term. Both are records of work done on the branch forest, so their citations into that forest's plan and benchmark carry paragraph markers rather than being repointed at files that would make the record describe a run that never happened. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ifier timeout - a third run went green
I wrote "it cannot be cleared from inside one" into this note after it took both the
Unit and the Integration lane on one run. The very next run passed every lane. That
sentence was false when I wrote it and would have been inherited by whoever hit this
next, from a note filed under misdirection - which is the wrong file to leave a stale
absolute in.
What the evidence actually supports is the strength the upstream write-up already used,
and I strengthened it past what it could carry: "does not reliably help" means a re-run
is a COIN FLIP, not that it never works. Two consecutive runs failed and the third
passed, which is exactly a per-run condition - the lanes share the runner's route to
Central, so they fail together and they recover together.
Three things the note now says that it could not say before, and they are worth more
than the sentence they replace:
* The exposure is every lane that builds this module, not one lane occasionally.
* It is per-run rather than per-branch, so a green run is luck and says nothing about
the next one.
* A green lane is NOT evidence this is closed. Closed is the unpack executions no
longer reaching Central at all, which is what "Delete when" already asks for.
The guidance section changes shape with it. Re-running is legitimate here and is worth
saying so explicitly, because it is not the retry-into-green this project bans: no test
outcome is being hidden, since no test ran. What it must not do is talk anyone out of
leaving the record behind, which is precisely what a lucky green invites.
Recording the correction rather than quietly editing the claim away, because the
mistake is the useful part: I turned "unreliable" into "impossible" in the space of one
commit, on the strength of two data points.
… it pulled Docker into the unit lane Naming the seam-on classifier in this module's `default-test` execution silently dropped the two excludes the root pom sets on surefire - `**/integrationTest*/**/*.java` and `**/*IT.java` - because an `<excludes>` list REPLACES the inherited one rather than adding to it. The effect was invisible in the only way that matters: everything passed. Every broker-backed arm in this module lives in an `integrationTests` package, so they were quietly promoted out of failsafe and into the surefire `test` phase, where they ran, needed Docker, and went green. A unit lane that has started requiring Docker does not announce it; it just takes longer until the day it runs somewhere without a daemon. Caught by reading the run rather than the tick: the module's `test` phase reported far more cases than its unit suite has, and the class names in the list were the integration ones. Fixed by repeating the root's two entries beside the new one - `combine.children="append"` would express the intent, but a literal list is what a reader can check against the root pom without knowing that attribute exists. Verified both directions: the `test` phase now runs the unit suite alone with the seam-off oracle still at its usual figures, and `verify` runs every broker-backed arm including the ones this stack adds. The generated stock-baseline fixtures are classified in the copyright scanner as regenerated-by-a- generator, which is the first of the four reasons that table accepts and the same one `*.patch` carries. They are rewritten wholesale by `StockBaselineFixtureSupport`, so a header would be destroyed on the next run. Two explicit paths rather than a blanket `*.tsv`: the next `.tsv` to arrive should be classified on its own merits, not inherit an exemption written for these. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
… timeout, which settles the attribution The run after the green one carried a documentation-only change - no Java, no pom, no workflow - and reproduced the identical failure, same execution, same artifact, on both the Unit and the Integration lane. That is a control arm rather than another sighting, and it is the piece this note was missing. Everything before it was consistent with "something about that branch": a markdown diff is not. There is now no reading in which the change under review is implicated, which matters because the note is filed under misdirection and the misdirection it guards against is precisely an author reading a red lane named "Unit Tests" as something they broke. Worth the churn of another push to record, on the reasoning this note itself now carries: a lucky green must not talk anyone out of leaving the record behind, and that applies to me as much as to the next reader.
…a developer's local repo `-pl .,parallel-consumer-streams` builds neither `parallel-consumer-core` nor its tests jar, and this module needs both. On a developer box they are already installed, so the lane ran; the first CI run failed in three seconds on an unresolved `0.6.0.0-SNAPSHOT`, which is the useful direction for that mistake to fail in - it is the same class as a check that passes because of something only the author's machine has. `-am` instead, which is what `bin/chaos-test.sh` already does for the same reason, and which also brings the parent in so the enforcer's reactor-convergence rule is satisfied without naming the root. Core's own tests are filtered out by the `-Dincluded.groups` the lane already passes, so the extra module costs a compile rather than a suite. Re-run locally after the change: both arms produced, every divergence explained, exit 0. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…f-master list The workstream signpost lists what master lacks so an agent listing this directory can see the shape of the work. This rung removes its own entry rather than leaving a reader to infer it from a PR number, which is the same discipline the rungs below used. What is left on that list is now all semantics - refusal is already down, lifecycle is the parent, and what remains is stream time, error surfacing and the example module. The distinction is worth stating rather than implying: master will lack the behaviour, not the means of measuring it, so the next rung arrives with a lane that can already tell it whether it changed anything. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…m that did not reproduce it
`PcDrivenStatefulProofTest.pcDrivenAggregationMatchesTheStockBaseline` timed out on repetition 2 of 3
in the Integration Tests lane - the first run that put that class on CI. One sighting, written down
rather than re-run past.
WHAT RULES OUT THE COMFORTABLE READING
The StreamThread's own progress line, logged as the drain gave up, reports 1207 polls against 14
completions in two minutes. A thread short of CPU polls LESS; this one went round its loop briskly and
finished almost nothing, which is the shape of work dispatched and not completed rather than of a
starved machine. The ambient probe was clean - no rebalance dwell, no lag stagnation, no frozen
partitions - so the group progressed at the broker while the task did not progress at the processor.
That is why "the runner was busy" is not available here, however busy it was, and why nothing has been
loosened, shortened or serialised.
THE CONTROL ARM, AND THE TERM IT COULD NOT VARY
Run rather than guessed: the whole streams integration suite locally at the lane's own settings,
forkCount=4 with reuseForks=true, alongside core's integration suite - green, including this class.
Uncontended at forkCount=1, green in three separate runs.
The instrumentation is confirmed rather than assumed. Failsafe pins no <forkCount> of its own, unlike
surefire where the root pom's <forkCount>${surefire.forkCount}</forkCount> makes a bare -DforkCount a
silent no-op, so the property does arrive - and the autopsy dump from the CI failure shows the lane
running with the same two values.
So contention is NOT established and is not being claimed. The one term the local arm cannot vary is
core count: four JVMs and four brokers on a developer machine is not four JVMs and four brokers on
two. The note names the experiment that would settle it - the same runner at forkCount=1, which holds
the machine fixed and varies only the load - and records the two exception lines that would explain
the rest immediately if they belong to this fork, unattributed because four forks share one
interleaved log and a class that interrupts workers by design was running in the same window.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…PR has landed `bin/check-branch-self-reference.sh` caught the new note citing the pull request that is carrying it. The citation is right and stays - a merged PR is a permanent link, and it is the shortest way to say which change first ran that class on CI - so the fix is to word it as it will read afterwards and confirm that with the `post-merge: checked` markers, the same treatment the invalid-timestamps flake note already carries. One sentence did need rewriting rather than confirming: "the lane re-runs on every push to that branch" describes something that stops existing at the merge. It now names the triggers that outlive it - every push and every merge to master - and says why a second sighting is the cheap next step: one occurrence cannot tell a rare race from a one-off. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…t read differently are not comparable The duplicate-code check flagged a genuine clone on this PR - `consumeByKey` plus `flatten`, shared between the seam-on dispatch arm and the stock control arm, byte-identical apart from the deadline. Both engines agreed on it, which is the tell that it is real rather than jscpd counting a shared import block. It is worth fixing rather than explaining away for a reason specific to this module: these two are a matched experiment. The whole method here is comparing arms that differ in exactly one term, and a copy of the reading logic in each is a second term waiting to appear the first time somebody tunes one of them. Drifted topic-reading logic is already a named source of flaky CI in this repository. Now on `BrokerStreamsIntegrationTest`, which both extend and which already holds precisely this kind of thing - the same broker, the same serdes, the same starting offset. The DEADLINE stays a parameter at each call site, because how long an arm waits is a property of the workload it runs and that class deliberately carries only what the arms must agree on. Its javadoc says which two arms produced the clone, so the next reader knows why it is shared rather than inlined. Verified rather than assumed: the whole streams integration suite green afterwards, both sides of the clone included. WHAT WAS LOOKED AT AND DELIBERATELY LEFT The other clone both engines agree on is the fixture FORMAT - the writer in `StockBaselineFixtureSupport` against the reader in `BaselineFixture`. Those two live in modules that cannot share code, and that is the point rather than an oversight: the example module must not depend on the streams module, or the baseline it generates would run the patched classes and stop being a baseline. Extracting it would need a third module whose only purpose is to reconnect the two things this design keeps apart. The remaining jscpd hits are import blocks and javadoc between sibling benchmark classes - jscpd is lexical and counts comments, which is why PMD CPD, which is token-based, reports neither. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
… and this is what it caught `StreamThreadTest.shouldChangeStateAtStartClose[1]` failed in the SEAM-OFF arm on a GitHub-hosted runner - "Condition not met within timeout 10000. Thread never shut down." No prior art anywhere in this repository: not in the ledger, not in the solutions, not on the tracker, open or closed. THE LANE DID THE RIGHT THING AND IT HAD NEVER DONE IT BEFORE The control-arm integrity check had only ever been exercised on synthetic reports. On this run it named the ledgered invalid-timestamps flake beside this failure, refused to produce a verdict on the divergence set while an UNdiagnosed control-arm failure stood, and said so in a verdict that outranks the classification - which was itself clean, no unexplained divergences at all. That is the fourth way this lane has now been shown to fail, and the only one that arrived by itself. THE DIAGNOSIS, AND IT IS ONE SIGHTING The same case PASSED in the seam-ON arm of the same run. That is close to a control arm handed over for free: two arms, same patched classes, same runner, minutes apart, differing only in the dispatch switch. A defect the patch introduced cannot be REPAIRED by turning the seam on - seam-on is strictly more machinery on top of the same patched code. What varies with load, and only with load, is whether a thread finishes shutting down inside a fixed ten seconds. Recorded with a flaky-case marker, and the note is explicit that this rests on one sighting plus that asymmetry, that a failure in BOTH arms of one run would void it, and that quarantine is not the lever here - it is one sighting on one PR, and the case lives inside Apache Kafka's own compiled test class where the annotation has nowhere to go. WHY THE MARKER IS NOT A SILENCER, SINCE IT DOES GREEN THIS LANE The lane still prints the failure, still names the note beside it, and still refuses to attribute it to the seam. What the marker buys is that one timing bound in Kafka's own test stops withholding the verdict on the divergence set - otherwise the lane is permanently red for a reason that is not a divergence, and a permanently red lane is one nobody reads, which is the failure this whole design was built against. ONE MORE THING THE RUN TAUGHT THE TRIAGE NOTE The lane reported a divergence attribution as having matched nothing, which normally means a rung fixed it. There is a benign second cause, now written down: a case failing in the CONTROL arm falls out of the divergence set entirely, because a divergence is green-off-and-red-on. The invalid-timestamps case does exactly that whenever its flake fires seam-off. So check which list a case is in before deleting an entry the lane called stale. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…so it is intermittent The same lane, the same code, a later push on the same PR: the whole streams integration suite passed including the class that stalled. Recorded because a green re-run is a data point about the rate and nothing at all about the cause, and this note exists so the next green one is not mistaken for a diagnosis. What it does change is the priority of the experiment the note already names. A standing break would be reproducible on demand; an intermittent one is exactly the case where re-running until green is tempting and useless, so the core-count arm - the same runner at forkCount=1, holding the machine fixed and varying only the load - is now the only thing that will settle it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…iliation branch Merges `feats/ks-streams-error-surfacing` (#395) into `feats/ks-streams-reconciled`, the branch on which the sibling rungs of #255 are brought together and the dispatch default is decided against one measurement of the whole. Everything semantic auto-merged: the typed control-flow passthrough and the settle wait, the commit fence on a pending failure, the derived-occupancy memory bound, `PcDispatchSwitch`'s new `BACKPRESSURE_PROPERTY`, the regenerated patch, the README's switch table, the pom's module description, two `docs/solutions/` write-ups and the reserved-flip note. Verified rather than assumed - `PcDispatchSwitch.java`, `README.md` and `pc-streams.patch` are byte-identical to that branch's versions after the merge. One conflict, on `docs/inflight/ci-streams-classifier-artifacts-escape-the-cache-warming-job.md`, and neither side was simply taken. This branch carries the post-fix version, because #379's warm step is now in the workflow here; the sibling carries the pre-fix version enriched with what its own CI runs measured. The resolution keeps the post-fix framing and adds the three durable findings the sibling established, which outlive the fix and describe the shape of this defect class wherever it appears again: it takes down every lane that builds the module rather than one, it is per-run rather than per-branch, and the attribution was settled by a markdown-only commit reproducing it rather than by argument. The workflow's own warm step is still present and is still the durable owner of the mechanism. Ordinary merge. No history rewritten, and the sibling branch is untouched.
…ion branch Merges `feats/ks-streams-stream-time-punctuation` (#396) into `feats/ks-streams-reconciled`, beside the error-surfacing rung it was cut in parallel with. This is the merge the two siblings were written to make possible, and it is where their shared files meet. ## The patch was RECONCILED AT THE SOURCE, never hand-edited `bin/regen-patch.sh`'s own header prescribes this and it was followed exactly. The patch showed 36 conflict blocks, almost all of them `@@` offset arithmetic; the Java underneath had **one**. - `process-sources` produced a pristine tree, and each side's patch was applied to its own fresh copy of it - #394's as the base, #395's as ours, #396's as theirs. - `diff -rq` against the base named the files each side actually edited: `StreamTask.java` on both, `KStream.java` on the stream-time side alone. The second was taken wholesale; the first was merged with `git merge-file --diff3`, which reported a single conflict - two adjacent field declarations, `pcPausedPartitions` and `pcPunctuatorWarnings`, resolved as the union they are. - The merged tree was copied over `target/kafka-patched` and `bin/regen-patch.sh` run with no maven invocation in between, which is the foot-gun that silently reverts the tree. - **Verified by content, not by hunk count.** No line either side's patch added is missing from the regenerated one except where the other side deliberately rewrote it, and every removal on both sides survives. 117 hunks and 121 hunks in, 121 out. The deliberate rewrites, checked one at a time rather than accepted as noise: the stream-time rung replaced `pcPrepare`'s `Runnable` return with `PreparedRecord` so the extracted timestamp travels with the work, and rewrote the `@DoNotCall` join strings because the reason is no longer "the clock is stuck"; the error-surfacing rung replaced the `StreamsException` wrap with the typed passthrough and `pollFailure()`'s `Failure` type. The merged `StreamTask` carries both - `pcClassifyFailure` and `PcTaskDispatcher.Failure` from one, `PreparedRecord` preparation from the other. ## The Java, and one duplication removed rather than left `PcTaskDispatcher` merged as six unions, each verified: the imports, the pump's two new locals, the per-record occupancy count now sitting beside a `PreparedRecord` rather than a `Runnable`, the completion drain releasing the stream-time hold and reading the partition before handing the container over, the two blocks of new query methods, and the close path doing both its clears. No line either parent added is absent from the result. `PcTaskDispatcherTest` needed real work rather than a merge: the error-surfacing rung's new cases were written against the `Runnable`-returning preparer, so thirteen of them are re-expressed through `PreparedRecords.prepared`, the shared helper the stream-time rung added for exactly this. The merge also produced two no-op preparers differing only in the case of one letter; the newer one survives and the four call sites of the other are re-pointed, because a merge that leaves two spellings of one helper is how the next reader adds a third. ## The trigger chain, reconciled into one statement Three rungs wrote into `PcDispatchSwitch`'s javadoc, each describing the chain as it stood when it was written, and merging them naively would have left a class documenting two contradictory presents. It now states the history once: refusal closed by #389, revival by #394, exception surfacing by #395 - each with a seam-on measurement and a control arm - and punctuation closed beside them by #396, which was never one of the triggers and is recorded as such. What remains is named and explicitly not a fourth trigger: Kafka's private processing-threads parameter combination, out of scope since the seam landed. **The default is NOT moved by this commit.** Every rung declined to move it on its own evidence and every one was right to; the flip is a decision for the reconciled tree measured as a whole, which is what this branch exists to do. The streams module compiles, main and test, on JDK 17. Ordinary merge, no history rewritten, and the sibling branch is untouched.
Merges `test/ks-streams-evidence-suite` (#398) into `feats/ks-streams-reconciled` - the seam-on divergence lane and its classifier, the broker-backed laws with their controls, and the realistic-domain benchmark with its noise floor. This is the rung whose whole purpose is to be run against the reconciled tree. It deliberately touched no main Java and no patch, precisely so that the two sibling rungs could change dispatch semantics underneath it, and that held: the merge conflicted on one file only. `BrokerStreamsIntegrationTest`, the shared broker base class, gained members from both sides at the same two points - the error-surfacing rung's `sleep` cost simulator, and the evidence rung's `consumeByKey` and `flatten` readers, the pair the duplicate-code check told it to hoist. Every conflict is a pure addition on both sides and every one is resolved as the union; the import block was re-sorted and its one genuinely duplicated entry collapsed rather than left doubled. `README.md`, `pom.xml` and the invalid-timestamps flake ledger auto-merged. The streams module and the example module compile, main and test, on JDK 17. Ordinary merge, no history rewritten, and the sibling branch is untouched. The lane's own ledger entries are NOT edited here - that is the next commit's job, because deleting an attribution and re-running the lane is the acceptance test for whether this merge preserved the fixes it claims to.
…n branch Merges `feats/ks-streams-example` (#391) into `feats/ks-streams-reconciled`. That branch sits on the seam rung rather than the lifecycle one, so this merge is where the demo meets refusal, lifecycle, error surfacing, stream time and the evidence suite for the first time. Everything auto-merged except the workstream signpost, `docs/inflight/branch-ks-streams-workstream.md`, where two rungs each struck their own entry from the same "still off master" list - the evidence suite removed itself, and the example module recorded its arrival. Resolved as the union: the list now names only what genuinely remains, and both rungs' paragraphs stand. The whole reactor compiles, main and test, on JDK 17. Ordinary merge, no history rewritten, and the sibling branch is untouched. **The example's framing is deliberately NOT changed here.** Its README and its run both say the demo sets the dispatch flag explicitly because the default is off, and that is still true at this commit - the flip is a measurement this branch has not yet taken. If the measurement moves the default, the framing moves with it, in the commit that moves it.
…lt stays OFF The decision every rung of #255 deferred, taken here, on the branch it was deferred to. The answer is NO FLIP, and the measurement is the point rather than the answer. ## What was run `bin/ci-streams-seam-on-evidence.sh` on the reconciled tree: Apache Kafka's own suite twice against the patched classes, seam off as the control arm and seam on as the measurement, report directories deleted first so a previous run's XML could not read as this one's. The control arm was **clean** - no failures at all, so no flake had to be named and the differencing had something trustworthy to difference against. Counts are deliberately not written here; they move with the branch, the Kafka version and the seam, and the lane re-derives them. ## The acceptance test, and it ran red first, which is what makes it a test The two attributions the sibling rungs owned - `stream-time-never-advances` and `exception-type-lost-in-the-worker` - were **deleted before the run**, so a fix the merge had lost would match no mechanism and be reported as unexplained rather than keep its old label by inertia. The first run exited 1 with three unexplained divergences. That is the lane doing its job. ## The control arm that told a lost fix from a mechanism that only partly closed A failure seen on one branch says nothing about that branch, so the same seam-on suite was run at #396's own tip, in a **detached** worktree - the sibling branch itself is not touched. That tip carries the stream-time work and not the error-surfacing work, so the single term that differs from the reconciled tree is the reconciliation. **Nothing was lost.** `shouldReinitializeRevivedTasksInAnyState` fails on all three parameters there and passes on two here; `shouldRecordBufferedRecords` fails there and passes here; the two `StreamThreadTest` punctuation cases pass in both. Every case is the same or better than at the sibling's own tip, so both rungs' fixes survived the merge. **And the residue changed mechanism rather than surviving unchanged.** `shouldPunctuateOnceStreamTimeAfterGap` moves from zero punctuations to six of stock's seven, and `shouldRespectPunctuateCancellationStreamTime` fails a different assertion than it did. Reading those as "still `stream-time-never-advances`" would have hidden the whole effect of the stream-time rung. They are re-attributed to a new class, `stream-time-lags-in-flight-work`, and `shouldReinitializeRevivedTasksInAnyState` to `processing-threads-mode`, which is what its remaining parameter has always been. The lane then exits 0 with every divergence explained and no stale entry. ## Why that is a fourth named trigger and not a shrug The lag IS the guarantee: the low-water mark never passes a record still in flight, which is what stops a punctuation closing a window over work still inside the processor chain. There is no code fix here and the design would be worse without it. What is wrong is the record. The README, `CONCEPTS.md` and the punctuation note all pin the mark **overtaking** stock as the known divergence and say nothing about it lagging - and lagging is what Kafka's own suite has now measured. Flipping the default while half of a two-directional divergence is undocumented would ship a config whose docs are half true. So the default stays off, and what closes the trigger is the lagging direction being recorded to the same standard as the overtaking one: a bound on how far behind the mark can be, or an honest statement that none is known. This is the fourth time the measurement has named the next reason, and the first time it did so from the reconciled whole rather than from one rung. ## Where it is written down The three sites that move together, re-pointed rather than left stale: `PcDispatchSwitch`'s javadoc, which owns the decision and now carries the whole chain in one statement; the module README's "Turning it on, and why it is off"; and the pom's module description. `docs/inflight/core-streams-punctuation-diverges-in-three-measured-ways.md` gains the measurement - it already listed this lag as something the divergence work owed, so the finding lands in the note that predicted it rather than in a second one. The seam-on ledger gains the two new classes, the control-arm table, and the deletion of `shouldRecordBufferedRecords` on that evidence rather than on one stale-report line. `docs/inflight/streams-dispatch-default-flip-is-reserved-until-the-rungs-reconcile.md` is **deleted**: its own "delete when" was the default being re-decided against a fresh seam-on measurement on the reconciled branch, whether or not it moved. It has been, so the reservation is discharged and the position it was holding open now lives in the chain.
…akes it met No code changes. Records what running the reconciled module actually produced, so the next reader inherits the sightings rather than re-diagnosing them. **What is green.** The module's own unit suite; the seam-off upstream oracle; all 35 broker-backed integration arms run alone against a real broker - the rebalance arm, both memory-bound arms, the five punctuation arms, the two proof arms, the commit-frontier crash-restart law, the shadowed stock control and the wake-on-work shutdown arm. `bin/check-all.sh` exits 0, every runnable gate passing. **Two flakes met and neither chased.** - `StreamThreadTest.shouldLogAndRecordSkippedRecordsForInvalidTimestamps[3]` - the ledgered thread-name flake, failing once in three runs of the upstream execution on this tree while the other two, taken as the evidence lane's control arm, were entirely clean. A pass/fail pair on an unchanged tree in one session, which is the strongest shape that ledger collects. - Seven `TransactionAndCommitModeTest` timeouts and one `PartitionStateCommittedOffsetIT.committedOffsetRemoved[3]` `RebalanceInProgressException`, in `parallel-consumer-core`, from a `-am` build that stood core's broker containers up alongside the streams module's. Attributed to the load-tightness family rather than to this branch, on a check rather than an assertion: this branch changes zero files under `parallel-consumer-core`, and re-running the streams module alone with core installed was green in every arm. The `committedOffsetRemoved[3]` sighting is the same case and the same exception as a row already in that ledger. Nothing was loosened, shortened, serialised or re-run into green.
#271 carries eleven unresolved review threads. Nine of them either close against a named rung of this stack or are answered as a documented contract. **Two do not**, and a thread is the worst place for a live defect to sit: it disappears with the PR. Both are the same shape. The patched `commitNeeded()` short-circuits to `pcAwareCommitNeeded()` whenever a dispatcher is present, and that helper asks one question - does PC hold uncommitted work. Stock asks two more, and both of them are real: an explicit `context().commit()` sets `commitRequested`, which nothing on this path reads, and a poll batch of nothing but control markers advances the consumer position through a sweep this path cannot run because `consumedOffsets` is gone by design. **Read against the reconciled tree rather than against the state the threads were written on**, which matters because four rungs have changed this area since - the refusal envelope, the task lifecycle, error surfacing and stream time. `grep -n 'commitRequested' parallel-consumer-streams/src/main/patch/pc-streams.patch` is the check that the first one still stands: it finds the declaration and the javadoc calling it "the one sanctioned cross-thread commit-state write", and no read. The note also says what settling them is NOT: a one-line `|| commitRequested` is the same shape as the `hasUncommittedWork() || commitNeeded` candidate the stream-time rung measured and rejected, because `validateClean()` reads the same answer and would start throwing on a clean close after a punctuate-only interval.
The self-reference gate flags any mention of the current branch. This one names the branch outright rather than saying "this branch", and the claim beside it is a command a reader can re-run against whatever that ref resolves to, so the sentence still reads correctly after the merge. Attested rather than reworded.
… the gate reads The `post-merge: checked` marker is line- or block-scoped, so a comment two lines above the sentence it vouches for silences nothing. Wrapped in `checked-begin`/`checked-end` instead, which is the form the rest of this directory uses. `bin/check-branch-self-reference.sh` now exits 0 and `bin/check-all.sh` passes every runnable gate.
…ngs, and what those runs prove The Unit lane of #388 and of #389 both failed on `StreamThreadTest.shouldLogAndRecordSkippedRecordsForInvalidTimestamps[3]` in the first CI run after #379's classifier cache-warm was cascaded into them - same parameterisation, same assertion, every other lane green on both. Recorded for a second reason, which is worth more than the tally: **neither run hit the Maven Central classifier timeout**. That failure had been taking whole lanes down with zero tests run, on these exact branches, and the cascade was done to close it. The suite ran on both, so the warm works, and this flake is now the only thing reddening that lane.
`check-branch-self-reference` flagged the two places this note cites #271. It is right to ask and the answer is that the citation is the point: the note exists precisely because a review thread stops being readable when its PR closes, while a landed PR is a permanent link. The sentences name the PR outright rather than saying "this PR", so they read the same afterwards. Attested with `checked-begin`/`checked-end` rather than reworded.
… 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.
Dependency ReviewThe following issues were found:
License Issues.github/workflows/streams-seam-on-evidence.yml
parallel-consumer-examples/parallel-consumer-example-streams-pc/pom.xml
OpenSSF ScorecardScorecard details
Scanned Files
|
5 tasks done
❌ 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 | 115 | 91 | :face_with_raised_eyebrow: +24 |
| Duplicated lines | 1656 | 1294 | :face_with_raised_eyebrow: +362 |
| Duplication | 1.15% | 0.99% | :face_with_monocle: +0.16% |
| Rule | Limit | Status |
|---|---|---|
| Max duplication | 2% | ✅ Pass (1.15%) |
| Max increase vs base | +0.1% | ❌ FAIL (+0.16%) |
⚠️ 24 new clones introduced
- 16 lines:
parallel-consumer-examples/parallel-consumer-example-streams/src/test/java/bz/stub/parallelconsumer/examples/streams/integrationTests/StockBaselineFixtureTest.java:69<->parallel-consumer-examples/parallel-consumer-example-streams/src/test/java/bz/stub/parallelconsumer/examples/streams/integrationTests/StockStatefulBaselineFixtureTest.java:95 - 10 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/RebalanceUnderPcDispatchTest.java:14<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/StreamsBenchmarkHarness.java:17 - 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 - 7 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PunctuatorEffectSurvivalTest.java:14<->parallel-consumer-examples/parallel-consumer-example-streams-pc/src/main/java/bz/stub/parallelconsumer/examples/streams/pc/ArmRunner.java:8 - 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 - 8 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PostCommitCheckpointGapTest.java:9<->parallel-consumer-examples/parallel-consumer-example-streams-pc/src/main/java/bz/stub/parallelconsumer/examples/streams/pc/ArmRunner.java:7 - 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 - 11 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PcDrivenStreamsDispatchTest.java:7<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/ShadowedStreamsControlTest.java:6 - 14 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PcDrivenStreamsDispatchTest.java:17<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/ShadowedStreamsControlTest.java:15 - 9 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PcDrivenStatefulProofTest.java:10<->parallel-consumer-examples/parallel-consumer-example-streams/src/test/java/bz/stub/parallelconsumer/examples/streams/integrationTests/StockStatefulBaselineFixtureTest.java:11 - 11 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PcDrivenStatefulProofTest.java:22<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PcDrivenStreamsProofTest.java:13 - 8 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PcDrivenProofSupport.java:9<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/ShadowedStreamsControlTest.java:7 - 70 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PaymentAuthorisationBenchmarkTest.java:10<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/WorkloadMatrixBenchmarkTest.java:9 - 56 lines:
parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/PartitionScalingBenchmarkTest.java:1<->parallel-consumer-streams/src/test/java/bz/stub/parallelconsumer/streams/integrationTests/WorkloadMatrixBenchmarkTest.java:1
...and 4 more
Powered by astubbs/duplicate-code-cross-check
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| // A fresh application id per arm. Reusing one would resume from the previous arm's committed | ||
| // offsets, and the second arm would sit there processing nothing. | ||
| props.put(StreamsConfig.APPLICATION_ID_CONFIG, armName + "-" + sequence + "-" + System.nanoTime()); | ||
| props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, broker.bootstrapServers()); |
There was a problem hiding this comment.
parallel-consumer-examples/parallel-consumer-example-streams/src/test/java/bz/stub/parallelconsumer/examples/streams/integrationTests/StockBaselineFixtureSupport.java:177
|
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## feats/ks-streams-task-lifecycle #454 +/- ##
=====================================================================
- Coverage 82.49% 73.51% -8.98%
- Complexity 1451 1479 +28
=====================================================================
Files 95 105 +10
Lines 5066 5717 +651
Branches 497 564 +67
=====================================================================
+ Hits 4179 4203 +24
- Misses 692 1318 +626
- Partials 195 196 +1
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
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 #395
depends on #396
depends on #398
depends on #391
Description
This is the reconciliation rung of the Kafka Streams stack: the four sibling rungs merged together, verified as one tree, and the dispatch default decided against a measurement of the whole rather than of any one rung. It sits on the task-lifecycle rung (#394) and contains, by ordinary merge commits, error surfacing (#395), stream time and punctuation (#396), the evidence suite (#398) and the example module (#391). The diff against the base is therefore mostly theirs; what is new here is small and is the point:
8f7d4c884). Every sibling rung declined to flipPcDispatchSwitchon its own evidence and reserved the decision for the merged module, because the seam-on numbers move with each of them. Merged, Kafka's own suite was run through the evidence lane twice - seam off as the control arm, seam on as the measurement - and two punctuation cases still diverge:StreamTaskTest.shouldPunctuateOnceStreamTimeAfterGapproduces six of stock's seven punctuations, andshouldRespectPunctuateCancellationStreamTimefails a different assertion than it did before stream time worked at all. That is the low-water mark doing its job - it never passes a record still in flight - and it is a divergence in the direction the documentation did not state: the README andCONCEPTS.mdpinned the mark overtaking stock and said nothing about it lagging. A default flipped on while half of a two-directional divergence is undocumented ships a config whose docs are half true, so it became the fourth named trigger.PcDispatchSwitch's javadoc owns the chain of reasons end to end;docs/inflight/core-streams-punctuation-diverges-in-three-measured-ways.mdis where the current position lives, and the divergence triage note gained thestream-time-lags-in-flight-workclass so the lane names it rather than re-diagnosing it.5f4e6de0f). Of the eleven open threads on feat(streams) astubbs#255: give a Kafka Streams topology PC's per-key concurrency #271, nine close against a named rung or are answered as a documented contract. Two survive all of it and are written down indocs/inflight/core-streams-two-commit-signals-the-pc-path-cannot-see.md: an explicitcontext().commit()sets avolatile commitRequestedthat nothing on the PC path reads, and a poll batch of nothing but control markers advances the consumer position without any signal reachingcommitNeeded(). Both were read against the reconciled tree, not the state the threads were written on. Neither is a one-line fix - the|| commitRequestedshape is the same one the stream-time rung measured and rejected.a7855e2ea,234e9cd24,d9d56acad). The whole module'stestandverifyphases were run on the merged tree; the invalid-timestamps flake was met on both cascaded rungs and is recorded with what those runs prove, a load-tightness sighting is recorded, andPostCommitCheckpointGapTest's post-commit arm is recorded as passing only beside its siblings - which is why it is red on feat(streams) astubbs#255: advance stream time on the PC path, and decide what to do about punctuation #396 alone and is expected to be.Why this rung exists as a PR at all. #271 was retargeted onto this branch as its base, so this branch is on the merge path to master and nothing else reviews the three decisions above. Without a PR here, the four leaves merge down through #394 and this branch - the one carrying the default decision - never merges anywhere.
Not in this PR
docs/inflight/branch-ks-streams-workstream.mdsays the first two rungs "have landed"; they have not (build(streams) astubbs#255: patch Kafka Streams at build time, with Kafka's own suite as the oracle #379 and feat(streams) astubbs#255: run a Kafka Streams topology on PC per-key concurrency, behind an opt-in switch #388 are open, andorigin/masterhas no streams module). That wording arrived from a lower rung (8aedb9fd5) and is corrected there, not here.Verification
README.md,pom.xml,PcDispatchSwitch.java(+60 / -20) - plus thirteendocs/inflight/notes.parallel-consumer-streams/target/seam-on-evidence-report.txtafterbin/ci-streams-seam-on-evidence.sh, re-derived rather than quoted, per the triage note.StreamThreadTest.shouldLogAndRecordSkippedRecordsForInvalidTimestamps[3]flake, or a Maven Central read timeout that the cache-warm commit below this rung fixes.Checklist
parallel-consumer-streams/README.md(stream time lags as well as overtakes; the default's fourth trigger),PcDispatchSwitchjavadoc, and thedocs/inflight/notes named abovedocs/features/- N/A - the module is not published and the seam defaults off; there is no user-facing feature to document until a rung makes it publishabledocs/inflight/working note (pr-/branch-) started at the PR's first commit -docs/inflight/branch-ks-streams-workstream.mdis the workstream's note and was updated on this branch; the reconciliation's own state lives incore-streams-punctuation-diverges-in-three-measured-ways.mdandtest-streams-seam-on-divergence-triage.md, both edited at the reconcile commitce-simplifyandce-code-reviewlocally - N/A - the rung's own code change is 60 lines of javadoc, README and pom; asking for@claude review thison the PR instead🤖 Generated with Claude Code
https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ