Repository navigation
Conversation
queuedMessagesNotProcessedOrCommittedIfSubmittedDuringShutdown was flaking at 3/45 CI runs, always in PERIODIC_TRANSACTIONAL_PRODUCER, always as [1, 1, 2] where [1, 2] was expected. #101 fixed this test once before, but for the opposite symptom - [2, 2], a commit that never happened - so this is a new fault, not that one returning. The extra commit is correct product behaviour. PC commits highestSequentialSucceeded + 1. Under KEY ordering this test has three independent shards, not two: primeFirstRecord() uses key "key-0" while the test's own records use "0" and "1". The key="1" record therefore completes while key="0" is blocked on v1, marking the partition dirty without being able to advance the base offset, so PC commits offset 1 again with updated incomplete-offset encoding - persisting that completion so a crash does not reprocess it. What was wrong was the assertion. assertCommits routed to KafkaTestUtils.assertCommits' containsExactlyElementsOf for the transactional mode, and to hasSameElementsAs - commented "duplicates are ok" - for the other two. The same call meant an exact sequence in one mode and a set in the others, which is why only param [1] ever failed. The helper's own javadoc claimed "Collects into a set - ignore repeated commits"; the implementation has never done that in this repo's history. No precondition could have fixed this. How many commit cycles occur depends on where the wall-clock commit ticks fall relative to independent shard completions, which is unbounded by design. Collapse a repeat of the previous commit before comparing, keyed on the PARTITION and applied before the genesis trim. Both parts of that are load-bearing, and both were found by review rather than by the original fix: - Keyed on the partition, because a commit history is flattened across partitions. Collapsing on the offset alone merges two partitions that commit the same offset in one round, which would hide a partition that failed to commit at all - detection the pre-collapse code had. - Before the trim, because trimming genesis first turns [1, 0, 1] into [1, 1] and then [1], silently accepting an offset that went backwards through 0. Reordering alone would have been wrong too: it converts the documented genesis race into an assertion for multi-partition histories. Keying on the partition is what makes both safe. The tolerance stays narrow: [1,1,2] passes as [1,2], while [1,2,1] - an offset going backwards - still fails, and [2,2] still fails against [1,2], keeping the #101 regression caught. Producer-side order is still asserted; consumer-side is deliberately left order-insensitive, because that list has no meaningful global ordering, and it therefore does not detect a backwards offset either - now stated plainly rather than implied away. awaitForCommitExact(partition, offset) moves from times(1) to atLeastOnce(). It is a wait, and requiring exactly one matching commit turns it into an assertion about commit-tick timing. Its only caller is processInKeyOrder, which is @disabled, so this is forward-looking rather than part of the fix. Experiment: the control arm swapped the key="1" record into the blocked key="0" shard, leaving the settle waits identical. 15/15 runs produced [1,1,2] with the independent shard present, 5/5 produced [1,2] without it. Two earlier predictions were refuted and are worth recording - a 250ms delay on the independent record did not reproduce it (the record then lands after v1 and the base jumps straight to 2), and a first control was contaminated by gating on a flag set inside the user function while the WorkManager's onSuccess lands later. Rejected: making the helper fully set-wise, which would have matched the stale javadoc but silently weakened 25 call sites; fixing only this test, which would have left the mode asymmetry for offsetsAreNeverCommittedForMessagesStillInFlightSimplest to hit next; and collapsing on the (offset, metadata) pair to catch information-free repeat commits - PartitionState's own javadoc documents that its pessimistic stateChangedSinceCommitStart flag "may cause an additional unnecessary commit on next commit cycle", so asserting on that would reintroduce this same flake class. The new regression test makes the race deterministic by latching the independent key until after the first commit has landed. It failed pre-fix in exactly the one mode that fails in CI and passes in all three now, and it asserts the repeat actually occurred so it cannot degrade into testing nothing. Sweep of the same class: offsetsAreNeverCommittedForMessagesStillInFlightSimplest is enabled and shares the exposure - now covered. processInKeyOrder and offsetsAreNeverCommittedForMessagesStillInFlightLong are @disabled. processInKeyOrderWorkNotReturnedDoesntBreakCommits pins PERIODIC_CONSUMER_SYNC and cannot reach the strict path. OffsetEncodingBackPressureTest, the 4/45 sibling flake, has no CommitMode parameterisation and no assertCommits - different mechanism. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CbFPtaYqqzqCH3gKr85fBg
Both from review on #260. theSameOffsetOnTwoPartitionsIsNotAMergeableRepeat had a lowercase "s" in "same". awaitForCommitExact(int, int) linked to the single-partition collapseRepeatedCommits(List) overload, where the partition-keyed two-argument overload is the one its reasoning actually rests on. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CbFPtaYqqzqCH3gKr85fBg
Dependency ReviewThe following issues were found:
License Issuesparallel-consumer-connect/pom.xml
parallel-consumer-streams/pom.xml
OpenSSF ScorecardScorecard details
Scanned Files
|
✅ Duplicate Code ReportTwo engines run in parallel for cross-validation. Each has its own thresholds tuned to its baseline - the real safety net is the per-engine "max increase vs base" check. ✅ PMD CPD
No new clones introduced by this PR. ✅ jscpd (language-agnostic)
|
✅ SpotBugs ReportNo bugs found (new bugs only — baseline from base branch excluded). |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 5a49271e35
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| + pcCollectedForCommit = pcDispatcher.collectCommitData(); | ||
| + committableOffsets = pcCollectedForCommit; |
There was a problem hiding this comment.
When a worker completes after prepareCommit() evaluates hasCommitDataOutstanding() and flushes the state manager/record collector, but before this later committableOffsetsAndMetadata() call drains completions, collectCommitData() advances the frontier to include that record even though its state/output was not covered by the flush. Committing this newer offset and then crashing can therefore skip the input on restart while losing its unflushed state or output. Freeze the PC commit snapshot before the pre-commit flush, or exclude completions arriving after that flush from the current commit.
Useful? React with 👍 / 👎.
| private static String outcomeOf(Element testCase) { | ||
| if (testCase.getElementsByTagName("failure").getLength() > 0) { | ||
| return "failed"; | ||
| } | ||
| if (testCase.getElementsByTagName("error").getLength() > 0) { | ||
| return "errored"; | ||
| } | ||
| if (testCase.getElementsByTagName("skipped").getLength() > 0) { | ||
| return "skipped"; | ||
| } | ||
| return "passed"; |
There was a problem hiding this comment.
In the CI unit and full-build lanes, -Dsurefire.rerunFailingTestsCount=2 applies to these new executions; Surefire records a test that fails initially and passes on rerun using flakyFailure/flakyError elements, neither of which this parser checks. Such a patched-arm regression is therefore classified as passed, allowing the behavior-neutrality proof to go green after an initial failure. Disable reruns for these control arms or reject the rerun-specific XML elements.
AGENTS.md reference: AGENTS.md:L185-L194
Useful? React with 👍 / 👎.
| One commit: `docs/plans/2026-08-08-001-feat-connect-sink-in-pc-plan.md`, carrying a | ||
| superseded-direction header. No code. Nothing pushed. |
There was a problem hiding this comment.
Once this commit lands, this entry tells every session scanning docs/inflight/ that the Connect work consists of one documentation commit with “No code,” even though the same commit adds the complete U1 module and tests. That makes the cross-branch state immediately false and contradicts the plan's R9 requirement; shrink or rename the note to describe only the genuinely open follow-ups instead of preserving the pre-implementation state.
AGENTS.md reference: docs/inflight/AGENTS.md:L32-L38
Useful? React with 👍 / 👎.
| Affects every module built on the build-time Kafka patch strategy - today `parallel-consumer-streams-spike` | ||
| (`feats/ks-on-pc-spike`), and the Connect work parked in `parked-connect-on-pc.md`. **This blocks | ||
| publishing either of them.** It does not block developing them, because both currently generate patched | ||
| classes locally and gitignore the output, which distributes nothing. |
There was a problem hiding this comment.
The existing docs/inflight/next-fork-packaging-docs-and-licensing.md already owns the Streams-and-Connect packaging, licensing, coordinates, and publication work, so adding this second note creates two live trackers for the same item. They already disagree—the existing note describes Streams as published while this commit disables publication—so a session opening only one inherits incorrect state; merge the new decisions into the existing item rather than tracking the same work twice.
AGENTS.md reference: docs/inflight/AGENTS.md:L6-L13
Useful? React with 👍 / 👎.
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| * the patched {@code suspend()} and commits on the way down, which would hand a simulated crash exactly | ||
| * the repair pass a real one never gets, and park each test repetition on the pool-termination wait. | ||
| */ | ||
| public void abortClose() { |
There was a problem hiding this comment.
Dead code: abortClose() is never called, so the R10 "kill-restart proof" it claims to exist for is never exercised.
The javadoc says this method "Exists for the kill-restart proof (R10)", but a repo-wide grep finds zero callers anywhere (production or test code). CommitFrontierCrashRestartTest — the only test that references R10/crash semantics — starts one KafkaStreams instance and, in its finally block, does an ordinary graceful streams.close(Duration.ofSeconds(30)). It never calls abortClose() and never restarts a second instance.
Failure scenario: a bug specific to the abrupt no-drain/no-report path (e.g. resource leak, bad interaction with a concurrent commit) would ship completely untested, since nothing in the suite ever invokes it. Either wire CommitFrontierCrashRestartTest (or a new test) to actually call abortClose() and restart, or remove the method/relabel the javadoc until that proof exists.
| + if (pcDispatcher != null ? pcDispatcher.hasCommitDataOutstanding() : commitNeeded) { | ||
| // we need to flush the store caches before flushing the record collector since it may cause some | ||
| // cached records to be processed and hence generate more records to be sent out | ||
| // | ||
| @@ -491,6 +548,18 @@ | ||
|
|
||
| case RUNNING: | ||
| case SUSPENDED: | ||
| + // PC spike (astubbs#255, U9): the commit carries PC's frontier and encoded holes, | ||
| + // wholesale (plan KTD-S7). The frontier is the lowest incomplete offset, so a commit can | ||
| + // never cover a record still in flight - loss is impossible by construction, where the | ||
| + // stock path below commits findOffset()'s consumer-position fallback, covering every | ||
| + // polled record. Streams' TopicPartitionMetadata (partition time, processor metadata) is | ||
| + // deliberately not written on this path; stock restarts decode PC's payload to UNKNOWN | ||
| + // and degrade gracefully. | ||
| + if (pcDispatcher != null) { | ||
| + pcCollectedForCommit = pcDispatcher.collectCommitData(); |
There was a problem hiding this comment.
Race window: the flush-gate and the commit-data collection drain the completion queue independently, so a worker's completion can be missed by one and picked up by the other.
prepareCommit()'s flush gate (line 346) calls pcDispatcher.hasCommitDataOutstanding(), which internally drains the completion queue and checks workManager.isDirty(). A few lines later, still in the same prepareCommit() invocation, collectCommitData() (line 362) drains the queue again and builds the actual committableOffsets. Both drainCompletions() calls mutate real WorkManager/PartitionState via handleFutureResult, fed asynchronously from worker threads via a ConcurrentLinkedQueue — this is not a static field read twice (unlike stock Streams, whose analogous committableOffsets construction reads single-threaded fields only the StreamThread ever writes).
Failure scenario: a worker completes a record in the gap between the two calls. The flush-gate's earlier drain sees nothing dirty yet and skips the state-store cache flush, but the second drain (moments later) picks up the new completion and includes its offset in what actually gets committed. For a stateful topology, a crash after that commit but before the next flush loses state-store data despite the offset being marked committed.
| + // exists that no successful commit has covered. The consumer-position sweep below reads | ||
| + // consumedOffsets, which the PC path no longer maintains. | ||
| + if (pcDispatcher != null) { | ||
| + return pcDispatcher.hasCommitDataOutstanding(); |
There was a problem hiding this comment.
Efficiency: commitNeeded() — called by Kafka Streams on its processing hot path, not just at commit boundaries — now does a queue drain plus an O(#partitions) scan instead of a cheap boolean read.
On the PC path, commitNeeded() calls pcDispatcher.hasCommitDataOutstanding(), which drains the completion queue (real per-item work via handleFutureResult) and then calls workManager.isDirty(), which in PartitionStateManager.isDirty() is partitionStates.values().stream().anyMatch(PartitionState::isDirty) — an O(number of assigned partitions) live scan, not a cached O(1) flag. Stock Streams' commitNeeded() is a plain boolean field read.
Failure scenario: under many assigned partitions and a workload where commitNeeded() is polled more often than the actual commit interval (Kafka Streams' architecture calls it beyond just commit time), this adds measurable per-call CPU overhead that only shows up under load, not in the existing tests. Consider caching a WorkManager-level aggregate dirty flag maintained incrementally instead of draining/scanning on every call.
| Paths.get(requiredProperty("pc.connect.patched.reports.dir"))); | ||
| } | ||
|
|
||
| private static URL codeSourceOf(Class<?> type) { |
There was a problem hiding this comment.
Reuse: codeSourceOf(Class<?>) is copy-pasted a third time. It's byte-for-byte identical to ShadowedClassLoadingTest.java:60 in this same new module, and to the pre-existing parallel-consumer-streams-spike's ShadowedClassLoadingTest.java:99. Three independent copies of the same reflection helper now have to be kept in sync by hand across two modules — factor it into one shared test-support class both spike modules (and both test classes within connect-spike) can call.
| return type.getProtectionDomain().getCodeSource().getLocation(); | ||
| } | ||
|
|
||
| private static String requiredProperty(String name) { |
There was a problem hiding this comment.
Reuse: requiredProperty(String) is duplicated verbatim from PatchHarnessTest.java:180 within the same new module (identical body: System.getProperty, null/blank check, same IllegalStateException message format). No shared testFixtures/testkit module exists in the repo to host it yet — at minimum factor this into one package-private shared helper class within parallel-consumer-connect-spike instead of maintaining two copies.
| final class WorkerSinkTaskRegressionReportsVerifier { | ||
|
|
||
| static final String UPSTREAM_TEST_CLASS = "org.apache.kafka.connect.runtime.WorkerSinkTaskTest"; | ||
| static final int EXPECTED_TEST_COUNT = 30; |
There was a problem hiding this comment.
Simplification: EXPECTED_TEST_COUNT = 30 is a redundant, brittle magic-number check layered on top of requireExactIdentities(), which already fails on any missing/extra manifest entry against the actual discovered test outcomes (see line ~140+). The count check adds no coverage requireExactIdentities doesn't already provide.
Cost: every time upstream Kafka's WorkerSinkTaskTest gains or loses a case and the manifest is updated, this unrelated constant must be remembered and bumped too, or the build fails with a misleading "manifest must contain exactly 30" message instead of surfacing the real diff. Consider dropping the constant and count guard, relying on the non-empty check plus requireExactIdentities.
| } | ||
| } | ||
|
|
||
| private static final class MethodSummary { |
There was a problem hiding this comment.
Simplification: MethodSummary hand-rolls equals/hashCode/toString (~39 lines) for a single-use test value class used in exactly one assertion. Lombok @Value — already used pervasively elsewhere in this codebase (e.g. parallel-consumer-core) and available on this module's classpath via the parent POM — would generate the same semantics in one annotation. Alternatively, AssertJ's tuple(...)/extracting(...) on the Method[] stream would eliminate the class entirely.
| + * postCommit, the partition stays dirty, and the next collection simply returns the same or newer | ||
| + * data - nothing is stranded, nothing is double-cleaned. | ||
| + */ | ||
| + private Map<TopicPartition, OffsetAndMetadata> pcCollectedForCommit; |
There was a problem hiding this comment.
Simplification: pcCollectedForCommit carries three logical states (null / empty / populated) across prepareCommit() and postCommit() instead of always being a non-null map, forcing a != null && !isEmpty() guard at the consumption site (lines 376-379) where the null and empty cases mean exactly the same thing ("nothing to acknowledge").
Initializing this field to Collections.emptyMap() and reassigning it to Collections.emptyMap() (instead of null) after consumption would collapse the guard to !pcCollectedForCommit.isEmpty() with no null-check needed anywhere, and remove the latent NPE risk for any future code that reads the field between the two methods.
| private PcConnectDispatchBridge() { | ||
| } | ||
|
|
||
| /** Returns the hard-coded disabled state; the call exists solely to force runtime linkage. */ |
There was a problem hiding this comment.
Altitude: this bridge's shape diverges from the already-proven PcTaskDispatcher seam it's meant to follow. enabled() is a bare hard-coded return false; with no field, config, or object behind it, whereas the streams-spike's equivalent seam (PcTaskDispatcher.createIfEnabled(...)) already returns a real nullable dispatcher instance checked via != null at call sites — a shape proven to extend cleanly to real dispatch logic.
Cost: when U2 wires up real dispatch, this boolean constant will have to become a stateful object and every call site rewired from a static boolean check to an instance-null check — a reshape of pcconnect.patch, not an incremental extension of it. Modeling the bridge on createIfEnabled now (even while permanently returning null/disabled) would avoid that redo.
| <connect.patched.reports.dir>${project.build.directory}/surefire-reports-worker-sink-patched</connect.patched.reports.dir> | ||
|
|
||
| <!-- This module is a local feasibility harness, not an artifact users may consume. --> | ||
| <maven.deploy.skip>true</maven.deploy.skip> |
There was a problem hiding this comment.
Altitude: the publication-disabled guard is copy-pasted independently into each spike module's pom rather than inherited from parallel-consumer-parent. Both parallel-consumer-streams-spike/pom.xml and this file carry their own copies of the three skip properties plus the central-publishing-maven-plugin skipPublishing block, and docs/inflight/release-spike-modules-publication-disabled.md explicitly warns "both halves, per module, always" must never be forgotten — prose compensating for a missing shared mechanism.
Cost: a future third spike module gets this guard only if someone remembers to copy both halves correctly; centralizing it as a shared property block / profile in the parent pom would make it inherited by default instead of relying on memory and documentation.
…r, and refresh the ledger The spike plan was written before the parent branch's U9 landed, so its U2 assumed a dispatcher without a commit surface. Verified against the merged tree: WorkPreparer exists as U2 assumed, no owner-thread guard exists yet, and the dispatcher now also carries collectCommitData, hasCommitDataOutstanding, onCommitSuccess and abortClose. U2's owner-thread guard therefore extends to the commit surface - those methods reach WorkManager directly and are exactly what a later commit-composition step will be tempted to call from a commit thread. The reject-commits tripwire on the dispatcher's MockConsumer stays sound post-U9: the Streams commit path uses Streams' own consumer and the crash-restart test a real broker, so nothing on the parent exercises those overloads. The plan's context section now names the U9 commit machinery and the follow-on design it unlocks: complete a record's work only when the owning task's preCommit() covers it, making PC's frontier the durable frontier with no clamping - recorded in full in the ledger entry. The parked- ledger entry becomes pr-connect-on-pc-spike.md: the work is no longer parked, it is PR #269 stacked on the ks branch. State, next steps, the frontier-composition design, and the publication gate are all pointed to from there; stale references in the plan and the packaging note now name the new file. Refs #240 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015GyZmjA6k1Fi9YQWavenke
Both test lanes on PR #269 failed on the same single cause: a Maven Central read timeout fetching kafka-streams-3.9.2-sources.jar during the streams spike's unpack step. Every module that reached its tests was green; no test failed. The cache job's dependency:go-offline resolves declared dependencies only. The spike modules fetch sources-classifier jars through dependency:unpack executions, which go-offline never sees, so each lane re-downloaded them from Central on every run and inherited its timeouts. Running the two modules' generate-sources in the cache job performs exactly those unpacks into the cached repository, so the lanes hit the cache instead of the network. Same failure class as the documented Azure-region Central timeouts that motivated the cache job originally; this closes the gap the new modules opened in it. Refs #240 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015GyZmjA6k1Fi9YQWavenke
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
|
Claude finished @astubbs's task in 5m 4s —— View job Code review — U1 Connect spike (
|
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
…hing else docs/ carried two naming conventions at once. Durable reference prose is lowercase - refactoring.md, and every entry under docs/inflight/ - while three files shouted in ALL_CAPS_WITH_UNDERSCORES for no reason anyone recorded. Renames them to match the majority: docs/QUARANTINED_TESTS.md -> docs/quarantined-tests.md docs/TODO_INDEX.md -> docs/todo-index.md docs/SELF_HOSTED_RUNNER.md -> docs/self-hosted-runner.md Root-level names are deliberately untouched. AGENTS.md, CONCEPTS.md, README.adoc, CHANGELOG.adoc and NOTICE are ecosystem conventions that tools and humans expect in caps - renaming those would break a convention rather than fix an inconsistency. Only this repository's own idiosyncrasy is in scope. The rename is the small half. All three paths are load-bearing: two are enforced by CI gates and one is generated, so every reference had to move with them. Swept 26 files - nine workflows, eight shell scripts, four core test classes, both AGENTS.md files including the "Where things live" table, refactoring.md, a solutions write-up, two plans, and actionlint.yaml. The CHANGELOG reference sits in the unreleased 0.6.0.0 section, which the release regenerates anyway, and it was a link that would now 404 - a factual correction, the one edit the changelog rules permit. Verified rather than assumed, because a missed reference here fails CI rather than failing loudly at edit time: zero remaining matches for the old names anywhere in the tree; check-quarantine-registry.sh consistent; todo-index.sh --check up to date, so the generator and its output agree on the new name; check-quarantine-owners.sh clean; check-issue-refs.sh clean across all changed files; and the 31 tests in QuarantineRegistryScriptTest, QuarantinedAnnotationContractTest and CheckQuarantineOwnersScriptTest green - those exercise the scripts end to end against the renamed registry. Refs #240 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015GyZmjA6k1Fi9YQWavenke
17e4807 to
f49b62e
Compare
…mbed direction was dropped Plans running Kafka Connect sink tasks under Parallel Consumer, resolving the Kafka Connect integration question that upstream never answered. Written as an embedded reimplementation on connect-api, then superseded during review. Committed with a superseded-direction header rather than dropped, because the offset analysis transfers to the patch approach and is the expensive part of the work: - The committed offset and its encoded incomplete-offset payload are both anchored to getOffsetHighestSequentialSucceeded() + 1, and the read side decodes against OffsetAndMetadata.offset(). Lowering the offset while keeping the payload shifts the whole decoded incomplete set. - PartitionState.onOffsetCommitSuccess() calls setClean(), and the commit gate is isTimeToCommitNow() && wm.isDirty() && !isRebalanceInProgress - so a clamped partition that goes clean is never committed again, and a sink that later flushes never advances the group. - preCommit() runs on pc-broker-poll, not the controller thread, and on the revoke path inside synchronized(commitCommand) - the monitor of the open deadlock in #29. Bounded acquisition is mandatory. - Connect's preCommit() contract: lower offsets honoured, higher rejected, an omitted partition means leave it, an empty map means skip the commit. Review also established the reason for the pivot: one SinkTask per partition plus PARTITION ordering caps concurrency at the partition count, matching a Connect worker at tasks.max, and every runtime feature the module skipped (SMTs, DLQ, ConfigProvider, plugin isolation) is one a patched Connect inherits for free. Refs #240 Refs confluentinc#119 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015GyZmjA6k1Fi9YQWavenke
…d park the Connect direction AGENTS.md gains a rule agents keep needing and never derive on their own: engineering folklore carries an unstated cost model built on human attention, fatigue and tedium, and an agent inherits the advice and the cost model together without being able to tell them apart. The advice is usually right. The cost model usually is not the agent's. Names what is cheap for an agent and must never steer a design - merge conflicts and patching a moving upstream, reading whole subsystems rather than inferring a contract from javadoc, exhaustiveness, wide mechanical refactors, keeping copies in sync - and, so the rule does not collapse into "nothing is hard", what genuinely still binds: wall-clock CI, non-determinism, judgment about people, anything unverifiable, and irreversible actions. Prompted by this issue. The shelved plan reimplemented part of Kafka Connect's WorkerSinkTask instead of patching Connect, because reimplementing looked smaller; it was more code and less capability, and its deferred list existed only because rebuilding those features is expensive for a human. Also parks the direction in docs/inflight/ per the "Where things live" table: the patch-Connect approach modelled on feats/ks-on-pc-spike, the sparse-offsets -across-tasks question the PoC exists to answer, the partition-file connectors that cannot lose whole-partition ownership, and the follow-up to select the operating mode per connector at runtime rather than globally. Refs #240 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015GyZmjA6k1Fi9YQWavenke
…fka fork Records what blocks publishing anything built on the build-time Kafka patch strategy. Development is not blocked: generating patched classes locally and gitignoring them distributes nothing. Publishing them is redistribution of a modified Apache Kafka, and only then do the obligations attach. Settled here: one published artifact per patched Kafka module, versioned independently, so a Connect consumer never pulls a patched Streams. Each publishes under our own groupId with a DISTINCT artifactId - reusing the upstream artifactId reads as a drop-in signal while preventing nothing, since Maven deduplicates on groupId and artifactId together. Naming enforces nothing either way; a bannedDependencies rule is what fails a build that resolves both jars instead of letting classpath order decide. The distribution model is a drop-in fork, not shading: relocation is unavailable because the patch depends on the class keeping the name Kafka's own internals reference by, and consumer-side patch application distributes no Kafka but is unusable. So the hazard is dependency hygiene, not classpath archaeology. Open: trademark - Apache 2.0 grants copyright and patent rights and explicitly not trademark, and Kafka is an ASF mark - the section 4 changed-files notice, reproducing Kafka's NOTICE, and per-Kafka-version coupling against the queued Kafka 4 move. A starting point for investigation, not advice; a human should confirm before anything publishes. Refs #240 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015GyZmjA6k1Fi9YQWavenke
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: c7a1c57f49
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| final Map<TopicPartition, OffsetAndMetadata> returned = lane.preCommit(toCommit); | ||
| if (returned == null || returned.isEmpty()) { | ||
| // The task opted out. Connect treats this as a successful no-op commit; here it simply means | ||
| // nothing became durable, so nothing completes and the frontier stays where it is. | ||
| return Collections.emptyMap(); | ||
| } | ||
| return returned; |
There was a problem hiding this comment.
When a SinkTask returns an offset above the toCommit value it received, stock WorkerSinkTask rejects that over-claim, but this method returns it unchanged and confirm() completes every deliverable record below it. For example, if offset 1 is still staged while offsets 0 and 2 are deliverable, the offered watermark is 1; a returned watermark of 3 then marks offset 2 durable even though the flush was only requested through 1, allowing PC's encoded holes to skip offset 2 after a crash. Reject or clamp each returned offset against the snapshot supplied to that lane.
Useful? React with 👍 / 👎.
| - **Offset commit composition.** `SinkTask.preCommit()` returns a per-task watermark; Parallel Consumer | ||
| tracks sparse completion and commits a frontier. Reconciling those is the hard part and is unsolved | ||
| here. Nothing in this module calls `preCommit`, `flush`, or an offset committer, and **completion means | ||
| the callback returned — which is not a durability claim.** |
There was a problem hiding this comment.
This section now contradicts the module in the same diff: PcSinkTaskDurabilityBarrier.pollWatermarks() calls preCommit, PcSinkTaskLaneRouter.runDurabilityCycle() interprets those watermarks, and the deferred completion seam explicitly avoids completing when the callback merely returns. Although the patched WorkerSinkTask remains disabled, describing offset composition itself as unsolved and absent misstates what the feasibility harness actually proves and obscures the barrier's documented limitations.
Useful? React with 👍 / 👎.
| The plan is `docs/plans/2026-08-09-001-feat-connect-on-pc-plan.md`; the superseded embed-direction plan | ||
| (`2026-08-08-001-...`) is kept for its offset analysis. U1 — the `parallel-consumer-connect` module and | ||
| its generated-`WorkerSinkTask` shadowing proof — is on the branch and green in CI. U2 (key-to-lane | ||
| dispatch, reconciled against the parent's commit surface) is next; the module README and boundary | ||
| write-up land with the plan's U3. Publication of both experimental modules is disabled — see |
There was a problem hiding this comment.
Fresh in this version, this replacement state note still says U2 is next even though the same diff adds PcSinkTaskLaneRouter, the durability barrier, their probes, and the U3-era README; later sections of this file even discuss those completed artifacts. Anyone scanning docs/inflight/ therefore receives mutually contradictory live state, so update the state block to the genuinely open next step rather than retaining the pre-U2 snapshot.
AGENTS.md reference: docs/inflight/AGENTS.md:L32-L38
Useful? React with 👍 / 👎.
| } | ||
|
|
||
| /** | ||
| * Runs a lifecycle callback under the same lock, so {@code open}, {@code close} and {@code stop} can | ||
| * never interleave with an in-flight {@code put}. |
There was a problem hiding this comment.
When a connector returns a reusable mutable offset map from preCommit, this method releases the lane lock while the router still retains and reads that map to build the cross-lane ceiling and later confirm completions. A worker can then enter put and mutate the same map, which is harmless under stock Connect's single-threaded call sequence but here can cause concurrent-iteration failures or make the router confirm a watermark advanced after the completed flush. Copy the returned map before unlocking so each durability cycle interprets the exact answer produced by that preCommit call.
Useful? React with 👍 / 👎.
…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.
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
…ms and Connect PRs, and sharpen the Pulsar comparison The partition-economics section recorded two limits as permanent that are only true of stock Kafka Streams and stock Connect, and open work on this fork contradicts both. #271 replaces the point where StreamTask selects the next record with PC's WorkManager, giving a topology per-key concurrency inside a partition - 57x on the quickest record and 8x median, with the 0.69x single-key control arm that makes those credible. #269 patches WorkerSinkTask rather than reimplementing it, so the connector is NOT replaced and transforms, dead-letter queues and plugin isolation are inherited; its own review killed the reimplementation direction because one SinkTask per partition caps concurrency at the partition count, which is the ceiling the whole argument is against. So partition count is set by the largest of broker spread, Streams task parallelism, Connect sink parallelism and consumer concurrency, and PC answers the last three, leaving the number a broker actually needs. That is the mechanism behind 5000 to 50. Recorded as a roadmap claim: 271 is an alpha proven on one partition, one task, one instance, and 269 is a deliberately inert shadowing proof. The operational half is restated as the compounding it actually is - fewer instances means fewer clients, and fewer partitions means less broker structure, so both sides of the broker's ledger fall at once. An earlier reading of it as a win in client variety was the owner's point misread. Also adds what else could layer on the seam, the pattern being that anything influencing which record runs next can be expressed there: per-key rate limiting is the strongest candidate, since PC already owns the dispatch decision and a separate limiter cannot compose with retry. The Pulsar entry gains the hypothesis the comparison turns on - that Key_Shared's unit of parallelism is the consumer where PC's is the key, leaving the intra-consumer head-of-line problem untouched - plus the two places Pulsar looks genuinely ahead, native individual acknowledgement and no commit-metadata ceiling. Records that 'is PC faster than Pulsar' should be refused rather than answered, since it collapses into contested Kafka-versus-Pulsar benchmark territory; the answerable form is concurrency per unit of infrastructure at a given ordering guarantee, and it is unmeasured. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…that lands it The drafted kafka-connect-experimental.yaml was removed from the docs corpus before #273 squash-merged, leaving its only reachable copy on the unmerged side-line origin/docs/v6-release-ideas-codex - one branch deletion away from loss, while master's next-experimental-module-records.md instructs the module-landing PR to recover it from history it cannot see. This PR is that module-landing PR, so the record rides here, recovered verbatim. This branch predates docs/features/staging/, so the file arrives alone; master already has the directory and the Streams twin, and the merge composes them. Merge prep for this PR moves the record up into docs/features/ with status published, a real since, and setup that resolves - re-checking the wording against what actually ships, per docs/features/staging/README.md. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
…istory master cannot see next-experimental-module-records.md told the module-landing PRs to recover the drafted feature records from git history - but the #273 squash-merge made those drafts unreachable from master; the only copies sit on the unmerged side-line origin/docs/v6-release-ideas-codex. The Streams draft's staged successor already lives in docs/features/staging/; the Connect draft is now recovered verbatim and committed on its module-landing PR (#269). The note now says where each draft actually is instead of giving an instruction that cannot be followed. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…y lane runDurabilityCycle() polled each lane in a bare loop and PcSinkTaskLane.preCommit rethrows, so one connector whose store was unreachable aborted the cycle for every lane: those later in the list were never asked, and because confirmation runs only after every watermark is gathered, no lane completed anything. Not a safety defect - nothing is confirmed, so nothing is over-committed - but it is the head-of-line stall this module exists to remove, reappearing one level up. A throwing lane now contributes an empty answer. Empty rather than partial is load-bearing: it neither confirms that lane's own records nor raises the cross-lane ceiling the inverted control rule reads, so the probe's negative control keeps working. Retried, not quarantined. A thrown preCommit is a connector's transient-failure signal, and WorkerSinkTask itself logs the failed commit and tries again on the next interval rather than retiring the task; diverging from the runtime being patched would need a reason, and one network blip is not it. Retrying is safe because a throw confirms nothing: the records stay in the barrier's deliverable map, PC's frontier holds at the lowest incomplete offset, and a later cycle completes the whole backlog. Quarantining would also need an un-quarantine path, which nothing here has. Only RuntimeException is caught - an Error means the JVM is in trouble rather than the connector, and continuing the cycle around it would hide that. DurabilityCycleResult returns the failures rather than logging and dropping them, because a lane that never answers never completes a record, and the only other symptom is a frontier that quietly stops advancing. The confirm phase is deliberately NOT isolated. A throwing CompletionHandle would stall lanes identically, but U6's goal names preCommit and all four of its test scenarios concern preCommit, so this is recorded rather than silently widened. Red first: both new arms failed with ConnectException propagating PcSinkTaskLane.preCommit -> pollWatermarks -> runDurabilityCycle, which is the defect itself. The retry arm counts preCommit calls per cycle, so retry-versus-quarantine is observed rather than asserted. Verified with the plan's gate: connect 34 + 32 + 32 tests, 0 failures, Kafka's own WorkerSinkTaskTest 30/30 in both the stock and patched-disabled arms. The reactor's one failure is ProducerManagerTest.producedRecordsCantBeInTransactionWithoutItsOffsetDirect, the documented producer-lock getElapsed flake - see docs/inflight/test-local-core-failures-2026-08-10.md, third sighting today. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
…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>
|
Process note: this PR is the carrier of the roadmap entry |
… gate, and the self-tuning promotion (#314) The roadmap now carries the 2026-08-14..17 ideation wave, says how real every entry is, and enforces that it stays true. Six new entries at next-0x - the horizon the 1.0 exit criterion inspects: the language-proxy sidecar (#242), the polyglot proof demo, the performance comparison matrix, distributed throttling (#228), self-tuning concurrency (#227, promoted from the backlog to follow STRATEGY.md's self-tuning bet), and the batch-composition settlement (#145, confluentinc#266) that must precede confluentinc#915. Three backlog entries record what was considered and deliberately not scheduled: the proxy HTTP/REST-compat gateway, the Dapr component, and the actor/IPC revival. The running-instance-visibility entry is renamed web-gui - the name everything else uses. Every entry now carries stage and stage_detail - an eight-value ladder (idea, ideated, requirements-drafted, planned, limited-poc, poc, in-progress, implemented) ordered by the most advanced artifact that exists, with definitions deliberately hard to flatter: implemented requires proven use, and a brand-new system nobody has run in anger is a poc however large. An optional stage_delivery records the readiness judgment (draft vs pending-merge) for entries carried by an open PR. The schema makes stage/stage_detail entry_required, so no entry can claim a horizon without saying how real the work behind it is - built for the v6 release announcement to draw from directly. The ladder's ownership rule - the PR advancing a track moves its entry's stage in the same change - gets its own gate: roadmap-stage-gate.js compares the claiming entry's stage block between base and head (editing another entry does not count), accepts only the fork's qualified astubbs#N carrier form, and offers a reasoned roadmap-stage: N/A opt-out. Seventeen unit tests run before the gate in CI, one pinned against the real roadmap file. A merge-checklist line covers entries the gate cannot reach. Also corrects the module-records note that pointed at git history the #273 squash made unreachable - the drafted Connect feature record now rides its module-landing PR (#269) - and merges master so the tree actually contains the STRATEGY.md self-tuning section the roadmap cites (a review catch: the branch was cut six minutes before #308 merged). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…tely does not carry `bin/check-file-refs.sh` was red on this PR - five citations that do not resolve - and it was red before any of today's work, so the PR Checklist gate has been failing on it. Each names a file on purpose rather than by mistake, which is exactly what the gate's paragraph-level `file-refs: N/A` marker is for: - `pr-323-docs-outstanding.md` names `bin/check-inflight-tags.sh` and `bin/lib/inflight-tags.sh` to say they are NOT here - they ship in #324, and that separation is the entry's whole subject. - `core-dapr-adapter.md` names a solutions write-up and a `parallel-consumer-streams` script a future probe would create. - `release-experimental-module-records.md` names a staging feature record created by the module-landing PR #269. Markers are placed on the paragraph the gate itself names, and each was then tested by REMOVING it to confirm the gate goes red without it - one was redundant and is deleted. An unnecessary N/A is worse than none: it silently exempts whatever else lands in that paragraph later.
The Kafka Streams (#255) and Connect (#240) work is the largest thing this repo is doing and it is invisible from master: no module, no plans, no notes. An agent listing docs/inflight/ found only five sideways references from other notes - "when #271 lands", "the Streams and Connect rows" - and no way in. Deliberately signposts, not handovers. Each workstream already has a detailed note ON its branch, travelling with the code it describes, and those stay where they are; these two files exist so someone on master can find them. They carry only what is true from here and what a command cannot answer. Streams: that PR #271's head is the BASE of a dozen sibling branches rather than the tip of the work, so reading the PR head as the state is wrong; that the cluster still predates the package rename while #271's own head does not; that it does not gate 0.6.0.0. Connect: that #269 is based on feats/ks-on-pc-spike rather than master, so its diff is not its own work and it cannot land first; and that feats/connect-on-pc-spike-codex-version is NOT a competing design awaiting a decision. Its plan document is byte-identical to the live branch's copy and its code differs only by the connectspike -> connect rename - the live branch is a strict superset. That was worth establishing precisely because the pair reads as an unmade decision, which would cost the next reader a comparison with no answer in it. Tagged coordination on both: what it costs to not know either is a collision or a wasted rebase, not a product defect. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019infZnCpPvX8c6QWFC1uLC
…nnect sinks would plug into Owner direction, 2026-09-10. A route may end in a sink: a functional interface that receives the value or the record, returns on success and throws to retry, so it is the route one function and needs nothing on the wire. The console sink becomes one sink among others, and a Connect sink-connector task wrapped as a sink is the adapter the Connect-on-PC work (#240, #269) targets, rather than a second seam. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LpkQjk4NxtZtPERHvyxeQu
…ne is picked by cost and risk, and the sink terminal is not the Connect-on-PC work Owner review of the third round's outcome, 2026-09-10. The export fraction default is seventy percent, not sixty: five points below the engine's pause threshold is margin enough, and sixty leaves capacity on the table. The inflight note and the issue-response draft for #237 carry the same number. The first milestone is chosen in planning by implementation cost and risk, not by the README-minimal list, which becomes its upper bound: the owner wants a subset that can ship in the next release beside the bug-fix line, something to show at low risk. The candidate to cost first is the entry point with typed routes, the outcomes with park in place, and the sandbox, with no export, breaker, batch mode or handle operations. R35's sink terminal and the Connect-on-PC work (#240, #269) serve different people: the sink lets a route use an existing connector task as its client so the user writes none; Connect-on-PC is for people configuring connectors with this library transparently underneath. They share the task and the sink seam, nothing else, and the plan no longer calls one the adapter for the other. The facade creates a producer only when something needs one: produced types, a dead-letter destination, or the transactional commit mode. A definition with none of these opens no producer, so the type-level rules are what turn producing without one into a compile error. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LpkQjk4NxtZtPERHvyxeQu
1 participant
Description
Groundwork for running Kafka Connect sink tasks on Parallel Consumer (#240, upstream confluentinc#119): the direction decision, the offset-correctness analysis, and U1 of the feasibility spike - a generated, patched
WorkerSinkTaskproven to shadow the stock class while staying observably inert.Stacked on
feats/ks-on-pc-spike- that branch is this PR's base, so the diff here shows only the Connect work. No ks PR exists yet; when the parent lands, this retargets tomaster. History will be rewritten and squashed at merge prep.What's in here
WorkerSinkTaskonconnect-api. Review falsified its premise: oneSinkTaskper partition underPARTITIONordering caps concurrency at exactly the partition count - the same ceiling as a Connect worker attasks.max, and every feature it deferred (SMTs, DLQ,ConfigProvider, plugin isolation) is one a patched Connect inherits instead. The plan is kept with a superseded-direction header because its offset analysis (metadata-anchoring invariant, clamp-stall mechanism, broker-poll threading) transfers unchanged. The adopted direction - patch Connect, stack on the Streams-spike machinery - lives indocs/inflight/parked-connect-on-pc.md.docs/plans/2026-08-09-001-feat-connect-on-pc-spike-plan.md,parallel-consumer-connect-spike/): a development-only module that generatesWorkerSinkTaskfrom the released 3.9.2 sources jar, applies a two-line tracked patch (an import plus a hard-disabled bridge field - a method call, deliberately, so the linkage cannot be constant-folded away), and proves shadowing with two isolated arms of Kafka's own publishedWorkerSinkTaskTest: a stock arm and a patched arm, checked against an exact-identity manifest so zero-discovery cannot pass. A source-delta guard pins the patch to exactly the reviewed hunks. U2 (key-to-lane dispatch) and the frontier-composed commit design are follow-on work.maven.deploy.skip/install.skip/gpg.skipand thecentral-publishing-maven-pluginskipPublishingblock - the plugin config is the half that actually stops Central publication. Nothing publishes, releases or snapshots, until the fork packaging/licensing questions are resolved (docs/inflight/release-spike-modules-publication-disabled.md; do not reverse before merge).Verification note
CI on this PR is the first execution evidence for the spike module's test harness - no workflow ran on the contributing branch. Treat a green run here as the actual U1 proof, and a red one as a genuine finding, not noise.
Checklist
🤖 Generated with Claude Code
https://claude.ai/code/session_015GyZmjA6k1Fi9YQWavenke