Repository navigation
Conversation
There was a problem hiding this comment.
Pull request overview
Note
Copilot was unable to run its full agentic suite in this review.
Adds optional Virtual Thread (JDK 21+) execution for the worker pool and refactors several synchronization points from synchronized to ReentrantLock, along with new tests and release notes updates.
Changes:
- Introduces
useVirtualThreadsoption with JVM capability validation. - Updates worker pool abstractions from
ThreadPoolExecutortoExecutorServiceand adds VT-specific behavior. - Replaces
synchronizedblocks/methods withReentrantLockin metrics, producer transaction begin, load factor stepping, and supplier memoization.
Reviewed changes
Copilot reviewed 11 out of 11 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| parallel-consumer-vertx/src/main/java/io/confluent/parallelconsumer/vertx/VertxParallelEoSStreamProcessor.java | Adapts override signature to ExecutorService to match new worker pool abstraction. |
| parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/AbstractParallelEoSStreamProcessorConfigurationTest.java | Adds VT-focused tests and refactors module setup for tests. |
| parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/metrics/PCMetrics.java | Switches close/remove locking from synchronized to ReentrantLock. |
| parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ProducerManager.java | Replaces synchronized transaction-begin guard with an explicit lock. |
| parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ExternalEngine.java | Adapts override signature to ExecutorService. |
| parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/DynamicLoadFactor.java | Refactors stepping synchronization to ReentrantLock. |
| parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java | Adds VT worker pool creation and generalizes worker pool type; introduces lock-based commit coordination and VT-safe pool stats/queue handling. |
| parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelConsumerOptions.java | Adds useVirtualThreads flag + validation for VT support. |
| parallel-consumer-core/src/main/java/io/confluent/csid/utils/SupplierUtils.java | Refactors memoization locking from synchronized to ReentrantLock. |
| README.adoc | Notes the new VT feature and lock refactor in release notes. |
| CHANGELOG.adoc | Notes the new VT feature and lock refactor in changelog. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| workerThreadPool = SupplierUtils.memoize(() -> { | ||
| ExecutorService executor = setupWorkerPool(newOptions.getMaxConcurrency()); | ||
|
|
||
| if (!options.isUseVirtualThreads() && !(executor instanceof ThreadPoolExecutor)) { | ||
| throw new IllegalStateException( | ||
| "When Virtual Threads are disabled, the worker pool must be an instance of ThreadPoolExecutor. " + | ||
| "If you are using a custom ExecutorService, make sure it extends ThreadPoolExecutor." | ||
| ); | ||
| } | ||
| return executor; | ||
| }); |
| protected ExecutorService setupWorkerPool(int poolSize) { | ||
| if (options.isUseVirtualThreads()) { | ||
| try { | ||
| // Thread.ofVirtual().name("pc-vt-").factory() | ||
| Object builder = Class.forName("java.lang.Thread").getMethod("ofVirtual").invoke(null); | ||
| Class<?> builderClass = Class.forName("java.lang.Thread$Builder"); | ||
| builderClass.getMethod("name", String.class, long.class).invoke(builder, "pc-vt-", 0); | ||
| ThreadFactory factory = (ThreadFactory) builderClass.getMethod("factory").invoke(builder); | ||
|
|
||
| // Executors.newThreadPerTaskExecutor(factory) | ||
| return (ExecutorService) java.util.concurrent.Executors.class | ||
| .getMethod("newThreadPerTaskExecutor", ThreadFactory.class) | ||
| .invoke(null, factory); | ||
| } catch (Exception e) { | ||
| throw new IllegalStateException("Virtual threads not supported on this JVM", e); | ||
| } | ||
| } |
| private void validateVirtualThreadsSupport() { | ||
| if (useVirtualThreads) { | ||
| try { | ||
| // Thread.ofVirtual(), Executors.newThreadPerTaskExecutor(ThreadFactory) | ||
| Class.forName("java.lang.Thread").getMethod("ofVirtual"); | ||
| java.util.concurrent.Executors.class.getMethod("newThreadPerTaskExecutor", java.util.concurrent.ThreadFactory.class); | ||
| } catch (NoSuchMethodException | ClassNotFoundException e) { | ||
| throw new UnsupportedOperationException( | ||
| "useVirtualThreads is enabled, but the current JVM does not support Virtual Threads or the required ExecutorService." | ||
| ); | ||
| } | ||
| } | ||
| } |
| public void removeMeter(Meter meter) { | ||
| lock.lock(); | ||
| try { | ||
| if (meter != null) { | ||
| removeMeter(meter.getId()); | ||
| } | ||
| } finally { | ||
| lock.unlock(); | ||
| } | ||
| } |
| @Test | ||
| void testVirtualThreadActivationAndConcurrencyLimit() throws Exception { | ||
| Assumptions.assumeTrue(isVirtualThreadsSupported(), "JDK 21+ required for Virtual Threads"); | ||
|
|
||
| int maxConcurrency = 10; | ||
| var vtOptions = ParallelConsumerOptions.<String, String>builder() | ||
| .consumer(consumer) | ||
| .useVirtualThreads(true) | ||
| .maxConcurrency(maxConcurrency) | ||
| .build(); |
| private boolean isVirtualThreadsSupported() { | ||
| try { | ||
| Class.forName("java.lang.Thread").getMethod("ofVirtual"); | ||
| return true; | ||
| } catch (Exception e) { | ||
| return false; | ||
| } | ||
| } |
|
Hi @astubbs , is there any plans to get this PR reviewed? |
|
@claude take a look at the review so far |
|
Claude finished @astubbs's task in 2m 26s —— View job Claude's review of the VT support PRTasks
I read through the diff ( On the existing Copilot comments
Additional observations
SummaryThe two most actionable items from the existing review are the forced |
The rebuilt ledger opened with a table of every open PR, its branch, its worktree and its ahead/behind counts. That is a second PR tracker: `gh pr list` is always right, this table is wrong within a day, and a reader cannot tell which one they are looking at. Same for branch divergence and worktree paths, which `git` and bin/worktree-status.sh already report. Replaced with the part no command can answer - #29/#31 pointing at the pinned mirror and why the retarget is not mechanical, #38 blocked on a missing ArchUnit engine rather than the version bump, #51 colliding with #57 over PCMetrics, and which PR owns which files. Divergence counts are dropped throughout; what stays is the consequence ("rebase before review"), not the number behind it. Measured flake rates stay - an experiment's results are not re-derivable from a command. The rule is in the header, where the file's other writing rules live. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
A bare #NN means THIS fork. A bare number above the fork's own highest issue is therefore almost always an upstream reference pointing at the wrong repo - harmless-looking today, and wrong for real once the fork's numbering reaches it. Now that every upstream issue is mirrored, there is nearly always a fork number to use instead. The gate runs in the existing PR Checklist job, on ADDED lines only, and adapts as the repo grows: it reads the fork's current highest issue number from the API rather than hard-coding a threshold that rots. Three files are exempt because a bare number legitimately means upstream in them - CHANGELOG.adoc (its own header says entries below 0.6.0.0 predate the fork), upstream-map.yaml (every number is upstream by construction), and upstream-pr-analysis.adoc. Those are named in the module rather than left to judgement because a naive sweep on 2026-08-05 rewrote exactly those three wrongly. Same reason the rule skips javadoc member links, code spans and URLs. Escape hatch mirrors the changelog gate: "issue-refs: N/A - <reason>". Logic is a testable module with 11 assertions running before the gate, following changelog-ref-gate.js - that gate has shipped two real misjudgements, so this one proves its own rules first. PHASE 5 finished. Five single-mirror PRs carry both numbers in the title (#29, #31, #51, #57, #116). #53 and #106 span or only partly address several mirrors, so they get a Related issues section listing them WITHOUT a Fixes keyword - verified 0 closing references on each. A title listing three mirror numbers helps nobody, and Fixes on partial work is the error the #233 / #326 / confluentinc#857 mirrors already document. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QqHpNSXC39ANv9kG1ZvUzn
Review note: the changelog hunk needs removing
Why removing them loses nothing: the section for the release being cut is generated from What it asks of you instead: the commit message is now the raw material. A subject that says what changed and, where it matters, what it changed for a user - with the diagnosis and rejected alternatives in the body. Specific to this PRTwo hunks: Worth noting the second would not earn an entry under the new rules even at release time. The entry test is "can a user or operator observe this without reading our repo - API, behaviour, performance, logs and metrics, or the published artifact?" A |
* Added boolean useVirtualThreads in ParallelConsumerOptions * Generalize workerThreadPool logics to handle Virtual Threads * Added tests refactor: migrate synchronized into ReentrantLock * prevents virtual threads pinning fix: disable load factor in queue target for vt * Bypasses `DynamicLoadFactor` in `getQueueTargetLoaded()` when using VT.
fca394a to
b4d5c2d
Compare
|
Cleaned CHANGELOG.adoc changes, squashed into single commit |
…nish The sweep brings 37 of the 38 open PRs across mechanically. #51 is the exception and it is not a mechanical problem: its head lives on a contributor's fork, so there is no origin ref to work from and, more to the point, nowhere to put the result. A renamed local branch with no push destination is not a delivered result. Recorded rather than solved because every option is somebody's call - push to their fork if maintainer edits are on, ask the contributor to run the procedure, or close it - and because doing nothing has a deadline: once the rename lands, that PR's diff is against paths that no longer exist. The two lower-priority items are recorded with it: the required `dups: similarity` check false-positives on any renamed branch, because its base comparison is keyed by file path; and the accelerators URL that the path-form scan reads as a package, which is 2 branches and does not inherit the #289 conflict-free argument. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…shape most of it shared Written after the sweep rather than before it, so it records measurements instead of expectations. The plan predicted one blocker per branch. There were six classes, and five of them had nothing to do with renaming: they were master cleanups that older branches did not have. B, D, E and the license plugin are all the same shape - a change that landed "ahead of the rename so it stays mechanical", which made it mechanical only for branches containing it. That is the lesson worth carrying out of this, and it is why a future sweep screens for its prerequisites up front instead of meeting them one refusal at a time. C is the one that generalises past this project: `git checkout <ref> -- <file>` is a whole-file overwrite, so a branch's own contribution to that file goes with it, and the failure surfaces four steps downstream looking like rename damage. Also here: the real merge cost, measured rather than rehearsed (3 conflicts on a small branch, 22 on the largest, all mechanical); the mis-pairing fear retired with evidence, since branches carry seven, eight and ten of the near-identical arch tests and every one paired inside its own module at R100; and #51, the one branch this procedure cannot finish, recorded as needing a human because its head lives on a fork and there is nowhere to put the result. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…shape most of it shared Written after the sweep rather than before it, so it records measurements instead of expectations. The plan predicted one blocker per branch. There were six classes, and five of them had nothing to do with renaming: they were master cleanups that older branches did not have. B, D, E and the license plugin are all the same shape - a change that landed "ahead of the rename so it stays mechanical", which made it mechanical only for branches containing it. That is the lesson worth carrying out of this, and it is why a future sweep screens for its prerequisites up front instead of meeting them one refusal at a time. C is the one that generalises past this project: `git checkout <ref> -- <file>` is a whole-file overwrite, so a branch's own contribution to that file goes with it, and the failure surfaces four steps downstream looking like rename damage. Also here: the real merge cost, measured rather than rehearsed (3 conflicts on a small branch, 22 on the largest, all mechanical); the mis-pairing fear retired with evidence, since branches carry seven, eight and ten of the near-identical arch tests and every one paired inside its own module at R100; and #51, the one branch this procedure cannot finish, recorded as needing a human because its head lives on a fork and there is nowhere to put the result. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…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>
…per test, qualified refs, retired links Three corpus repairs behind red gate lanes on this branch: - docs/quarantined-tests.md carried two live entries for PCMetricsTest.metricsRegisterBinding - the original mechanism diagnosis and the #265 re-quarantine - which the registry check rightly reads as drift. The re-quarantine entry is the current one; it now carries a pointer to the first diagnosis (which still stands, in bug-pcmetrics-committed-offset-vs-completion-count.md) and the superseded block is gone. History keeps the original. - docs/data/landing-page.yaml still linked docs/inflight/next-landing-page.md, retired by b0a5da3 when the owner dropped the workstream; both mentions are reworded inline, the same repair that commit applied to the six notes it caught. - Six inflight notes carried bare #NN refs on lines this stack adds vs master: each now names its repo (confluentinc#346/#390/#315/confluentinc#606, #51 - each resolved in both repos first), a quoted upstream issue body takes the line-scoped exempt marker, and two ordinals that were never issue refs at all ("problem 1", "hypothesis 6") are written as ordinals. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019HwDcZdfv1Ne4u3g6ThrG9
|
Hi @devingryu , thanks for the contribution and the ideas! =D I've followed up since with a full virtual threads implementation that is also inline with all the other work going on: #360, so I'll close this PR. |
… 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
Copy of confluentinc#908
Adds option to use Virtual Threads for worker pool. (See also: #896)
useVirtualThreadsoption and execution logic to support Java 21+ Virtual Threads (Project Loom).synchronizedblocks toReentrantLockto prevent VT pinning.setupWorkerPooltoExecutorServiceand improved queue management and backpressure logic accordingly.Concerns
Test Environment Compatibility
The newly added VT-related tests require JDK 21+ and are configured to skip execution using
Assumptionsif the environment does not support it. It seems that current CI environment is based on JDK 17, if so the tests I added will be skipped in CI.Concurrency in
isPoolQueueLowWhen VT mode is enabled, we cannot directly query the queue size, so
WorkerManager.numberRecordsOutForProcessingis used to calculate the load. Please review if this approach creates any potential concurrency issues in a multi-threaded environment.Defensive Logic
To ensure the stability of the existing logic, if VT is disabled (
!useVirtualThreads),setupWorkerPoolis enforced to return aThreadPoolExecutor(throwing an exception otherwise), and type checks have been added. I am concerned this might introduce a breaking change (compatibility issue) for existing users who inheritAbstractParallelEoSStreamProcessorto use custom Executors.Checklist