Ensure decaton works with older jdk

.github/workflows/ci.yml

@@ -12,14 +12,23 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        java: [21]
+        java: [8, 11, 17, 21]
     steps:
     - uses: actions/checkout@v2
     - name: Setup java
-      uses: actions/setup-java@v1
+      uses: actions/setup-java@v4
       with:
-        java-version: ${{ matrix.java }}
+        distribution: temurin
+        java-version: |
+          ${{ matrix.java }}
+          21
+
     - name: Execute test
       uses: eskatos/gradle-command-action@v1
       with:
-        arguments: build jmhJar integrationTest
+        # Java is installed on JAVA_HOME_{java major version}_X64
+        # refs: https://github.com/actions/setup-java/tree/v4.1.0?tab=readme-ov-file#install-multiple-jdks
+        arguments: |
+          -Ptest.java.major.version=${{ matrix.java }}
+          -Porg.gradle.java.installations.fromEnv=JAVA_HOME_${{ matrix.java }}_X64
+          build jmhJar integrationTest

build.gradle

@@ -105,6 +105,13 @@ subprojects {
             showStackTraces true
             showStandardStreams false
         }
+        def testJavaVersion = findProperty("test.java.major.version")
+        if (testJavaVersion != null) {
+            // https://docs.gradle.org/8.5/userguide/toolchains.html#toolchains_for_tasks
+            javaLauncher = javaToolchains.launcherFor {
+                languageVersion = JavaLanguageVersion.of(testJavaVersion)
+            }
+        }
     }
 
     afterEvaluate {

client/src/test/java/com/linecorp/decaton/client/DecatonClientTest.java

@@ -39,8 +39,7 @@
 
 @ExtendWith(MockitoExtension.class)
 public class DecatonClientTest {
-    @Spy
-    private final DecatonClient<HelloTask> decaton = new DecatonClient<HelloTask>() {
+    private static class NoopClient implements DecatonClient<HelloTask> {
         @Override
         public CompletableFuture<PutTaskResult> put(String key, HelloTask task) {
             return null;
@@ -73,7 +72,10 @@ public CompletableFuture<PutTaskResult> put(String key, HelloTask task,
         public void close() throws Exception {
             // noop
         }
-    };
+    }
+
+    @Spy
+    private final DecatonClient<HelloTask> decaton = new NoopClient();
 
     @Test
     public void testPutAsyncHelperOnSuccess() throws Exception {

processor/build.gradle

@@ -19,7 +19,6 @@ dependencies {
 
     itImplementation project(":protobuf")
     itImplementation project(":testing")
-    itImplementation project(":benchmark")
     itImplementation "io.micrometer:micrometer-registry-prometheus:$micrometerVersion"
     itImplementation "org.hamcrest:hamcrest:$hamcrestVersion"
 }

processor/src/it/java/com/linecorp/decaton/processor/VThreadCoreFunctionalityTest.java

@@ -22,6 +22,8 @@
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.condition.EnabledForJreRange;
+import org.junit.jupiter.api.condition.JRE;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
 import com.linecorp.decaton.processor.runtime.ProcessorProperties;
@@ -33,6 +35,7 @@
 import com.linecorp.decaton.testing.RandomExtension;
 import com.linecorp.decaton.testing.processor.ProcessorTestSuite;
 
+@EnabledForJreRange(min = JRE.JAVA_21)
 public class VThreadCoreFunctionalityTest {
     @RegisterExtension
     public static KafkaClusterExtension rule = new KafkaClusterExtension();

processor/src/test/java/com/linecorp/decaton/processor/runtime/internal/ProcessingContextImplTest.java

@@ -67,6 +67,8 @@
 import com.linecorp.decaton.protocol.Decaton.TaskMetadataProto;
 import com.linecorp.decaton.protocol.Sample.HelloTask;
 
+import lombok.RequiredArgsConstructor;
+
 @ExtendWith(MockitoExtension.class)
 public class ProcessingContextImplTest {
     private static class NamedProcessor implements DecatonProcessor<HelloTask> {
@@ -90,6 +92,22 @@ public void process(ProcessingContext<HelloTask> ctx, HelloTask task)
         }
     }
 
+    @RequiredArgsConstructor
+    private static class AsyncCompleteProcessor implements DecatonProcessor<byte[]> {
+        private final CountDownLatch latch;
+
+        @Override
+        public void process(ProcessingContext<byte[]> context, byte[] task) throws InterruptedException {
+            Completion comp = context.deferCompletion();
+            new Thread(() -> {
+                try {
+                    latch.await();
+                } catch (InterruptedException ignored) {}
+                comp.complete();
+            }).start();
+        }
+    }
+
     private static final HelloTask TASK = HelloTask.getDefaultInstance();
 
     private static final DecatonTaskRequest REQUEST =
@@ -354,21 +372,7 @@ public void testPush_Level2_MultiPush_SyncAndAsync() throws InterruptedException
     @Timeout(5)
     public void testRetry() throws InterruptedException {
         CountDownLatch retryLatch = new CountDownLatch(1);
-        DecatonProcessor<byte[]> retryProcessor = spy(
-                // This can't be a lambda for mockito
-                new DecatonProcessor<byte[]>() {
-                    @Override
-                    public void process(ProcessingContext<byte[]> context, byte[] task)
-                            throws InterruptedException {
-                        Completion comp = context.deferCompletion();
-                        new Thread(() -> {
-                            try {
-                                retryLatch.await();
-                            } catch (InterruptedException ignored) {}
-                            comp.complete();
-                        }).start();
-                    }
-                });
+        DecatonProcessor<byte[]> retryProcessor = spy(new AsyncCompleteProcessor(retryLatch));
         TaskRequest request = new TaskRequest(
                 new TopicPartition("topic", 1), 1, null, "TEST".getBytes(StandardCharsets.UTF_8), null, null, REQUEST.toByteArray(), null);
         DecatonTask<byte[]> task = new DecatonTask<>(
@@ -399,21 +403,7 @@ public void testRetry_NOT_CONFIGURED() throws InterruptedException {
     @Timeout(5)
     public void testRetryAtCompletionTimeout() throws InterruptedException {
         CountDownLatch retryLatch = new CountDownLatch(1);
-        DecatonProcessor<byte[]> retryProcessor = spy(
-                // This can't be a lambda for mockito
-                new DecatonProcessor<byte[]>() {
-                    @Override
-                    public void process(ProcessingContext<byte[]> context, byte[] task)
-                            throws InterruptedException {
-                        Completion comp = context.deferCompletion();
-                        new Thread(() -> {
-                            try {
-                                retryLatch.await();
-                            } catch (InterruptedException ignored) {}
-                            comp.complete();
-                        }).start();
-                    }
-                });
+        DecatonProcessor<byte[]> retryProcessor = spy(new AsyncCompleteProcessor(retryLatch));
         TaskRequest request = new TaskRequest(
                 new TopicPartition("topic", 1), 1, null, "TEST".getBytes(StandardCharsets.UTF_8), null, null, REQUEST.toByteArray(), null);
         DecatonTask<byte[]> task = new DecatonTask<>(

testing/build.gradle

@@ -39,5 +39,6 @@ dependencies {
     implementation "org.junit.jupiter:junit-jupiter:$junitVersion"
     implementation "org.hamcrest:hamcrest:$hamcrestVersion"
 
-    testRuntimeOnly "ch.qos.logback:logback-classic:1.4.11"
+    // We keep using 1.3.x for a while until we drop Java 8 support.
+    runtimeOnly "ch.qos.logback:logback-classic:1.3.14"
 }