KAFKA-13273: Add support for Java 17

Jenkinsfile

@@ -142,10 +142,10 @@ pipeline {
           }
         }
 
-        stage('JDK 16 and Scala 2.13') {
+        stage('JDK 17 and Scala 2.13') {
           agent { label 'ubuntu' }
           tools {
-            jdk 'jdk_16_latest'
+            jdk 'jdk_17_latest'
           }
           options {
             timeout(time: 8, unit: 'HOURS') 
@@ -157,7 +157,7 @@ pipeline {
           steps {
             doValidation()
             doTest(env)
-            echo 'Skipping Kafka Streams archetype test for Java 16'
+            echo 'Skipping Kafka Streams archetype test for Java 17'
           }
         }
 
@@ -231,14 +231,14 @@ pipeline {
           }
         }
 
-        stage('JDK 16 and Scala 2.12') {
+        stage('JDK 17 and Scala 2.12') {
           when {
             not { changeRequest() }
             beforeAgent true
           }
           agent { label 'ubuntu' }
           tools {
-            jdk 'jdk_16_latest'
+            jdk 'jdk_17_latest'
           }
           options {
             timeout(time: 8, unit: 'HOURS') 
@@ -250,7 +250,7 @@ pipeline {
           steps {
             doValidation()
             doTest(env)
-            echo 'Skipping Kafka Streams archetype test for Java 16'
+            echo 'Skipping Kafka Streams archetype test for Java 17'
           }
         }
       }

README.md

@@ -4,7 +4,7 @@ See our [web site](https://kafka.apache.org) for details on the project.
 
 You need to have [Java](http://www.oracle.com/technetwork/java/javase/downloads/index.html) installed.
 
-We build and test Apache Kafka with Java 8, 11 and 16. We set the `release` parameter in javac and scalac
+We build and test Apache Kafka with Java 8, 11 and 17. We set the `release` parameter in javac and scalac
 to `8` to ensure the generated binaries are compatible with Java 8 or higher (independently of the Java version
 used for compilation). Java 8 support has been deprecated since Apache Kafka 3.0 and will be removed in Apache
 Kafka 4.0 (see [KIP-750](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181308223) for more details).

build.gradle

@@ -102,8 +102,22 @@ ext {
 
   defaultMaxHeapSize = "2g"
   defaultJvmArgs = ["-Xss4m", "-XX:+UseParallelGC"]
-  if (JavaVersion.current() == JavaVersion.VERSION_16)  
-    defaultJvmArgs.add("--illegal-access=permit")
+
+  // "JEP 403: Strongly Encapsulate JDK Internals" causes some tests to fail when they try
+  // to access internals (often via mocking libraries). We use `--add-opens` as a workaround
+  // for now and we'll fix it properly (where possible) via KAFKA-13275.
+  if (JavaVersion.current().isCompatibleWith(JavaVersion.VERSION_16))
+    defaultJvmArgs.addAll(
+      "--add-opens=java.base/java.io=ALL-UNNAMED",
+      "--add-opens=java.base/java.nio=ALL-UNNAMED",
+      "--add-opens=java.base/java.nio.file=ALL-UNNAMED",
+      "--add-opens=java.base/java.util.concurrent=ALL-UNNAMED",
+      "--add-opens=java.base/java.util.regex=ALL-UNNAMED",
+      "--add-opens=java.base/java.util.stream=ALL-UNNAMED",
+      "--add-opens=java.base/java.text=ALL-UNNAMED",
+      "--add-opens=java.base/java.time=ALL-UNNAMED",
+      "--add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED"
+    )
 
   userMaxForks = project.hasProperty('maxParallelForks') ? maxParallelForks.toInteger() : null
   userIgnoreFailures = project.hasProperty('ignoreFailures') ? ignoreFailures : false
@@ -359,7 +373,7 @@ subprojects {
   // The suites are for running sets of tests in IDEs.
   // Gradle will run each test class, so we exclude the suites to avoid redundantly running the tests twice.
   def testsToExclude = ['**/*Suite.class']
-  // Exclude PowerMock tests when running with Java 16 until a version of PowerMock that supports Java 16 is released
+  // Exclude PowerMock tests when running with Java 16 or newer until a version of PowerMock that supports the relevant versions is released
   // The relevant issues are https://github.com/powermock/powermock/issues/1094 and https://github.com/powermock/powermock/issues/1099
   if (JavaVersion.current().isCompatibleWith(JavaVersion.VERSION_16)) {
     testsToExclude.addAll([
@@ -372,8 +386,7 @@ subprojects {
       "**/WorkerSinkTaskTest.*", "**/WorkerSinkTaskThreadedTest.*", "**/WorkerSourceTaskTest.*",
       "**/WorkerTaskTest.*", "**/WorkerTest.*", "**/RestServerTest.*",
       // streams tests
-      "**/KafkaStreamsTest.*", "**/RepartitionTopicsTest.*", "**/RocksDBMetricsRecorderTest.*",
-      "**/StreamsMetricsImplTest.*", "**/StateManagerUtilTest.*", "**/TableSourceNodeTest.*"
+      "**/KafkaStreamsTest.*"
     ])
   }
 

docs/upgrade.html

@@ -19,7 +19,12 @@
 
 <script id="upgrade-template" type="text/x-handlebars-template">
 
-<h5><a id="upgrade_300_notable" href="#upgrade_300_notable">Notable changes in 3.0.0</a></h5>
+<h5><a id="upgrade_310_notable" href="#upgrade_310_notable">Notable changes in 3.1.0</a></h5>
+<ul>
+    <li>Apache Kafka supports Java 17.</li>
+</ul>
+
+<h5><a id="upgrade_310_notable" href="#upgrade_300_notable">Notable changes in 3.0.0</a></h5>
 <ul>
     <li>ZooKeeper has been upgraded to version 3.6.3.</li>
     <li>A preview of KRaft mode is available, though upgrading to it from the 2.8 Early Access release is not possible. See

gradle/dependencies.gradle

@@ -62,7 +62,7 @@ versions += [
   checkstyle: "8.36.2",
   commonsCli: "1.4",
   dropwizardMetrics: "4.1.12.1",
-  gradle: "7.1.1",
+  gradle: "7.2",
   grgit: "4.1.0",
   httpclient: "4.5.13",
   easymock: "4.3",
@@ -100,7 +100,7 @@ versions += [
   lz4: "1.7.1",
   mavenArtifact: "3.8.1",
   metrics: "2.2.0",
-  mockito: "3.9.0",
+  mockito: "3.12.4",
   netty: "4.1.62.Final",
   powermock: "2.0.9",
   reflections: "0.9.12",

gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,6 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionSha256Sum=9bb8bc05f562f2d42bdf1ba8db62f6b6fa1c3bf6c392228802cc7cb0578fe7e0
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.1.1-all.zip
+distributionSha256Sum=a8da5b02437a60819cad23e10fc7e9cf32bcb57029d9cb277e26eeff76ce014b
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.2-all.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java

@@ -101,9 +101,7 @@ public void testRejectVotesFromSameEpochAfterResigningLeadership() throws Except
         int epoch = 2;
 
         RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
-            .updateRandom(random -> {
-                Mockito.doReturn(0).when(random).nextInt(DEFAULT_ELECTION_TIMEOUT_MS);
-            })
+            .updateRandom(r -> r.mockNextInt(DEFAULT_ELECTION_TIMEOUT_MS, 0))
             .withElectedLeader(epoch, localId)
             .build();
 
@@ -126,9 +124,7 @@ public void testRejectVotesFromSameEpochAfterResigningCandidacy() throws Excepti
         int epoch = 2;
 
         RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
-            .updateRandom(random -> {
-                Mockito.doReturn(0).when(random).nextInt(DEFAULT_ELECTION_TIMEOUT_MS);
-            })
+            .updateRandom(r -> r.mockNextInt(DEFAULT_ELECTION_TIMEOUT_MS, 0))
             .withVotedCandidate(epoch, localId)
             .build();
 
@@ -151,11 +147,9 @@ public void testGrantVotesFromHigherEpochAfterResigningLeadership() throws Excep
         int epoch = 2;
 
         RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
-                .updateRandom(random -> {
-                    Mockito.doReturn(0).when(random).nextInt(DEFAULT_ELECTION_TIMEOUT_MS);
-                })
-                .withElectedLeader(epoch, localId)
-                .build();
+            .updateRandom(r -> r.mockNextInt(DEFAULT_ELECTION_TIMEOUT_MS, 0))
+            .withElectedLeader(epoch, localId)
+            .build();
 
         // Resign from leader, will restart in resigned state
         assertTrue(context.client.quorum().isResigned());
@@ -181,11 +175,9 @@ public void testGrantVotesFromHigherEpochAfterResigningCandidacy() throws Except
         int epoch = 2;
 
         RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
-                .updateRandom(random -> {
-                    Mockito.doReturn(0).when(random).nextInt(DEFAULT_ELECTION_TIMEOUT_MS);
-                })
-                .withVotedCandidate(epoch, localId)
-                .build();
+            .updateRandom(r -> r.mockNextInt(DEFAULT_ELECTION_TIMEOUT_MS, 0))
+            .withVotedCandidate(epoch, localId)
+            .build();
 
         // Resign from candidate, will restart in candidate state
         assertTrue(context.client.quorum().isCandidate());
@@ -235,11 +227,9 @@ public void testInitializeAsResignedAndBecomeCandidate() throws Exception {
         int epoch = 2;
 
         RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
-                .updateRandom(random -> {
-                    Mockito.doReturn(0).when(random).nextInt(DEFAULT_ELECTION_TIMEOUT_MS);
-                })
-                .withElectedLeader(epoch, localId)
-                .build();
+            .updateRandom(r -> r.mockNextInt(DEFAULT_ELECTION_TIMEOUT_MS, 0))
+            .withElectedLeader(epoch, localId)
+            .build();
 
         // Resign from leader, will restart in resigned state
         assertTrue(context.client.quorum().isResigned());
@@ -262,9 +252,7 @@ public void testInitializeAsResignedLeaderFromStateStore() throws Exception {
         int epoch = 2;
 
         RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
-            .updateRandom(random -> {
-                Mockito.doReturn(0).when(random).nextInt(DEFAULT_ELECTION_TIMEOUT_MS);
-            })
+            .updateRandom(r -> r.mockNextInt(DEFAULT_ELECTION_TIMEOUT_MS, 0))
             .withElectedLeader(epoch, localId)
             .build();
 
@@ -728,9 +716,7 @@ public void testEndQuorumIgnoredAsCandidateIfOlderEpoch() throws Exception {
         Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
 
         RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
-            .updateRandom(random -> {
-                Mockito.doReturn(jitterMs).when(random).nextInt(Mockito.anyInt());
-            })
+            .updateRandom(r -> r.mockNextInt(jitterMs))
             .withUnknownLeader(epoch - 1)
             .build();
 
@@ -1238,9 +1224,7 @@ public void testRetryElection() throws Exception {
         Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
 
         RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
-            .updateRandom(random -> {
-                Mockito.doReturn(exponentialFactor).when(random).nextInt(Mockito.anyInt());
-            })
+            .updateRandom(r -> r.mockNextInt(exponentialFactor))
             .build();
 
         context.assertUnknownLeader(0);
@@ -2184,9 +2168,7 @@ public void testFetchShouldBeTreatedAsLeaderAcknowledgement() throws Exception {
         Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
 
         RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
-            .updateRandom(random -> {
-                Mockito.doReturn(0).when(random).nextInt(DEFAULT_ELECTION_TIMEOUT_MS);
-            })
+            .updateRandom(r -> r.mockNextInt(DEFAULT_ELECTION_TIMEOUT_MS, 0))
             .withUnknownLeader(epoch - 1)
             .build();
 
@@ -2395,9 +2377,7 @@ public void testClusterAuthorizationFailedInBeginQuorumEpoch() throws Exception
         Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
 
         RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
-            .updateRandom(random -> {
-                Mockito.doReturn(0).when(random).nextInt(DEFAULT_ELECTION_TIMEOUT_MS);
-            })
+            .updateRandom(r -> r.mockNextInt(DEFAULT_ELECTION_TIMEOUT_MS, 0))
             .withUnknownLeader(epoch - 1)
             .build();
 
@@ -2799,4 +2779,5 @@ public void testObserverFetchWithNoLocalId() throws Exception {
     private static KafkaMetric getMetric(final Metrics metrics, final String name) {
         return metrics.metrics().get(metrics.metricName(name, "raft-metrics"));
     }
+
 }

raft/src/test/java/org/apache/kafka/raft/MockableRandom.java

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+package org.apache.kafka.raft;
+
+import java.util.OptionalInt;
+import java.util.Random;
+import java.util.function.IntFunction;
+
+/**
+ * A Random instance that makes it easy to modify the behavior of certain methods for test purposes.
+ */
+class MockableRandom extends Random {
+
+    private IntFunction<OptionalInt> nextIntFunction = __ -> OptionalInt.empty();
+
+    public MockableRandom(long seed) {
+        super(seed);
+    }
+
+    public void mockNextInt(int expectedBound, int returnValue) {
+        this.nextIntFunction = b -> {
+            if (b == expectedBound)
+                return OptionalInt.of(returnValue);
+            else
+                return OptionalInt.empty();
+        };
+    }
+
+    public void mockNextInt(int returnValue) {
+        this.nextIntFunction = __ -> OptionalInt.of(returnValue);
+    }
+
+    @Override
+    public int nextInt(int bound) {
+        return nextIntFunction.apply(bound).orElse(super.nextInt(bound));
+    }
+}