Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -33,10 +33,11 @@ repositories {
java {
// these settings apply to all jvm tooling, including groovy
toolchain {
languageVersion = JavaLanguageVersion.of(21)
languageVersion = JavaLanguageVersion.of(25)
}
sourceCompatibility = 17
targetCompatibility = 17
// Compile target (--release 17 / source-target) is set once in
// io.seqera.java-library-conventions, which every module applies. Keeping it
// there avoids duplicate/conflicting configureEach across sibling conventions.
}

group = 'io.seqera'
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ plugins {
}

dependencies {
implementation "org.apache.groovy:groovy:4.0.24"
implementation "org.apache.groovy:groovy:4.0.31"
}

group = 'io.seqera'
22 changes: 16 additions & 6 deletions buildSrc/src/main/groovy/io.seqera.java-library-conventions.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -33,10 +33,20 @@ repositories {
java {
// these settings apply to all jvm tooling, including groovy
toolchain {
languageVersion = JavaLanguageVersion.of(21)
languageVersion = JavaLanguageVersion.of(25)
Comment thread
pditommaso marked this conversation as resolved.
}
sourceCompatibility = 17
targetCompatibility = 17
}

// Target Java 17. For Java sources use --release, which also constrains the API
// surface to the Java 17 stdlib; Groovy does not honor --release, so its bytecode
// level is pinned via source/target. --release and -source/-target cannot coexist
// on the same javac task, hence the split across compile task types.
tasks.withType(JavaCompile).configureEach {
options.release = 17
}
tasks.withType(GroovyCompile).configureEach {
sourceCompatibility = '17'
targetCompatibility = '17'
}

test {
Expand All @@ -54,9 +64,9 @@ dependencies {

testImplementation 'ch.qos.logback:logback-core:1.5.25'
testImplementation 'ch.qos.logback:logback-classic:1.5.25'
testImplementation "org.apache.groovy:groovy:4.0.24"
testImplementation "org.apache.groovy:groovy-nio:4.0.24"
testImplementation "org.apache.groovy:groovy-test:4.0.24"
testImplementation "org.apache.groovy:groovy:4.0.31"
testImplementation "org.apache.groovy:groovy-nio:4.0.31"
testImplementation "org.apache.groovy:groovy-test:4.0.31"
testImplementation "cglib:cglib-nodep:3.3.0"
testImplementation "org.objenesis:objenesis:3.4"
testImplementation "org.spockframework:spock-core:2.3-groovy-4.0"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -35,10 +35,11 @@ repositories {
java {
// these settings apply to all jvm tooling, including groovy
toolchain {
languageVersion = JavaLanguageVersion.of(21)
languageVersion = JavaLanguageVersion.of(25)
}
sourceCompatibility = 17
targetCompatibility = 17
// Compile target (--release 17 / source-target) is set once in
// io.seqera.java-library-conventions, which every module applies. Keeping it
// there avoids duplicate/conflicting configureEach across sibling conventions.
}

test {
Expand All @@ -56,9 +57,9 @@ dependencies {

testFixturesImplementation 'ch.qos.logback:logback-core:1.5.25'
testFixturesImplementation 'ch.qos.logback:logback-classic:1.5.25'
testFixturesImplementation "org.apache.groovy:groovy:4.0.24"
testFixturesImplementation "org.apache.groovy:groovy-nio:4.0.24"
testFixturesImplementation "org.apache.groovy:groovy-test:4.0.24"
testFixturesImplementation "org.apache.groovy:groovy:4.0.31"
testFixturesImplementation "org.apache.groovy:groovy-nio:4.0.31"
testFixturesImplementation "org.apache.groovy:groovy-test:4.0.31"
testFixturesImplementation "cglib:cglib-nodep:3.3.0"
testFixturesImplementation "org.objenesis:objenesis:3.4"
testFixturesImplementation "org.spockframework:spock-core:2.3-groovy-4.0"
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
/*
* Copyright 2026, Seqera Labs
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/

plugins {
// Inherit the standard library setup (java-library, groovy, maven-publish,
// test deps, publishing) and its --release 17 default.
id 'io.seqera.java-library-conventions'
}

group = 'io.seqera'

// Libraries that depend on Micronaut target Java 25: Micronaut runs on the Java 25
// service runtime (and Micronaut 5 requires it), so these are never loaded by the
// Nextflow Java 17 runtime. Override the java-library-conventions --release 17
// default; --release is dropped since it cannot coexist with -source/-target on the
// same javac task.
tasks.withType(JavaCompile).configureEach {
options.release = null
sourceCompatibility = '25'
Comment thread
pditommaso marked this conversation as resolved.
targetCompatibility = '25'
}
tasks.withType(GroovyCompile).configureEach {
sourceCompatibility = '25'
targetCompatibility = '25'
}
2 changes: 1 addition & 1 deletion lib-cache-tiered-redis/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
*/

plugins {
id 'io.seqera.java-library-conventions'
id 'io.seqera.micronaut-library-conventions'
id 'io.seqera.groovy-library-conventions'
id "io.micronaut.minimal.library" version '4.5.3'
}
Expand Down
7 changes: 4 additions & 3 deletions lib-cmd-queue-redis/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
*/

plugins {
id 'io.seqera.java-library-conventions'
id 'io.seqera.micronaut-library-conventions'
id 'io.seqera.groovy-library-conventions'
id "io.micronaut.minimal.library" version '4.5.3'
}
Expand All @@ -32,8 +32,9 @@ dependencies {
// JSON serialization
implementation project(':lib-serde-jackson')

// Work queue
implementation project(':lib-data-workqueue')
// Work queue - 'api' because CommandQueue publicly extends AbstractWorkQueue and exposes
// WorkQueue / MessageConsumer from this module in its public API (consumers subclass it).
api project(':lib-data-workqueue')
implementation project(':lib-data-workqueue-redis')
implementation project(':lib-serde-moshi')

Expand Down
4 changes: 2 additions & 2 deletions lib-data-broadcast-redis/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
*/

plugins {
id 'io.seqera.java-library-conventions'
id 'io.seqera.micronaut-library-conventions'
id 'groovy'
id "io.micronaut.minimal.library" version '4.5.3'
}
Expand All @@ -33,7 +33,7 @@ dependencies {
implementation "io.micronaut:micronaut-runtime:${micronautCoreVersion}"

// Groovy dependencies only for tests
testImplementation "org.apache.groovy:groovy:4.0.24"
testImplementation "org.apache.groovy:groovy:4.0.31"
testImplementation testFixtures(project(':lib-fixtures-redis'))
testImplementation "io.micronaut:micronaut-inject-groovy:${micronautCoreVersion}"
testImplementation "io.micronaut.test:micronaut-test-spock:${micronautTestVersion}"
Expand Down
4 changes: 2 additions & 2 deletions lib-data-count-redis/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
*/

plugins {
id 'io.seqera.java-library-conventions'
id 'io.seqera.micronaut-library-conventions'
id 'io.seqera.groovy-library-conventions'
// Micronaut minimal lib
// https://micronaut-projects.github.io/micronaut-gradle-plugin/latest/
Expand All @@ -35,7 +35,7 @@ dependencies {
implementation "io.micronaut:micronaut-runtime:${micronautCoreVersion}"

testImplementation testFixtures(project(':lib-fixtures-redis'))
testImplementation "org.apache.groovy:groovy:4.0.24"
testImplementation "org.apache.groovy:groovy:4.0.31"
testImplementation "io.micronaut:micronaut-inject-groovy:${micronautCoreVersion}"
testImplementation "io.micronaut.test:micronaut-test-spock:${micronautTestVersion}"
testImplementation 'org.testcontainers:testcontainers:1.21.4'
Expand Down
8 changes: 4 additions & 4 deletions lib-data-queue-redis/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
*/

plugins {
id 'io.seqera.java-library-conventions'
id 'io.seqera.micronaut-library-conventions'
id 'io.seqera.groovy-library-conventions'
// Micronaut minimal lib
// https://micronaut-projects.github.io/micronaut-gradle-plugin/latest/
Expand All @@ -35,9 +35,9 @@ dependencies {

compileOnly "io.micronaut:micronaut-inject-groovy:${micronautCoreVersion}"
implementation "ch.qos.logback:logback-classic:1.5.25"
implementation "org.apache.groovy:groovy:4.0.24"
implementation "org.apache.groovy:groovy-nio:4.0.24"
implementation "org.apache.groovy:groovy-templates:4.0.24"
implementation "org.apache.groovy:groovy:4.0.31"
implementation "org.apache.groovy:groovy-nio:4.0.31"
implementation "org.apache.groovy:groovy-templates:4.0.31"
implementation "io.micronaut:micronaut-runtime:${micronautCoreVersion}"
implementation "io.micronaut:micronaut-websocket:${micronautCoreVersion}"
implementation 'com.github.ben-manes.caffeine:caffeine:3.1.8'
Expand Down
4 changes: 2 additions & 2 deletions lib-data-range-redis/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
*/

plugins {
id 'io.seqera.java-library-conventions'
id 'io.seqera.micronaut-library-conventions'
id 'io.seqera.groovy-library-conventions'
// Micronaut minimal lib
// https://micronaut-projects.github.io/micronaut-gradle-plugin/latest/
Expand All @@ -34,7 +34,7 @@ dependencies {
implementation "io.micronaut:micronaut-runtime:${micronautCoreVersion}"

testImplementation testFixtures(project(':lib-fixtures-redis'))
testImplementation "org.apache.groovy:groovy:4.0.24"
testImplementation "org.apache.groovy:groovy:4.0.31"
testImplementation "io.micronaut:micronaut-inject-groovy:${micronautCoreVersion}"
testImplementation "io.micronaut.test:micronaut-test-spock:${micronautTestVersion}"
testImplementation 'org.testcontainers:testcontainers:1.21.4'
Expand Down
2 changes: 1 addition & 1 deletion lib-data-store-future-redis/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
*/

plugins {
id 'io.seqera.java-library-conventions'
id 'io.seqera.micronaut-library-conventions'
id 'io.seqera.groovy-library-conventions'
// Micronaut minimal lib
// https://micronaut-projects.github.io/micronaut-gradle-plugin/latest/
Expand Down
2 changes: 1 addition & 1 deletion lib-data-store-state-redis/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
*/

plugins {
id 'io.seqera.java-library-conventions'
id 'io.seqera.micronaut-library-conventions'
id 'io.seqera.groovy-library-conventions'
// Micronaut minimal lib
// https://micronaut-projects.github.io/micronaut-gradle-plugin/latest/
Expand Down
8 changes: 4 additions & 4 deletions lib-data-stream-redis/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -173,7 +173,7 @@ finishes or the consumer dies.
│ (PEL, │ renew (XCLAIM … JUSTID) │ • hand it to the executor │
│ group) │◀───────────── heartbeat daemon ───┤ (never runs it inline) │
│ │ every claim-timeout/3 │ │
│ │ ack (XACK + XDEL) │ worker (virtual thread)
│ │ ack (XACK + XDEL) │ worker (executor thread) │
│ │◀──────────── on terminal ─────────┤ accept(msg): │
└──────────┘ │ ├─ true → ack + free slot │
▲ │ └─ false → keep lease, │
Expand All @@ -185,9 +185,9 @@ finishes or the consumer dies.
**Three mechanisms:**

1. **Async dispatch (no head-of-line blocking).** The dispatcher thread never runs a
handler; it hands each message to a worker executor and moves on. Handlers run on a
shared **virtual-thread** executor by default, or the Micronaut `@Named(BLOCKING)`
executor when supplied via `withHandlerExecutor(...)`. A `Semaphore` sized by
handler; it hands each message to a worker executor and moves on. Handlers run on the
executor supplied via `withHandlerExecutor(...)` — **mandatory, no default** (Micronaut
consumers inject the `@Named(BLOCKING)` executor). A `Semaphore` sized by
`concurrency()` bounds how many messages are in flight at once (backpressure: excess
messages stay in the stream).

Expand Down
10 changes: 5 additions & 5 deletions lib-data-stream-redis/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
*/

plugins {
id 'io.seqera.java-library-conventions'
id 'io.seqera.micronaut-library-conventions'
id 'groovy'
// Micronaut minimal lib
// https://micronaut-projects.github.io/micronaut-gradle-plugin/latest/
Expand Down Expand Up @@ -45,10 +45,10 @@ dependencies {
testImplementation "io.micrometer:micrometer-core:1.12.4"

// Groovy dependencies only for tests
testImplementation "org.apache.groovy:groovy:4.0.24"
testImplementation "org.apache.groovy:groovy-nio:4.0.24"
testImplementation "org.apache.groovy:groovy-templates:4.0.24"
testImplementation "org.apache.groovy:groovy-json:4.0.24"
testImplementation "org.apache.groovy:groovy:4.0.31"
testImplementation "org.apache.groovy:groovy-nio:4.0.31"
testImplementation "org.apache.groovy:groovy-templates:4.0.31"
testImplementation "org.apache.groovy:groovy-json:4.0.31"
testImplementation project(':lib-lang')
testImplementation testFixtures(project(':lib-fixtures-redis'))
testImplementation "io.micronaut:micronaut-inject-groovy:${micronautCoreVersion}"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,9 @@
import java.io.Closeable;
import java.time.Duration;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
Expand Down Expand Up @@ -78,6 +78,9 @@
* // Usage
* MyMessageStream stream = new MyMessageStream(underlyingStream);
*
* // Supply the handler executor (mandatory, no default) before adding consumers
* stream.withHandlerExecutor(executorService);
*
* // Add consumer for a specific stream
* stream.addConsumer("user-events", event -> {
* processUserEvent(event);
Expand Down Expand Up @@ -148,16 +151,14 @@ String key() {
*/
private final Map<String, Long> active = new ConcurrentHashMap<>();

/** Shared virtual-thread executor used when no handler executor is supplied. */
private static final ExecutorService DEFAULT_WORKERS = Executors.newVirtualThreadPerTaskExecutor();

/**
* Executor that runs the message handlers, on virtual threads. Defaults to a shared
* virtual-thread-per-task executor; Micronaut consumers replace it with the injected
* {@code BLOCKING} executor via {@link #withHandlerExecutor}. Handler concurrency is
* bounded by {@link #slots}, not by this executor, so it is never sized or shut down here.
* Executor that runs the message handlers. Supplied by the consumer via
* {@link #withHandlerExecutor} before the first {@link #addConsumer} — there is no default,
* so {@link #startProcessing()} fails fast if it was never set. Micronaut consumers pass the
* injected {@code BLOCKING} executor. Handler concurrency is bounded by {@link #slots}, not by
* this executor, so it is never sized or shut down here.
*/
private volatile ExecutorService pool = DEFAULT_WORKERS;
private volatile ExecutorService pool;

/**
* Gates new intake: a permit is acquired when a lease is picked up and held for the
Expand Down Expand Up @@ -329,8 +330,10 @@ public void addConsumer(String streamId, MessageConsumer<M> consumer) {
* consumer is registered.
*/
private void startProcessing() {
// a handler executor must be supplied via withHandlerExecutor() before processing starts
Objects.requireNonNull(pool, "Handler executor not set - call withHandlerExecutor() before addConsumer()");
// 'slots' — not the executor — bounds how many commands may be in flight at once;
// handlers run on cheap virtual threads, so the cap is a memory/heartbeat ceiling.
// the cap is a memory/heartbeat ceiling, independent of the executor's threading model.
this.slots = new Semaphore(Math.max(1, concurrency()));
this.scheduler = new ScheduledThreadPoolExecutor(1, daemonFactory(name() + "-repoll-" + count.get()));
this.heartbeat = new ScheduledThreadPoolExecutor(1, daemonFactory(name() + "-heartbeat-" + count.get()));
Expand All @@ -340,18 +343,16 @@ private void startProcessing() {
}

/**
* Supply the executor used to run message handlers. Micronaut-managed consumers pass the
* injected {@code @Named(TaskExecutors.BLOCKING)} {@link ExecutorService} (virtual-thread
* backed on JDK 21) <strong>before</strong> the first {@link #addConsumer}. When not
* supplied, a shared virtual-thread executor is used. The executor is never shut down by
* {@link #close()} (it is shared / container-managed).
* Supply the executor used to run message handlers. Consumers must call this
* <strong>before</strong> the first {@link #addConsumer} — there is no default executor.
* Micronaut-managed consumers pass the injected {@code @Named(TaskExecutors.BLOCKING)}
* {@link ExecutorService}. The executor is never shut down by {@link #close()}
* (it is shared / container-managed).
*
* @param executor the shared handler executor; ignored if {@code null}
* @param executor the shared handler executor; must not be {@code null}
*/
public void withHandlerExecutor(@Nullable ExecutorService executor) {
if (executor != null) {
this.pool = executor;
}
public void withHandlerExecutor(ExecutorService executor) {
this.pool = Objects.requireNonNull(executor, "Handler executor cannot be null");
}

private static ThreadFactory daemonFactory(String prefix) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -34,10 +34,12 @@ class TestStream extends AbstractMessageStream<TestMessage> {

TestStream(MessageStream<String> target) {
super(target)
withHandlerExecutor(TestWorkerPool.INSTANCE)
}

TestStream(MessageStream<String> target, StreamMetrics metrics) {
super(target, metrics)
withHandlerExecutor(TestWorkerPool.INSTANCE)
}

static TestStream withRegistry(MessageStream<String> target, MeterRegistry registry) {
Expand Down
Loading
Loading