Scala3 migration

Jenkinsfile

@@ -161,6 +161,25 @@ pipeline {
           }
         }
 
+        stage('JDK 11 and Scala 3.0') {
+          agent { label 'ubuntu' }
+          tools {
+            jdk 'jdk_11_latest'
+          }
+          options {
+            timeout(time: 8, unit: 'HOURS')
+            timestamps()
+          }
+          environment {
+            SCALA_VERSION=3.0
+          }
+          steps {
+            doValidation()
+            doTest(env)
+            echo 'Skipping Kafka Streams archetype test for Java 16'
+          }
+        }
+
         stage('ARM') {
           agent { label 'arm4' }
           options {
@@ -208,6 +227,76 @@ pipeline {
           }
         }
 
+        stage('JDK 8 and Scala 3.0') {
+          when {
+            not { changeRequest() }
+            beforeAgent true
+          }
+          agent { label 'ubuntu' }
+          tools {
+            jdk 'jdk_1.8_latest'
+            maven 'maven_3_latest'
+          }
+          options {
+            timeout(time: 8, unit: 'HOURS')
+            timestamps()
+          }
+          environment {
+            SCALA_VERSION=3.0
+          }
+          steps {
+            doValidation()
+            doTest(env)
+            tryStreamsArchetype()
+          }
+        }
+
+        stage('JDK 15 and Scala 3.0') {
+          when {
+            not { changeRequest() }
+            beforeAgent true
+          }
+          agent { label 'ubuntu' }
+          tools {
+            jdk 'jdk_15_latest'
+          }
+          options {
+            timeout(time: 8, unit: 'HOURS')
+            timestamps()
+          }
+          environment {
+            SCALA_VERSION=3.0
+          }
+          steps {
+            doValidation()
+            doTest(env)
+            tryStreamsArchetype()
+          }
+        }
+
+        stage('JDK 16 and Scala 3.0') {
+          when {
+            not { changeRequest() }
+            beforeAgent true
+          }
+          agent { label 'ubuntu' }
+          tools {
+            jdk 'jdk_16_latest'
+          }
+          options {
+            timeout(time: 8, unit: 'HOURS')
+            timestamps()
+          }
+          environment {
+            SCALA_VERSION=3.0
+          }
+          steps {
+            doValidation()
+            doTest(env)
+            tryStreamsArchetype()
+          }
+        }
+
         stage('JDK 11 and Scala 2.12') {
           when {
             not { changeRequest() }

build.gradle

@@ -83,7 +83,7 @@ allprojects {
           libs.javassist,
           // ensure we have a single version in the classpath despite transitive dependencies
           libs.scalaLibrary,
-          libs.scalaReflect,
+          libs.scalaCompiler,
           libs.jacksonAnnotations,
           // be explicit about the Netty dependency version instead of relying on the version set by
           // ZooKeeper (potentially older and containing CVEs)
@@ -562,27 +562,33 @@ subprojects {
 
   tasks.withType(ScalaCompile) {
     scalaCompileOptions.additionalParameters = [
-      "-deprecation",
-      "-unchecked",
-      "-encoding", "utf8",
-      "-Xlog-reflective-calls",
-      "-feature",
-      "-language:postfixOps",
-      "-language:implicitConversions",
-      "-language:existentials",
-      "-Xlint:constant",
-      "-Xlint:delayedinit-select",
-      "-Xlint:doc-detached",
-      "-Xlint:missing-interpolator",
-      "-Xlint:nullary-unit",
-      "-Xlint:option-implicit",
-      "-Xlint:package-object-classes",
-      "-Xlint:poly-implicit-overload",
-      "-Xlint:private-shadow",
-      "-Xlint:stars-align",
-      "-Xlint:type-parameter-shadow",
-      "-Xlint:unused"
+            "-unchecked",
+            "-encoding",
+            "UTF-8",
+            "-feature",
+            "-language:postfixOps",
+            "-language:implicitConversions"
     ]
+    if (versions.baseScala == '2.13') {
+      scalaCompileOptions.additionalParameters += [
+              "-deprecation",
+              "-Xlog-reflective-calls",
+              "-language:existentials",
+              "-Xlint:constant",
+              "-Xlint:delayedinit-select",
+              "-Xlint:doc-detached",
+              "-Xlint:missing-interpolator",
+              "-Xlint:nullary-unit",
+              "-Xlint:option-implicit",
+              "-Xlint:package-object-classes",
+              "-Xlint:poly-implicit-overload",
+              "-Xlint:private-shadow",
+              "-Xlint:stars-align",
+              "-Xlint:type-parameter-shadow",
+              "-Ytasty-reader",
+              "-Xlint:unused"
+      ]
+    }
 
     // See README.md for details on this option and the meaning of each value
     if (userScalaOptimizerMode.equals("method"))
@@ -598,12 +604,16 @@ subprojects {
       scalaCompileOptions.additionalParameters += inlineFrom
     }
 
-    if (versions.baseScala != '2.12') {
+    if (versions.baseScala == '2.13') {
       scalaCompileOptions.additionalParameters += ["-opt-warnings", "-Xlint:strict-unsealed-patmat"]
       // Scala 2.13.2 introduces compiler warnings suppression, which is a pre-requisite for -Xfatal-warnings
       scalaCompileOptions.additionalParameters += ["-Xfatal-warnings"]
     }
 
+    if (versions.baseScala == '3') {
+      scalaCompileOptions.additionalParameters += ["-source:3.0-migration"]
+    }
+
     // these options are valid for Scala versions < 2.13 only
     // Scala 2.13 removes them, see https://github.com/scala/scala/pull/6502 and https://github.com/scala/scala/pull/5969
     if (versions.baseScala == '2.12') {
@@ -817,6 +827,7 @@ project(':core') {
     // even though the `core` module doesn't expose any public API
     api project(':clients')
     api libs.scalaLibrary
+    api libs.scalaCompiler
 
     implementation project(':server-common')
     implementation project(':metadata')
@@ -833,7 +844,6 @@ project(':core') {
     implementation libs.scalaCollectionCompat
     implementation libs.scalaJava8Compat
     // only needed transitively, but set it explicitly to ensure it has the same version as scala-library
-    implementation libs.scalaReflect
     implementation libs.scalaLogging
     implementation libs.slf4jApi
     implementation(libs.zookeeper) {
@@ -1868,6 +1878,7 @@ project(':streams:streams-scala') {
     api project(':streams')
 
     api libs.scalaLibrary
+    api libs.scalaCompiler
     api libs.scalaCollectionCompat
 
     testImplementation project(':core')
@@ -2197,6 +2208,7 @@ project(':jmh-benchmarks') {
     implementation libs.slf4jlog4j
     implementation libs.scalaLibrary
     implementation libs.scalaJava8Compat
+    implementation libs.scalaCompiler
   }
 
   tasks.withType(JavaCompile) {

core/src/main/java/kafka/server/builders/LogManagerBuilder.java

@@ -18,6 +18,7 @@
 package kafka.server.builders;
 
 import kafka.api.ApiVersion;
+import kafka.api.ApiVersion$;
 import kafka.log.CleanerConfig;
 import kafka.log.LogConfig;
 import kafka.log.LogManager;
@@ -45,7 +46,7 @@ public class LogManagerBuilder {
     private long flushStartOffsetCheckpointMs = 10000L;
     private long retentionCheckMs = 1000L;
     private int maxPidExpirationMs = 60000;
-    private ApiVersion interBrokerProtocolVersion = ApiVersion.latestVersion();
+    private ApiVersion interBrokerProtocolVersion = ApiVersion$.MODULE$.latestVersion();
     private Scheduler scheduler = null;
     private BrokerTopicStats brokerTopicStats = null;
     private LogDirFailureChannel logDirFailureChannel = null;

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

@@ -1036,8 +1036,8 @@ object ConsumerGroupCommand extends Logging {
     val timeoutMsOpt = parser.accepts("timeout", TimeoutMsDoc)
                              .withRequiredArg
                              .describedAs("timeout (ms)")
-                             .ofType(classOf[Long])
-                             .defaultsTo(5000)
+                             .ofType(classOf[java.lang.Long])
+                             .defaultsTo(5000L)
     val commandConfigOpt = parser.accepts("command-config", CommandConfigDoc)
                                   .withRequiredArg
                                   .describedAs("command config property file")

core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala

@@ -238,14 +238,14 @@ object ReassignPartitionsCommand extends Logging {
       executeAssignment(adminClient,
         opts.options.has(opts.additionalOpt),
         Utils.readFileAsString(opts.options.valueOf(opts.reassignmentJsonFileOpt)),
-        opts.options.valueOf(opts.interBrokerThrottleOpt),
-        opts.options.valueOf(opts.replicaAlterLogDirsThrottleOpt),
-        opts.options.valueOf(opts.timeoutOpt))
+        opts.options.valueOf(opts.interBrokerThrottleOpt).longValue(),
+        opts.options.valueOf(opts.replicaAlterLogDirsThrottleOpt).longValue(),
+        opts.options.valueOf(opts.timeoutOpt).longValue())
     } else if (opts.options.has(opts.cancelOpt)) {
       cancelAssignment(adminClient,
         Utils.readFileAsString(opts.options.valueOf(opts.reassignmentJsonFileOpt)),
         opts.options.has(opts.preserveThrottlesOpt),
-        opts.options.valueOf(opts.timeoutOpt))
+        opts.options.valueOf(opts.timeoutOpt).longValue())
     } else if (opts.options.has(opts.listOpt)) {
       listReassignments(adminClient)
     } else {
@@ -1473,25 +1473,25 @@ object ReassignPartitionsCommand extends Logging {
                       .ofType(classOf[String])
     val disableRackAware = parser.accepts("disable-rack-aware", "Disable rack aware replica assignment")
     val interBrokerThrottleOpt = parser.accepts("throttle", "The movement of partitions between brokers will be throttled to this value (bytes/sec). " +
-      "This option can be included with --execute when a reassignment is started, and it can be altered by resubmitting the current reassignment " +
-      "along with the --additional flag. The throttle rate should be at least 1 KB/s.")
-      .withRequiredArg()
-      .describedAs("throttle")
-      .ofType(classOf[Long])
-      .defaultsTo(-1)
+                      "This option can be included with --execute when a reassignment is started, and it can be altered by resubmitting the current reassignment " +
+                      "along with the --additional flag. The throttle rate should be at least 1 KB/s.")
+                      .withRequiredArg()
+                      .describedAs("throttle")
+                      .ofType(classOf[java.lang.Long])
+                      .defaultsTo(-1L)
     val replicaAlterLogDirsThrottleOpt = parser.accepts("replica-alter-log-dirs-throttle",
-      "The movement of replicas between log directories on the same broker will be throttled to this value (bytes/sec). " +
-        "This option can be included with --execute when a reassignment is started, and it can be altered by resubmitting the current reassignment " +
-        "along with the --additional flag. The throttle rate should be at least 1 KB/s.")
-      .withRequiredArg()
-      .describedAs("replicaAlterLogDirsThrottle")
-      .ofType(classOf[Long])
-      .defaultsTo(-1)
+          "The movement of replicas between log directories on the same broker will be throttled to this value (bytes/sec). " +
+                      "This option can be included with --execute when a reassignment is started, and it can be altered by resubmitting the current reassignment " +
+                      "along with the --additional flag. The throttle rate should be at least 1 KB/s.")
+                      .withRequiredArg()
+                      .describedAs("replicaAlterLogDirsThrottle")
+                      .ofType(classOf[java.lang.Long])
+                      .defaultsTo(-1L)
     val timeoutOpt = parser.accepts("timeout", "The maximum time in ms to wait for log directory replica assignment to begin.")
                       .withRequiredArg()
                       .describedAs("timeout")
-                      .ofType(classOf[Long])
-                      .defaultsTo(10000)
+                      .ofType(classOf[java.lang.Long])
+                      .defaultsTo(10000L)
     val additionalOpt = parser.accepts("additional", "Execute this reassignment in addition to any " +
       "other ongoing ones. This option can also be used to change the throttle of an ongoing reassignment.")
     val preserveThrottlesOpt = parser.accepts("preserve-throttles", "Do not modify broker or topic throttles.")

core/src/main/scala/kafka/admin/TopicCommand.scala

@@ -22,7 +22,7 @@ import java.util.{Collections, Properties}
 import joptsimple._
 import kafka.common.AdminCommandFailedException
 import kafka.log.LogConfig
-import kafka.utils._
+import kafka.utils.{immutable => _, _}
 import org.apache.kafka.clients.CommonClientConfigs
 import org.apache.kafka.clients.admin.CreatePartitionsOptions
 import org.apache.kafka.clients.admin.CreateTopicsOptions

core/src/main/scala/kafka/admin/ZkSecurityMigrator.scala

@@ -279,7 +279,7 @@ class ZkSecurityMigrator(zkClient: KafkaZkClient) extends Logging {
         }
         future match {
           case Some(a) =>
-            Await.result(a, 6000 millis)
+            Await.result(a, 6000.millis)
             futures.synchronized { futures.dequeue() }
             recurse()
           case None =>

core/src/main/scala/kafka/controller/PartitionStateMachine.scala

@@ -477,7 +477,7 @@ class ZkPartitionStateMachine(config: KafkaConfig,
     } else {
       val (logConfigs, failed) = zkClient.getLogConfigs(
         partitionsWithNoLiveInSyncReplicas.iterator.map { case (partition, _) => partition.topic }.toSet,
-        config.originals()
+        config.originals
       )
 
       partitionsWithNoLiveInSyncReplicas.map { case (partition, leaderAndIsr) =>

core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala

@@ -17,14 +17,6 @@
 
 package kafka.coordinator.group
 
-import java.io.PrintStream
-import java.nio.ByteBuffer
-import java.nio.charset.StandardCharsets
-import java.util.Optional
-import java.util.concurrent.atomic.AtomicBoolean
-import java.util.concurrent.locks.ReentrantLock
-import java.util.concurrent.{ConcurrentHashMap, TimeUnit}
-
 import com.yammer.metrics.core.Gauge
 import kafka.api.{ApiVersion, KAFKA_0_10_1_IV0, KAFKA_2_1_IV0, KAFKA_2_1_IV1, KAFKA_2_3_IV0}
 import kafka.common.OffsetAndMetadata
@@ -33,8 +25,8 @@ import kafka.log.AppendOrigin
 import kafka.metrics.KafkaMetricsGroup
 import kafka.server.{FetchLogEnd, ReplicaManager, RequestLocal}
 import kafka.utils.CoreUtils.inLock
+import kafka.utils.{immutable => _, _}
 import kafka.utils.Implicits._
-import kafka.utils._
 import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.apache.kafka.clients.consumer.internals.ConsumerProtocol
 import org.apache.kafka.common.internals.Topic
@@ -48,6 +40,13 @@ import org.apache.kafka.common.requests.{OffsetCommitRequest, OffsetFetchRespons
 import org.apache.kafka.common.utils.{Time, Utils}
 import org.apache.kafka.common.{KafkaException, MessageFormatter, TopicPartition}
 
+import java.io.PrintStream
+import java.nio.ByteBuffer
+import java.nio.charset.StandardCharsets
+import java.util.Optional
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.locks.ReentrantLock
+import java.util.concurrent.{ConcurrentHashMap, TimeUnit}
 import scala.collection._
 import scala.collection.mutable.ArrayBuffer
 import scala.jdk.CollectionConverters._

core/src/main/scala/kafka/log/LazyIndex.scala

@@ -21,7 +21,7 @@ import java.io.File
 import java.nio.file.{Files, NoSuchFileException}
 import java.util.concurrent.locks.ReentrantLock
 
-import LazyIndex._
+
 import kafka.utils.CoreUtils.inLock
 import kafka.utils.threadsafe
 import org.apache.kafka.common.utils.Utils
@@ -46,8 +46,8 @@ import org.apache.kafka.common.utils.Utils
   *                  `AbstractIndex` instance.
   */
 @threadsafe
-class LazyIndex[T <: AbstractIndex] private (@volatile private var indexWrapper: IndexWrapper, loadIndex: File => T) {
-
+class LazyIndex[T <: AbstractIndex] private (@volatile private var indexWrapper: LazyIndex.IndexWrapper, loadIndex: File => T) {
+  import LazyIndex._
   private val lock = new ReentrantLock()
 
   def file: File = indexWrapper.file

core/src/main/scala/kafka/log/LogConfig.scala

@@ -172,7 +172,7 @@ case class LogConfig(props: java.util.Map[_, _], overriddenConfigs: Set[String]
   }
 
   def overriddenConfigsAsLoggableString: String = {
-    val overriddenTopicProps = props.asScala.collect {
+    val overriddenTopicProps: Map[String, Object] = props.asScala.collect {
       case (k: String, v) if overriddenConfigs.contains(k) => (k, v.asInstanceOf[AnyRef])
     }
     ConfigUtils.configMapToRedactedString(overriddenTopicProps.asJava, configDef)