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,18 @@ ext {
 
   defaultMaxHeapSize = "2g"
   defaultJvmArgs = ["-Xss4m", "-XX:+UseParallelGC"]
-  if (JavaVersion.current() == JavaVersion.VERSION_16)  
-    defaultJvmArgs.add("--illegal-access=permit")
+  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 +369,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 +382,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);
-            })
+            .mockRandomNextInt(KafkaRaftClientTest::return0IfBoundEqualsElectionTimeout)
             .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);
-            })
+            .mockRandomNextInt(KafkaRaftClientTest::return0IfBoundEqualsElectionTimeout)
             .withVotedCandidate(epoch, localId)
             .build();
 
@@ -151,9 +147,7 @@ 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);
-                })
+                .mockRandomNextInt(KafkaRaftClientTest::return0IfBoundEqualsElectionTimeout)
                 .withElectedLeader(epoch, localId)
                 .build();
 
@@ -181,9 +175,7 @@ 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);
-                })
+                .mockRandomNextInt(KafkaRaftClientTest::return0IfBoundEqualsElectionTimeout)
                 .withVotedCandidate(epoch, localId)
                 .build();
 
@@ -235,9 +227,7 @@ 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);
-                })
+                .mockRandomNextInt(KafkaRaftClientTest::return0IfBoundEqualsElectionTimeout)
                 .withElectedLeader(epoch, localId)
                 .build();
 
@@ -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);
-            })
+            .mockRandomNextInt(KafkaRaftClientTest::return0IfBoundEqualsElectionTimeout)
             .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());
-            })
+            .mockRandomNextInt(__ -> Optional.of(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());
-            })
+            .mockRandomNextInt(__ -> Optional.of(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);
-            })
+            .mockRandomNextInt(KafkaRaftClientTest::return0IfBoundEqualsElectionTimeout)
             .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);
-            })
+            .mockRandomNextInt(KafkaRaftClientTest::return0IfBoundEqualsElectionTimeout)
             .withUnknownLeader(epoch - 1)
             .build();
 
@@ -2799,4 +2779,11 @@ public void testObserverFetchWithNoLocalId() throws Exception {
     private static KafkaMetric getMetric(final Metrics metrics, final String name) {
         return metrics.metrics().get(metrics.metricName(name, "raft-metrics"));
     }
+
+    private static Optional<Integer> return0IfBoundEqualsElectionTimeout(int bound) {
+        if (bound == DEFAULT_ELECTION_TIMEOUT_MS)
+            return Optional.of(0);
+        else
+            return Optional.empty();
+    }
 }

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

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.raft;
 
+import java.util.function.IntFunction;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.memory.MemoryPool;
@@ -60,7 +61,6 @@
 import org.apache.kafka.snapshot.SnapshotReader;
 import org.apache.kafka.test.TestCondition;
 import org.apache.kafka.test.TestUtils;
-import org.mockito.Mockito;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
@@ -77,7 +77,6 @@
 import java.util.OptionalLong;
 import java.util.Random;
 import java.util.Set;
-import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
 import static org.apache.kafka.raft.RaftUtil.hasValidTopicPartition;
