Repository navigation
Conversation
`State` described the controller's lifecycle but lived in `io.confluent.parallelconsumer.internal`, so nothing could be exposed publicly without leaking an internal type. Move it to `io.confluent.parallelconsumer` unchanged - same constants, same `getValue()` ints, which are a published contract through the `pc.status` and `pc.poller.status` gauges. Add `State#isRunningOrPaused()`: true for UNUSED/RUNNING/PAUSED, false for DRAINING/CLOSING/CLOSED. Named after what it tests rather than "healthy", following `KafkaStreams.State#isRunningOrRebalancing()` - a method name travels into call sites and log lines where the caveat that RUNNING says nothing about progress does not. BREAKING: `io.confluent.parallelconsumer.internal.State` moves package. An enum has no source-compatible shim, so this is deliberate and taken before 0.6.0.0 ships. The type was effectively unreachable - there is no accessor anywhere that returns one. Register the move in `bin/check-copyright-headers.sh`'s rename list so the moved file is still recognised as upstream-derived.
…g its setter Prerequisites for reading health from a caller thread. `state` and `failureReason` were both plain fields. `state` is written by at least three threads and `failureReason` by the control thread and `closeOnException`, while a health check reads both from a fourth with no happens-before edge. The visible failure mode is specific: a reader sees a fresh `RUNNING` beside a stale null cause, and reports a consumer that has already recorded a fatal error as healthy. Both are now `volatile`. Lombok's bare `@Setter` on `state` generated a *public* `setState`, while no getter existed - so a caller holding the concrete type could force the instance to `CLOSED`. Narrowed to `AccessLevel.PACKAGE`. Transitions in this class assign the field directly; the setter is only used by a test in the same package. BREAKING: `AbstractParallelEoSStreamProcessor#setState(State)` is no longer public. It was a Lombok default rather than designed API, and shipping a public read surface beside an unguarded write hole would be incoherent. Guarded by reflection rather than ArchUnit - the shared convention suite uses `ImportOption.OnlyIncludeTests`, so a rule about production modifiers would pass vacuously against an empty set.
Carries the control-loop run state, the broker-poller run state, the failure cause as an `Optional`, and one derived verdict `isHealthy()` for an orchestrator to act on. Three things the type is deliberate about: - A present failure cause overrides the state. A control loop that died can leave the state field untouched, so state alone would report a crashed instance as healthy. - The poller state is diagnostic and excluded from the verdict. The two states diverge in entirely normal operation - `pauseIfRunning()` moves only the control loop, and before `poll()` the control loop reads UNUSED while the poller already reads RUNNING. Folding it in would manufacture false alarms. - The verdict is liveness-scoped, and the javadoc says so: it answers "does this need restarting", never "is this making progress". A consumer processing nothing still reports RUNNING and still reports healthy. Progress is observed through the `pc.*` meters. Both states are `@NonNull` so an incomplete snapshot is rejected at `build()` rather than surfacing as an NPE when the verdict is computed.
Closes the cast. Building a Spring Boot Actuator health indicator previously meant downcasting `ParallelConsumer` to `ParallelEoSStreamProcessor`, because the only health-adjacent method on the interface was `isClosedOrFailed()` and the failure cause was declared on the concrete class. `getHealth()` returns a `PCHealth` snapshot from the interface, so no cast is needed. What is genuinely new is the detail, not the up/down answer: the typed run state, the failure cause, and - because the two are reported together - the ability to tell a clean shutdown from a crash, which `isClosedOrFailed()` returns `true` for alike. Added as a `default` method so no third-party implementor breaks; there is no binary-compatibility gate in this build to catch that. The default derives what it can from `isClosedOrFailed()`, and its javadoc says plainly that those states are derived rather than observed and that its empty failure cause carries no clean-versus-crash meaning - without that, the fallback would report a crashed instance as a clean shutdown, inverting the distinction the API exists for. The override reads each volatile field exactly once into a local, failure cause first, so a snapshot cannot contradict itself; worst case it reports a failure beside a slightly stale state, never a healthy verdict for an instance that had already failed. `BrokerPollSystem` gains a package-private `getRunState()` read only - its initialiser and transitions are untouched, so `pc.poller.status` gauge values are unchanged.
Adds a Health Checks section covering the cast-free `getHealth()` call, what each field of the snapshot means, and - explicitly - what the verdict does not claim: healthy means not shut down and not failed, never "making progress". A consumer processing nothing still reports RUNNING and still reports healthy, so the section points at the Micrometer meters for progress rather than letting a reader assume the health check covers it. The example is a tagged region in CoreApp, following the convention already used for the close modes: it compiles, so it cannot silently go stale, and it declares the variable as the `ParallelConsumer` interface - which is the whole point of the feature and would stop compiling if that stopped being true. CONCEPTS.md gains run state, health verdict and failure cause, using the control-loop / broker-poller vocabulary already defined there. README.adoc is generated - regenerated via `./mvnw process-sources`.
…ng changes The health-check API answers #126 but only the interface half of #157 - the sharper "is it stuck?" question is not answered, and #157 stays open as its home. Records why, because the reasoning is the expensive part to rediscover: `lastCommitTime` is not a progress marker (commits are only attempted when the work manager is dirty, and only success sets dirty, so an all-failing or idle workload freezes it on a healthy consumer); every stall documented in this repo happens at `state == RUNNING`; and our own ProgressProbe still has open RED calibration. Names #222's offset-gap as the candidate mechanism and PCHealth as the extension point it would land on. Also records the residual exposure plainly: the verdict reads healthy throughout the 120-second zero-poll incident in docs/solutions. docs/refactoring.md gains the two release-gated follow-ups, a new subsection for the two breaking changes already taken, and drops the "rename the enum" bullet - that landed long ago, and leaving it would have had major-release prep budget for finished work against the very type this change relocates.
The plan doc for this branch, including the doc-review corrections that caught three build-breaking assumptions before implementation started.
The snapshot read the failure cause first, on the stated reasoning that it could then never report a healthy verdict for an instance that had already failed. Reviewing the actual write order shows that is backwards. Both failure paths write `failureReason` *before* transitioning the state - `closeOnException` sets it then closes, and the control loop's error handler sets it then calls `doClose`. So reading the cause first admits this interleaving: read cause just before the failure lands (it is still null), then read state just after the transition (CLOSED). The result is CLOSED with no failure cause - which is precisely the shape this API defines as a clean shutdown, reported for a crash. Reading the state first closes that window. If the read observes a state the failure produced, the happens-before edge on that volatile read guarantees the cause written before it is visible too. The window is narrow and neither order can be exercised deterministically from a test, so this is argued from the write order rather than reproduced. Found by the pre-review simplification pass, which flagged the original javadoc's causal claim as unsupported - it was worse than unsupported.
The hand-written `getRunState()` is what `@Getter(AccessLevel.PACKAGE)` generates, and the field two lines below already uses plain `@Getter`. Its explanation moves onto the field, which is where it belongs. Also restores the import block ordering disturbed by the State package move, and documents on `State#isRunningOrPaused()` that its unmapped-state throw would surface out of a health check - so a new constant has to be classified in the same change that adds it.
Three independent reviewers converged on this, and it is the defect that mattered: `getHealth()` read only `state` and `failureReason`, dropping the control-thread-liveness term `isClosedOrFailed()` carries. The control task's handler is `catch (Exception e)`, so a Throwable that is not an Exception - OutOfMemoryError, StackOverflowError, a LinkageError from lazy classloading - terminates the thread without writing either field. State stays RUNNING, no cause is recorded, and `isHealthy()` returned true forever for a consumer that will never process, commit or close again. The new API was therefore strictly weaker than the boolean it was built to replace, on exactly the failure a liveness probe exists to catch: a subsystem dead inside a JVM that is still up. A Kubernetes probe wired to it would have kept that pod in rotation. Reconciled against `areMyThreadsDone()` - the same term `isClosedOrFailed()` uses - so the verdict can never contradict it. `controlThreadFuture` becomes volatile, since the health path now reads it from a caller thread. Two tests, both of which failed before the fix: one kills the control loop with an Error, one asserts across the lifecycle that `isClosedOrFailed()` implies not-healthy. Also from the review: - `PCHealth.isStateObserved()` marks the coarse interface fallback. Without it, a crashed third-party implementation reported CLOSED with no cause - byte-identical to the clean shutdown this API defines - and a caller holding only the snapshot had no way to tell. The caveat now travels on the value, not just in supertype javadoc. - The poller state is documented as a lifecycle flag, not a liveness signal: it reads RUNNING both before the poll thread starts and after it dies. - The concrete `getFailureCause()` is deprecated toward the snapshot's Optional one. They share a name and not a null contract, so `health.getFailureCause() != null` is always true, and silently so. - `toString()` now carries the verdict, which is what an operator actually reads in a log line. - The setState guard rejects `protected` too, since a public subclass in another package would inherit it onto the user-facing type. - The poller state is now asserted after a close; every other assertion expected its own initialiser, so wiring the accessor to a constant would have passed the suite. - The third-party stub carries a warning that implementing a new method there to fix a red build deletes the guarantee it guards.
…uals The plan's U4 step 3 specified reading the failure cause before the run state - the reverse of what shipped, and the order that reintroduces the crash-reported-as-clean-shutdown defect. A future reader reconciling the code to the plan would have put the bug back. Corrected, with the reasoning, plus the control-thread reconciliation step the review added. Review residuals that were judged real but not actioned go in the inflight ledger rather than a new file, per the one-place rule.
Dependency Review✅ No vulnerabilities or license issues or OpenSSF Scorecard issues found.Scanned FilesNone |
|
Claude finished @astubbs's task in 6m 35s —— View job Claude's review of #226Todo list
SummaryThis is a well-scoped, carefully reasoned change. No bugs found; comments below are minor/optional. What I verified directly:
One thing I could not verify: the claim "two tests failed before the fix" (for Minor / optional notes
The defect-review section in the PR description (dead-control-thread liveness gap, and the read-order/write-order ordering bug) both check out against the current diff — i.e., both are actually fixed in what's on this branch now, not just described as fixed. |
✅ Duplicate Code ReportTwo engines run in parallel for cross-validation. Each has its own thresholds tuned to its baseline - the real safety net is the per-engine "max increase vs base" check. ✅ PMD CPD
No new clones introduced by this PR. ✅ jscpd (language-agnostic)
No new clones introduced by this PR. Powered by astubbs/duplicate-code-cross-check |
✅ SpotBugs ReportNo bugs found (new bugs only — baseline from base branch excluded). |
`crashedControlLoopIsNotHealthyAndReportsTheRecordedFailure` errored in CI while passing locally, three retries in a row. The test body was fine; teardown threw. The test awaited `getFailureCause() != null`, but the cause is written *before* the close transition - the same ordering this PR's `getHealth()` read order depends on - so the await returns with the close still in flight. The test base deliberately skips closing an already-errored PC, and a half-closed one fails that guard, so `@AfterEach` calls `close()` and `waitForClose` rethrows the recorded failure out of teardown. Awaiting `isClosedOrFailed()` is both the stronger assertion and the condition the base's guard actually tests, so it is deterministic rather than timing-dependent. Local runs happened to finish closing first, which is why only CI saw it.
|
Claude finished @astubbs's task in 3m 57s —— View job Claude's review of #226Todo list
SummaryOnly one commit landed since my last review of this PR: What changed: Why this is the correct fix, verified by reading the code directly:
Verified by running the tests directly (not just inspection): No issues found. This is a solid, minimal fix with a commit message that accurately explains the root cause. My prior "no bugs found" assessment for the rest of the PR still stands — this commit doesn't touch any of the production code or other tests I reviewed previously. |
|
Thanks — the independent re-derivation of the read/write ordering and the check that 1. 2.
3. 4. Breaking-change gating — no action needed. Agreed. On the one thing you couldn't verify (that the two tests failed before That gap is also why the reconciliation exists at all: three independent reviewers converged on One thing worth naming that no reviewer caught, since it bit CI and not local: |
Review suggested `@Deprecated(since = "0.6.0.0")`. That cannot compile here - `since()` was added to `java.lang.Deprecated` in Java 9 and core targets Java 8 bytecode via Jabel with `--release 8`, so javac rejects the member outright. Recorded in the `@deprecated` Javadoc tag instead, which is what the Javadoc tool surfaces anyway.
|
Claude finished @astubbs's task in 1m 38s —— View job Claude's review of #226Todo list
SummaryOnly one commit landed since my last review: What changed: Context: this directly answers my prior minor note suggesting Verified directly:
No issues found. This is a trivial, correctly-reasoned fix to a documentation-only gap from my previous review pass. My prior "no bugs found" assessment for the rest of the PR still stands — this commit doesn't touch any production logic, tests, or other files. |
…ing checkout dropped
…hanges) Directory move only, so every path is 100% similar and git's exact-rename detection cannot fail on it. The content edits follow in the next commit. Generated by bin/rename-packages.sh.
Text edits only. No file moves in this commit, so it cannot dilute the rename detection in its parent. Generated by bin/rename-packages.sh.
…ttering drafts The first cut of the ladder let three kinds of overstatement through, all named in review: - 'implemented' now means the implementation intended to ship. A draft that will be replaced is at most a poc, and a draft that is not even the surviving direction does not advance the stage at all - so virtual-threads (#51 carries the upstream draft, not the fork's) drops to idea, and the health-check surface (#226, a working draft not yet accepted as the shipping surface) is a poc. - 'limited-poc' joins the ladder below poc: a slice runs but the central claim is not proven end-to-end. The Streams and Connect previews move there - 'poc' alone read as 'we proved we can do it', which neither spike claims. Java-and-Kafka drops to idea: compiling at 17 is exploration; the Java 11 + Kafka 4 work has not started. - 'planned' now means an implementation plan - the how. The self-tuning plan from #308 decides the what, so it is requirements-drafted. One promotion the review surfaced: the docs site has a drafted design on #316 (explicitly design only - no site is built there), so it moves from idea to planned. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…f the GitHub flag The previous commit defined stage_delivery as mirroring the carrying PR's GitHub draft state. That was wrong twice over: the flag has not been used consistently here, so mirroring it records noise - and if it were consistent, the field would violate 'never write down what a command can answer'. The field earns its place precisely because it records what no command knows: whether the author believes the artifact is ready to land. Reassigned on that judgment: the proxy sidecar (#293) and the docs-site design (#316) are pending-merge; Streams (#271), Connect (#269), the web GUI (#268, blocked on #267), the health surface (#226) and the DLQ requirements (#313) are drafts still being worked. The note records the direction of travel: PRs the data calls drafts ought to become actual GitHub drafts over time, so the flag regains meaning. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…st move its entry The stages block says the PR advancing a track updates its entry's stage in the same change. Until now nothing caught a miss, which is the exact failure mode docs/agent-harness.md warns about: a rule that is merely available rots the field it protects. Three layers, mirroring the issue-ref gate's shape: - roadmap-stage-gate.js (+ 10 unit tests, run first in CI): reads the BASE branch's roadmap, finds entries whose pull_request names this PR, and requires the PR to touch docs/data/roadmap.yaml or opt out with 'roadmap-stage: N/A - <reason>' in its body. Base, not head, deliberately - master is the authority on which entries claim a PR. Dependency-free narrow parse, pinned by a test against the real file so a roadmap rewrite fails the test rather than silently blinding the gate. - A merge-checklist line, so the rule also arrives in every merge-prep prompt via the existing injection hook - covering entries the gate cannot reach (tracking issue only, no pull_request). - pull_request fields added to the three carrying entries that lacked one (health-check-surface -> #226, dead-letter-queue -> #313, docs-site -> #316), so the gate reaches all seven carriers. actionlint was not runnable locally (not in mise); the workflow YAML parses, and CI's own lint covers the rest. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
Process note: this PR is the carrier of the roadmap entry |
…anscript excerpts dropped The owner supplied the agent-written handoff document from the strategy weekend as a cross-reference against the note set. Preserved verbatim under docs/ideation/ with a provenance header (self-dated 29 August - it predates the 30th's exchanges, which is why it says "W2" where the later transcript says "Voice" and carries none of the Dune codenames). Where it and a note disagree, the header says flag it, never silently reconcile. The diff surfaced two whole chapters the pasted excerpts never carried, each now a note pointing at the preserved document for detail. core-partition-virtualization.md: attack partition rigidity from both sides - the intervention hierarchy (internal shards -> move -> extract the pathological shard via an elastic shadow route -> merge -> evolve the topic generation -> only then scale, "do not scale machines because Kafka happened to hash badly"), the partition-placement ceiling as the one ceiling the runtime can remove itself, epoch-anchored routing state riding the frontier protocol, topic generations making partition count reversible, count optimization from real key distributions, virtual records with the full virtualization stack, and the boundaries (vanilla apps keep vanilla semantics; internal Streams topics are the safest first target). core-runtime-services-and-compat.md: distributed Interactive Queries as a first-class service, semantic health (which directly extends #226 and candidates an answer to #157's stall question via the admission states), embedded compatibility facades (Schema Registry with Karapace/Apicurio named as prior art, Consul, Envoy/xDS - "compatibility is a distribution strategy"), CDC to served read models with the replicas-vs-read-models language precision, and backfill/replay as ordinary eligibility. Smaller catches landed on their owners: the W2/Voice naming discrepancy flagged in the glossary for the owner to settle, plus the trademark-clearance backlog item; the 13-stage tutorial curriculum and the canonical-example-must-compose law wired into the executable-progression note; the OSS/enterprise commercial-shape hypothesis and the Observe -> Recommend -> Shadow -> Enforce authority ladder onto the thesis briefing; the architectural-stubs rule and the section-24 seam inventory onto shared-execution-resources; broker-as-resource-graph-node onto the fleet note; and the birthday-card story seed with its explanation rule onto the content series. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016PFA3CXkGc2m32cBJMAWTK
… with its premise corrected Owner ruling, 2026-09-08, on the sweep's proposal 19. The note goes to `deferred - after v6` and keeps the audit as the record. Its central premise was corrected on the way rather than left for the next reader to trip on: `state` is volatile on master today, so the visibility half of the rows that named it is closed; the check-then-set race in pauseIfRunning/resumeIfPaused and the absent per-method contract are what the blocker rests on. #226 no longer carries the volatility change (it is the health-check surface PR) and #51 is closed, so the two bullets that described them as live work are rewritten. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01W511R2TZFAqEiqgS5yogTv
Description
Closes #126 (mirror of confluentinc#71), open since 2021, and answers the interface half of #157.
Building a Spring Boot Actuator health indicator meant downcasting
ParallelConsumertoParallelEoSStreamProcessor, because the only health-adjacent method on the interface wasisClosedOrFailed()and the failure cause was declared on the concrete class.getHealth()now returns aPCHealthsnapshot from the interface, so no cast is needed.What is genuinely new is the detail, not the up/down answer. For a running, paused or cleanly closed consumer the verdict matches what
!isClosedOrFailed()already gave. What is new is the typed run state, the cast-free failure cause, and — because the two are reported together — the ability to tell a clean shutdown from a crash, whichisClosedOrFailed()returnstruefor alike.Deliberately not included: a stall / progress signal
#157 asks the sharper question — can I tell a stuck consumer from a working one? That is not answered here, and #157 stays open as its home. Three findings from this repo's own evidence decided it:
lastCommitTimeis not a progress marker — commits are only attempted when the work manager is dirty, and only success sets dirty, so an all-failing or idle workload freezes it on a perfectly healthy consumer.RUNNING, so the state field cannot carry the signal either.ProgressProbestill has open RED calibration — it has never reproduced a true unbounded stall on master.The Javadoc and README say this out loud rather than implying coverage. Reasoning and the candidate mechanism are recorded in
docs/inflight/parked-health-stall-signal.md.Breaking changes (both before 0.6.0.0 ships)
Statemoves fromio.confluent.parallelconsumer.internaltoio.confluent.parallelconsumer, so the snapshot can expose it without leaking an internal type. An enum has no source-compatible shim, so this is irreversible once published — taken now because the type was effectively unreachable (no accessor anywhere returned one).AbstractParallelEoSStreamProcessor#setState(State)is no longer public. It was a Lombok@Setterdefault rather than designed API, and let any caller holding the concrete type force the instance toCLOSED.Two further changes are release-gated and queued in
docs/refactoring.md.The defect review caught
The first implementation read only
stateandfailureReason, dropping the control-thread-liveness termisClosedOrFailed()carries. The control task catchesException, so anError— OOM,StackOverflowError, aLinkageError— kills the thread without writing either field, andisHealthy()returned true forever for a dead consumer. The new API was strictly weaker than the boolean it replaces, on exactly the failure a liveness probe exists to catch. Fixed by reconciling against the same term, with two tests that failed before the fix.A second ordering defect was caught earlier: the snapshot read the failure cause before the run state, which admits a null cause paired with a
CLOSEDstate — the shape of a clean shutdown, reported for a crash.Checklist
README.adoc), Javadoc,CONCEPTS.md,docs/inflight/,docs/refactoring.mdN/A- no CI runner or workflow files touched