KAFKA-12342; Merge RaftClient and MetaLogManager interfaces and remove shim

build.gradle

@@ -1056,6 +1056,7 @@ project(':metadata') {
 
   dependencies {
     implementation project(':clients')
+    implementation project(':raft')
     implementation libs.jacksonDatabind
     implementation libs.jacksonJDK8Datatypes
     implementation libs.metrics
@@ -1254,7 +1255,6 @@ project(':raft') {
 
   dependencies {
     implementation project(':clients')
-    implementation project(':metadata')
     implementation libs.slf4jApi
     implementation libs.jacksonDatabind
 

checkstyle/import-control.xml

@@ -220,6 +220,7 @@
     <allow pkg="org.apache.kafka.controller" />
     <allow pkg="org.apache.kafka.metadata" />
     <allow pkg="org.apache.kafka.metalog" />
+    <allow pkg="org.apache.kafka.raft" />
     <allow pkg="org.apache.kafka.queue" />
     <allow pkg="org.apache.kafka.test" />
     <allow pkg="org.apache.kafka.timeline" />
@@ -231,6 +232,8 @@
     <allow pkg="org.apache.kafka.common.message" />
     <allow pkg="org.apache.kafka.common.metadata" />
     <allow pkg="org.apache.kafka.common.protocol" />
+    <allow pkg="org.apache.kafka.metalog" />
+    <allow pkg="org.apache.kafka.raft" />
     <allow pkg="org.apache.kafka.test" />
   </subpackage>
 
@@ -239,6 +242,8 @@
     <allow pkg="org.apache.kafka.common.protocol" />
     <allow pkg="org.apache.kafka.metadata" />
     <allow pkg="org.apache.kafka.metalog" />
+    <allow pkg="org.apache.kafka.raft" />
+    <allow pkg="org.apache.kafka.snapshot" />
     <allow pkg="org.apache.kafka.queue" />
     <allow pkg="org.apache.kafka.test" />
   </subpackage>
@@ -277,7 +282,6 @@
     <allow pkg="net.sourceforge.argparse4j" />
     <allow pkg="org.apache.kafka.common"/>
     <allow pkg="org.apache.kafka.metadata"/>
-    <allow pkg="org.apache.kafka.metalog"/>
     <allow pkg="org.apache.kafka.queue"/>
     <allow pkg="org.apache.kafka.raft"/>
     <allow pkg="org.apache.kafka.shell"/>
@@ -401,7 +405,6 @@
     <allow pkg="org.apache.kafka.common.record" />
     <allow pkg="org.apache.kafka.common.requests" />
     <allow pkg="org.apache.kafka.common.protocol" />
-    <allow pkg="org.apache.kafka.metalog" />
     <allow pkg="org.apache.kafka.test"/>
     <allow pkg="com.fasterxml.jackson" />
     <allow pkg="net.jqwik"/>

core/src/main/scala/kafka/raft/RaftManager.scala

@@ -37,7 +37,7 @@ import org.apache.kafka.common.security.JaasContext
 import org.apache.kafka.common.security.auth.SecurityProtocol
 import org.apache.kafka.common.utils.{LogContext, Time}
 import org.apache.kafka.raft.RaftConfig.{AddressSpec, InetAddressSpec, NON_ROUTABLE_ADDRESS, UnknownAddressSpec}
-import org.apache.kafka.raft.{FileBasedStateStore, KafkaRaftClient, RaftClient, RaftConfig, RaftRequest, RecordSerde}
+import org.apache.kafka.raft.{FileBasedStateStore, KafkaRaftClient, LeaderAndEpoch, RaftClient, RaftConfig, RaftRequest, RecordSerde}
 
 import scala.jdk.CollectionConverters._
 
@@ -101,6 +101,10 @@ trait RaftManager[T] {
     epoch: Int,
     records: Seq[T]
   ): Option[Long]
+
+  def leaderAndEpoch: LeaderAndEpoch
+
+  def client: RaftClient[T]
 }
 
 class KafkaRaftManager[T](
@@ -125,10 +129,10 @@ class KafkaRaftManager[T](
   private val dataDir = createDataDir()
   private val metadataLog = buildMetadataLog()
   private val netChannel = buildNetworkChannel()
-  private val raftClient = buildRaftClient()
-  private val raftIoThread = new RaftIoThread(raftClient, threadNamePrefix)
+  val client: KafkaRaftClient[T] = buildRaftClient()
+  private val raftIoThread = new RaftIoThread(client, threadNamePrefix)
 
-  def kafkaRaftClient: KafkaRaftClient[T] = raftClient
+  def kafkaRaftClient: KafkaRaftClient[T] = client
 
   def startup(): Unit = {
     // Update the voter endpoints (if valid) with what's in RaftConfig
@@ -151,7 +155,7 @@ class KafkaRaftManager[T](
 
   def shutdown(): Unit = {
     raftIoThread.shutdown()
-    raftClient.close()
+    client.close()
     scheduler.shutdown()
     netChannel.close()
     metadataLog.close()
@@ -160,7 +164,7 @@ class KafkaRaftManager[T](
   override def register(
     listener: RaftClient.Listener[T]
   ): Unit = {
-    raftClient.register(listener)
+    client.register(listener)
   }
 
   override def scheduleAtomicAppend(
@@ -183,9 +187,9 @@ class KafkaRaftManager[T](
     isAtomic: Boolean
   ): Option[Long] = {
     val offset = if (isAtomic) {
-      raftClient.scheduleAtomicAppend(epoch, records.asJava)
+      client.scheduleAtomicAppend(epoch, records.asJava)
     } else {
-      raftClient.scheduleAppend(epoch, records.asJava)
+      client.scheduleAppend(epoch, records.asJava)
     }
 
     Option(offset).map(Long.unbox)
@@ -202,7 +206,7 @@ class KafkaRaftManager[T](
       createdTimeMs
     )
 
-    raftClient.handle(inboundRequest)
+    client.handle(inboundRequest)
 
     inboundRequest.completion.thenApply { response =>
       response.data
@@ -306,4 +310,7 @@ class KafkaRaftManager[T](
     )
   }
 
+  override def leaderAndEpoch: LeaderAndEpoch = {
+    client.leaderAndEpoch
+  }
 }

core/src/main/scala/kafka/server/BrokerServer.scala

@@ -18,16 +18,17 @@
 package kafka.server
 
 import java.util
-import java.util.concurrent.{CompletableFuture, TimeUnit, TimeoutException}
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.locks.ReentrantLock
+import java.util.concurrent.{CompletableFuture, TimeUnit, TimeoutException}
 
 import kafka.cluster.Broker.ServerInfo
 import kafka.coordinator.group.GroupCoordinator
 import kafka.coordinator.transaction.{ProducerIdGenerator, TransactionCoordinator}
 import kafka.log.LogManager
 import kafka.metrics.KafkaYammerMetrics
 import kafka.network.SocketServer
+import kafka.raft.RaftManager
 import kafka.security.CredentialProvider
 import kafka.server.metadata.{BrokerMetadataListener, CachedConfigRepository, ClientQuotaCache, ClientQuotaMetadataManager, RaftMetadataCache}
 import kafka.utils.{CoreUtils, KafkaScheduler}
@@ -41,8 +42,7 @@ import org.apache.kafka.common.security.scram.internals.ScramMechanism
 import org.apache.kafka.common.security.token.delegation.internals.DelegationTokenCache
 import org.apache.kafka.common.utils.{AppInfoParser, LogContext, Time}
 import org.apache.kafka.common.{ClusterResource, Endpoint, KafkaException}
-import org.apache.kafka.metadata.{BrokerState, VersionRange}
-import org.apache.kafka.metalog.MetaLogManager
+import org.apache.kafka.metadata.{ApiMessageAndVersion, BrokerState, VersionRange}
 import org.apache.kafka.raft.RaftConfig
 import org.apache.kafka.raft.RaftConfig.AddressSpec
 import org.apache.kafka.server.authorizer.Authorizer
@@ -54,16 +54,16 @@ import scala.jdk.CollectionConverters._
  * A Kafka broker that runs in KRaft (Kafka Raft) mode.
  */
 class BrokerServer(
-                    val config: KafkaConfig,
-                    val metaProps: MetaProperties,
-                    val metaLogManager: MetaLogManager,
-                    val time: Time,
-                    val metrics: Metrics,
-                    val threadNamePrefix: Option[String],
-                    val initialOfflineDirs: Seq[String],
-                    val controllerQuorumVotersFuture: CompletableFuture[util.Map[Integer, AddressSpec]],
-                    val supportedFeatures: util.Map[String, VersionRange]
-                  ) extends KafkaBroker {
+  val config: KafkaConfig,
+  val metaProps: MetaProperties,
+  val raftManager: RaftManager[ApiMessageAndVersion],
+  val time: Time,
+  val metrics: Metrics,
+  val threadNamePrefix: Option[String],
+  val initialOfflineDirs: Seq[String],
+  val controllerQuorumVotersFuture: CompletableFuture[util.Map[Integer, AddressSpec]],
+  val supportedFeatures: util.Map[String, VersionRange]
+) extends KafkaBroker {
 
   import kafka.server.Server._
 
@@ -179,7 +179,7 @@ class BrokerServer(
       credentialProvider = new CredentialProvider(ScramMechanism.mechanismNames, tokenCache)
 
       val controllerNodes = RaftConfig.voterConnectionsToNodes(controllerQuorumVotersFuture.get()).asScala
-      val controllerNodeProvider = RaftControllerNodeProvider(metaLogManager, config, controllerNodes)
+      val controllerNodeProvider = RaftControllerNodeProvider(raftManager, config, controllerNodes)
 
       clientToControllerChannelManager = BrokerToControllerChannelManager(
         controllerNodeProvider,
@@ -282,7 +282,7 @@ class BrokerServer(
         metaProps.clusterId, networkListeners, supportedFeatures)
 
       // Register a listener with the Raft layer to receive metadata event notifications
-      metaLogManager.register(brokerMetadataListener)
+      raftManager.register(brokerMetadataListener)
 
       val endpoints = new util.ArrayList[Endpoint](networkListeners.size())
       var interBrokerListener: Endpoint = null

core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala

@@ -21,6 +21,7 @@ import java.util.concurrent.LinkedBlockingDeque
 import java.util.concurrent.atomic.AtomicReference
 
 import kafka.common.{InterBrokerSendThread, RequestAndCompletionHandler}
+import kafka.raft.RaftManager
 import kafka.utils.Logging
 import org.apache.kafka.clients._
 import org.apache.kafka.common.Node
@@ -31,9 +32,10 @@ import org.apache.kafka.common.requests.AbstractRequest
 import org.apache.kafka.common.security.JaasContext
 import org.apache.kafka.common.security.auth.SecurityProtocol
 import org.apache.kafka.common.utils.{LogContext, Time}
-import org.apache.kafka.metalog.MetaLogManager
+import org.apache.kafka.metadata.ApiMessageAndVersion
 
 import scala.collection.Seq
+import scala.compat.java8.OptionConverters._
 import scala.jdk.CollectionConverters._
 
 trait ControllerNodeProvider {
@@ -77,15 +79,14 @@ class MetadataCacheControllerNodeProvider(
 }
 
 object RaftControllerNodeProvider {
-  def apply(metaLogManager: MetaLogManager,
+  def apply(raftManager: RaftManager[ApiMessageAndVersion],
             config: KafkaConfig,
             controllerQuorumVoterNodes: Seq[Node]): RaftControllerNodeProvider = {
-
     val controllerListenerName = new ListenerName(config.controllerListenerNames.head)
     val controllerSecurityProtocol = config.listenerSecurityProtocolMap.getOrElse(controllerListenerName, SecurityProtocol.forName(controllerListenerName.value()))
     val controllerSaslMechanism = config.saslMechanismControllerProtocol
     new RaftControllerNodeProvider(
-      metaLogManager,
+      raftManager,
       controllerQuorumVoterNodes,
       controllerListenerName,
       controllerSecurityProtocol,
@@ -98,7 +99,7 @@ object RaftControllerNodeProvider {
  * Finds the controller node by checking the metadata log manager.
  * This provider is used when we are using a Raft-based metadata quorum.
  */
-class RaftControllerNodeProvider(val metaLogManager: MetaLogManager,
+class RaftControllerNodeProvider(val raftManager: RaftManager[ApiMessageAndVersion],
                                  controllerQuorumVoterNodes: Seq[Node],
                                  val listenerName: ListenerName,
                                  val securityProtocol: SecurityProtocol,
@@ -107,14 +108,7 @@ class RaftControllerNodeProvider(val metaLogManager: MetaLogManager,
   val idToNode = controllerQuorumVoterNodes.map(node => node.id() -> node).toMap
 
   override def get(): Option[Node] = {
-    val leader = metaLogManager.leader()
-    if (leader == null) {
-      None
-    } else if (leader.nodeId() < 0) {
-      None
-    } else {
-      idToNode.get(leader.nodeId())
-    }
+    raftManager.leaderAndEpoch.leaderId.asScala.map(idToNode)
   }
 }
 

core/src/main/scala/kafka/server/ControllerServer.scala

@@ -17,9 +17,9 @@
 
 package kafka.server
 
-import java.util.concurrent.{CompletableFuture, TimeUnit}
 import java.util
 import java.util.concurrent.locks.ReentrantLock
+import java.util.concurrent.{CompletableFuture, TimeUnit}
 
 import kafka.cluster.Broker.ServerInfo
 import kafka.log.LogConfig
@@ -38,7 +38,6 @@ import org.apache.kafka.common.utils.{LogContext, Time}
 import org.apache.kafka.common.{ClusterResource, Endpoint}
 import org.apache.kafka.controller.{Controller, QuorumController, QuorumControllerMetrics}
 import org.apache.kafka.metadata.{ApiMessageAndVersion, VersionRange}
-import org.apache.kafka.metalog.MetaLogManager
 import org.apache.kafka.raft.RaftConfig
 import org.apache.kafka.raft.RaftConfig.AddressSpec
 import org.apache.kafka.server.authorizer.Authorizer
@@ -49,15 +48,14 @@ import scala.jdk.CollectionConverters._
  * A Kafka controller that runs in KRaft (Kafka Raft) mode.
  */
 class ControllerServer(
-                        val metaProperties: MetaProperties,
-                        val config: KafkaConfig,
-                        val metaLogManager: MetaLogManager,
-                        val raftManager: RaftManager[ApiMessageAndVersion],
-                        val time: Time,
-                        val metrics: Metrics,
-                        val threadNamePrefix: Option[String],
-                        val controllerQuorumVotersFuture: CompletableFuture[util.Map[Integer, AddressSpec]]
-                      ) extends Logging with KafkaMetricsGroup {
+  val metaProperties: MetaProperties,
+  val config: KafkaConfig,
+  val raftManager: RaftManager[ApiMessageAndVersion],
+  val time: Time,
+  val metrics: Metrics,
+  val threadNamePrefix: Option[String],
+  val controllerQuorumVotersFuture: CompletableFuture[util.Map[Integer, AddressSpec]]
+) extends Logging with KafkaMetricsGroup {
   import kafka.server.Server._
 
   val lock = new ReentrantLock()
@@ -148,7 +146,7 @@ class ControllerServer(
         setTime(time).
         setThreadNamePrefix(threadNamePrefixAsString).
         setConfigDefs(configDefs).
-        setLogManager(metaLogManager).
+        setRaftClient(raftManager.client).
         setDefaultReplicationFactor(config.defaultReplicationFactor.toShort).
         setDefaultNumPartitions(config.numPartitions.intValue()).
         setSessionTimeoutNs(TimeUnit.NANOSECONDS.convert(config.brokerSessionTimeoutMs.longValue(),

core/src/main/scala/kafka/server/KafkaRaftServer.scala

@@ -27,9 +27,8 @@ import kafka.server.KafkaRaftServer.{BrokerRole, ControllerRole}
 import kafka.utils.{CoreUtils, Logging, Mx4jLoader, VerifiableProperties}
 import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.utils.{AppInfoParser, Time}
-import org.apache.kafka.metadata.ApiMessageAndVersion
+import org.apache.kafka.metadata.{ApiMessageAndVersion, MetadataRecordSerde}
 import org.apache.kafka.raft.RaftConfig
-import org.apache.kafka.raft.metadata.{MetaLogRaftShim, MetadataRecordSerde}
 
 import scala.collection.Seq
 
@@ -56,7 +55,7 @@ class KafkaRaftServer(
   private val metrics = Server.initializeMetrics(
     config,
     time,
-    metaProps.clusterId.toString
+    metaProps.clusterId
   )
 
   private val controllerQuorumVotersFuture = CompletableFuture.completedFuture(
@@ -73,13 +72,11 @@ class KafkaRaftServer(
     controllerQuorumVotersFuture
   )
 
-  private val metaLogShim = new MetaLogRaftShim(raftManager.kafkaRaftClient, config.nodeId)
-
   private val broker: Option[BrokerServer] = if (config.processRoles.contains(BrokerRole)) {
     Some(new BrokerServer(
       config,
       metaProps,
-      metaLogShim,
+      raftManager,
       time,
       metrics,
       threadNamePrefix,
@@ -95,7 +92,6 @@ class KafkaRaftServer(
     Some(new ControllerServer(
       metaProps,
       config,
-      metaLogShim,
       raftManager,
       time,
       metrics,