@@ -128,7 +127,7 @@ public static final class Builder {
         private final MockMessageQueue messageQueue = new MockMessageQueue();
         private final MockTime time = new MockTime();
         private final QuorumStateStore quorumStateStore = new MockQuorumStateStore();
-        private final Random random = Mockito.spy(new Random(1));
+        private final MockableRandom random = new MockableRandom(1);
         private final LogContext logContext = new LogContext();
         private final MockLog log = new MockLog(METADATA_PARTITION,  Uuid.METADATA_TOPIC_ID, logContext);
         private final Set<Integer> voters;
@@ -164,8 +163,12 @@ Builder withVotedCandidate(int epoch, int votedId) throws IOException {
             return this;
         }
 
-        Builder updateRandom(Consumer<Random> consumer) {
-            consumer.accept(random);
+        /**
+         * If the function returns an empty option, `random.nextInt` behaves as usual. Otherwise, the
+         * integer returned from `function` is returned by `random.nextInt`.
+         */
+        Builder mockRandomNextInt(IntFunction<Optional<Integer>> function) {
+            random.mockNextInt(function);
             return this;
         }
 
@@ -273,6 +276,28 @@ public RaftClientTestContext build() throws IOException {
         }
     }
 
+    private static class MockableRandom extends Random {
+
+        private IntFunction<Optional<Integer>> nextIntFunction = __ -> Optional.empty();
+
+        public MockableRandom(long seed) {
+            super(seed);
+        }
+
+        /**
+         * If the function returns an empty option, `nextInt` behaves as usual. Otherwise, the
+         * integer returned from `function` is returned by `nextInt`.
+         */
+        public void mockNextInt(IntFunction<Optional<Integer>> function) {
+            this.nextIntFunction = function;
+        }
+
+        @Override
+        public int nextInt(int bound) {
+            return nextIntFunction.apply(bound).orElse(super.nextInt(bound));
+        }
+    }
+
     private RaftClientTestContext(
         Uuid clusterId,
         OptionalInt localId,

release.py

@@ -256,7 +256,7 @@ def command_stage_docs():
         sys.exit("%s doesn't exist or does not appear to be the kafka-site repository" % kafka_site_repo_path)
 
     prefs = load_prefs()
-    jdk15_env = get_jdk(prefs, 15)
+    jdk17_env = get_jdk(prefs, 17)
     save_prefs(prefs)
 
     version = get_version()
@@ -265,7 +265,7 @@ def command_stage_docs():
     # version due to already having bumped the bugfix version number.
     gradle_version_override = docs_release_version(version)
 
-    cmd("Building docs", "./gradlew -Pversion=%s clean siteDocsTar aggregatedJavadoc" % gradle_version_override, cwd=REPO_HOME, env=jdk15_env)
+    cmd("Building docs", "./gradlew -Pversion=%s clean siteDocsTar aggregatedJavadoc" % gradle_version_override, cwd=REPO_HOME, env=jdk17_env)
 
     docs_tar = os.path.join(REPO_HOME, 'core', 'build', 'distributions', 'kafka_2.13-%s-site-docs.tgz' % gradle_version_override)
 
@@ -426,7 +426,7 @@ def command_release_announcement_email():
 
 if not user_ok("""Requirements:
 1. Updated docs to reference the new release version where appropriate.
-2. JDK8 and JDK15 compilers and libraries
+2. JDK8 and JDK17 compilers and libraries
 3. Your Apache ID, already configured with SSH keys on id.apache.org and SSH keys available in this shell session
 4. All issues in the target release resolved with valid resolutions (if not, this script will report the problematic JIRAs)
 5. A GPG key used for signing the release. This key should have been added to public Apache servers and the KEYS file on the Kafka site
@@ -511,7 +511,7 @@ def command_release_announcement_email():
 apache_id = get_pref(prefs, 'apache_id', lambda: raw_input("Enter your apache username: "))
 
 jdk8_env = get_jdk(prefs, 8)
-jdk15_env = get_jdk(prefs, 15)
+jdk17_env = get_jdk(prefs, 17)
 
 def select_gpg_key():
     print("Here are the available GPG keys:")
@@ -600,7 +600,7 @@ def select_gpg_key():
 
 cmd("Building artifacts", "./gradlew clean && ./gradlewAll releaseTarGz", cwd=kafka_dir, env=jdk8_env, shell=True)
 cmd("Copying artifacts", "cp %s/core/build/distributions/* %s" % (kafka_dir, artifacts_dir), shell=True)
-cmd("Building docs", "./gradlew clean aggregatedJavadoc", cwd=kafka_dir, env=jdk15_env)
+cmd("Building docs", "./gradlew clean aggregatedJavadoc", cwd=kafka_dir, env=jdk17_env)
 cmd("Copying docs", "cp -R %s/build/docs/javadoc %s" % (kafka_dir, artifacts_dir))
 
 for filename in os.listdir(artifacts_dir):