KAFKA-17639 Add Java 23 to CI build matrix (#17409)

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

.github/workflows/build.yml

@@ -104,7 +104,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        java: [ 21, 11 ]  # If we change these, make sure to adjust ci-complete.yml
+        java: [ 23, 11 ]  # If we change these, make sure to adjust ci-complete.yml
     name: JUnit tests Java ${{ matrix.java }}
     steps:
       - name: Checkout code

.github/workflows/ci-complete.yml

@@ -43,7 +43,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        java: [ 21, 11 ]
+        java: [ 23, 11 ]
     steps:
       - name: Env
         run: printenv

clients/src/main/java/org/apache/kafka/common/internals/KafkaFutureImpl.java

@@ -163,7 +163,10 @@ public class KafkaFutureImpl<T> extends KafkaFuture<T> {
     public T get() throws InterruptedException, ExecutionException {
         try {
             return completableFuture.get();
-        } catch (ExecutionException e) {
+            // In Java 23, When a CompletableFuture is cancelled, get() will throw a CancellationException wrapping a
+            // CancellationException, thus we need to unwrap it to maintain the KafkaFuture behaviour. 
+            // see https://bugs.openjdk.org/browse/JDK-8331987
+        } catch (ExecutionException | CancellationException e) {
             maybeThrowCancellationException(e.getCause());
             throw e;
         }
@@ -178,7 +181,10 @@ public class KafkaFutureImpl<T> extends KafkaFuture<T> {
             TimeoutException {
         try {
             return completableFuture.get(timeout, unit);
-        } catch (ExecutionException e) {
+            // In Java 23, When a CompletableFuture is cancelled, get() will throw a CancellationException wrapping a
+            // CancellationException, thus we need to unwrap it to maintain the KafkaFuture behaviour. 
+            // see https://bugs.openjdk.org/browse/JDK-8331987
+        } catch (ExecutionException | CancellationException e) {
             maybeThrowCancellationException(e.getCause());
             throw e;
         }
@@ -192,6 +198,15 @@ public class KafkaFutureImpl<T> extends KafkaFuture<T> {
     public T getNow(T valueIfAbsent) throws ExecutionException {
         try {
             return completableFuture.getNow(valueIfAbsent);
+        } catch (CancellationException e) {
+            // In Java 23, When a CompletableFuture is cancelled, getNow() will throw a CancellationException wrapping a 
+            // CancellationException. whereas in Java < 23, it throws a CompletionException directly.
+            // see https://bugs.openjdk.org/browse/JDK-8331987
+            if (e.getCause() instanceof CancellationException) {
+                throw (CancellationException) e.getCause();
+            } else {
+                throw e;
+            }
         } catch (CompletionException e) {
             maybeThrowCancellationException(e.getCause());
             // Note, unlike CompletableFuture#get() which throws ExecutionException, CompletableFuture#getNow()
@@ -247,6 +262,15 @@ public class KafkaFutureImpl<T> extends KafkaFuture<T> {
         Throwable exception = null;
         try {
             value = completableFuture.getNow(null);
+        } catch (CancellationException e) {
+            // In Java 23, When a CompletableFuture is cancelled, getNow() will throw a CancellationException wrapping a 
+            // CancellationException. whereas in Java < 23, it throws a CompletionException directly.
+            // see https://bugs.openjdk.org/browse/JDK-8331987
+            if (e.getCause() instanceof CancellationException) {
+                exception = e.getCause();
+            } else { 
+                exception = e;
+            }
         } catch (CompletionException e) {
             exception = e.getCause();
         } catch (Exception e) {

core/src/test/scala/integration/kafka/api/SaslPlainSslEndToEndAuthorizationTest.scala

@@ -23,6 +23,7 @@ import kafka.utils.TestUtils.isAclSecure
 import kafka.zk.ZkData
 import org.apache.kafka.common.config.SaslConfigs
 import org.apache.kafka.common.config.internals.BrokerSecurityConfigs
+import org.apache.kafka.common.internals.SecurityManagerCompatibility
 import org.apache.kafka.common.network.ConnectionMode
 import org.apache.kafka.common.security.auth._
 import org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder
@@ -31,9 +32,7 @@ import org.apache.kafka.test.TestSslUtils
 import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
 import org.junit.jupiter.api.Test
 
-import java.security.AccessController
 import java.util.{Collections, Optional, Properties}
-import javax.security.auth.Subject
 import javax.security.auth.callback._
 import javax.security.auth.login.AppConfigurationEntry
 import scala.collection.Seq
@@ -89,7 +88,7 @@ object SaslPlainSslEndToEndAuthorizationTest {
   class TestClientCallbackHandler extends AuthenticateCallbackHandler {
     def configure(configs: java.util.Map[String, _], saslMechanism: String, jaasConfigEntries: java.util.List[AppConfigurationEntry]): Unit = {}
     def handle(callbacks: Array[Callback]): Unit = {
-      val subject = Subject.getSubject(AccessController.getContext)
+      val subject = SecurityManagerCompatibility.get().current()
       val username = subject.getPublicCredentials(classOf[String]).iterator().next()
       for (callback <- callbacks) {
         callback match {

jmh-benchmarks/src/main/java/org/apache/kafka/jmh/util/ByteUtilsBenchmark.java

@@ -41,10 +41,6 @@ import java.nio.ByteBuffer;
 import java.security.SecureRandom;
 import java.util.concurrent.TimeUnit;
 
-@OutputTimeUnit(TimeUnit.SECONDS)
-@Fork(3)
-@Warmup(iterations = 3, time = 1)
-@Measurement(iterations = 5, time = 1)
 /**
  * This benchmark calculates the empirical evidence of different implementation for encoding/decoding a protobuf
  * <a href="https://protobuf.dev/programming-guides/encoding/#varints">VarInt</a> and VarLong.
@@ -52,6 +48,10 @@ import java.util.concurrent.TimeUnit;
  * The benchmark uses JMH and calculates results for different sizes of variable length integer. We expect most of the
  * usage in Kafka code base to be 1 or 2 byte integers.
  */
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Fork(3)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
 public class ByteUtilsBenchmark {
     private static final int DATA_SET_SAMPLE_SIZE = 16384;