You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by cm...@apache.org on 2023/01/05 20:11:10 UTC
[kafka] 04/04: KAFKA-14458: Introduce RPC support during ZK migration #13028
This is an automated email from the ASF dual-hosted git repository.
cmccabe pushed a commit to branch 3.4
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 347042ccca23ee325ee0bc66011abad31756f6a1
Author: Akhilesh Chaganti <ak...@users.noreply.github.com>
AuthorDate: Wed Jan 4 16:54:58 2023 -0800
KAFKA-14458: Introduce RPC support during ZK migration #13028
Add infrastructure for sending UpdateMetadataRequest and LeaderAndIsr RPCs during the migration
process from ZK to KRaft. The new classes use ControllerChannelManager to send the RPCs. The
information to send comes from MetadataDelta and MetadataImage.
Reviewers: David Arthur <mu...@gmail.com>, Colin P. McCabe <cm...@apache.org>
---
core/src/main/scala/kafka/cluster/Broker.scala | 19 +-
.../controller/ControllerChannelContext.scala | 42 ++++
.../controller/ControllerChannelManager.scala | 243 +++++++++++++--------
.../scala/kafka/controller/ControllerContext.scala | 26 ++-
.../scala/kafka/controller/KafkaController.scala | 12 +-
.../MigrationControllerChannelContext.scala | 115 ++++++++++
.../kafka/migration/MigrationPropagator.scala | 204 +++++++++++++++++
.../kafka/migration/MigrationPropagatorBatch.scala | 85 +++++++
.../controller/ControllerChannelManagerTest.scala | 20 +-
.../kafka/server/BrokerEpochIntegrationTest.scala | 11 +-
.../unit/kafka/server/LeaderElectionTest.scala | 11 +-
.../unit/kafka/server/ServerShutdownTest.scala | 10 +-
.../java/org/apache/kafka/image/ClusterDelta.java | 12 +
.../java/org/apache/kafka/image/ClusterImage.java | 15 ++
.../java/org/apache/kafka/image/TopicsDelta.java | 8 +
.../java/org/apache/kafka/image/TopicsImage.java | 10 +
.../kafka/metadata/migration/LegacyPropagator.java | 37 ++++
17 files changed, 763 insertions(+), 117 deletions(-)
diff --git a/core/src/main/scala/kafka/cluster/Broker.scala b/core/src/main/scala/kafka/cluster/Broker.scala
index 9b1d741835c..ede63cd3c0a 100755
--- a/core/src/main/scala/kafka/cluster/Broker.scala
+++ b/core/src/main/scala/kafka/cluster/Broker.scala
@@ -18,7 +18,6 @@
package kafka.cluster
import java.util
-
import kafka.common.BrokerEndPointNotAvailableException
import kafka.server.KafkaConfig
import org.apache.kafka.common.feature.{Features, SupportedVersionRange}
@@ -26,9 +25,11 @@ import org.apache.kafka.common.feature.Features._
import org.apache.kafka.common.{ClusterResource, Endpoint, Node}
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.security.auth.SecurityProtocol
+import org.apache.kafka.metadata.{BrokerRegistration, VersionRange}
import org.apache.kafka.server.authorizer.AuthorizerServerInfo
import scala.collection.Seq
+import scala.compat.java8.OptionConverters._
import scala.jdk.CollectionConverters._
object Broker {
@@ -41,6 +42,22 @@ object Broker {
def apply(id: Int, endPoints: Seq[EndPoint], rack: Option[String]): Broker = {
new Broker(id, endPoints, rack, emptySupportedFeatures)
}
+
+ private def supportedFeatures(features: java.util.Map[String, VersionRange]): java.util
+ .Map[String, SupportedVersionRange] = {
+ features.asScala.map { case (name, range) =>
+ name -> new SupportedVersionRange(range.min(), range.max())
+ }.asJava
+ }
+
+ def fromBrokerRegistration(registration: BrokerRegistration): Broker = {
+ new Broker(
+ registration.id(),
+ registration.listeners().values().asScala.map(EndPoint.fromJava).toSeq,
+ registration.rack().asScala,
+ Features.supportedFeatures(supportedFeatures(registration.supportedFeatures()))
+ )
+ }
}
/**
diff --git a/core/src/main/scala/kafka/controller/ControllerChannelContext.scala b/core/src/main/scala/kafka/controller/ControllerChannelContext.scala
new file mode 100644
index 00000000000..0ab43256290
--- /dev/null
+++ b/core/src/main/scala/kafka/controller/ControllerChannelContext.scala
@@ -0,0 +1,42 @@
+/*
+ * 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 kafka.controller
+
+import kafka.cluster.Broker
+import org.apache.kafka.common.{TopicPartition, Uuid}
+
+trait ControllerChannelContext {
+ def isTopicDeletionInProgress(topicName: String): Boolean
+
+ def topicIds: collection.Map[String, Uuid]
+
+ def liveBrokerIdAndEpochs: collection.Map[Int, Long]
+
+ def liveOrShuttingDownBrokers: collection.Set[Broker]
+
+ def isTopicQueuedUpForDeletion(topic: String): Boolean
+
+ def isReplicaOnline(brokerId: Int, partition: TopicPartition): Boolean
+
+ def partitionReplicaAssignment(partition: TopicPartition): collection.Seq[Int]
+
+ def leaderEpoch(topicPartition: TopicPartition): Int
+
+ def liveOrShuttingDownBrokerIds: collection.Set[Int]
+
+ def partitionLeadershipInfo(topicPartition: TopicPartition): Option[LeaderIsrAndControllerEpoch]
+}
diff --git a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
index 145eca453a1..966262e6df3 100755
--- a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
+++ b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
@@ -16,40 +16,39 @@
*/
package kafka.controller
-import java.net.SocketTimeoutException
-import java.util.concurrent.{BlockingQueue, LinkedBlockingQueue, TimeUnit}
-
import com.yammer.metrics.core.{Gauge, Timer}
import kafka.api._
import kafka.cluster.Broker
import kafka.metrics.KafkaMetricsGroup
import kafka.server.KafkaConfig
-import kafka.utils._
import kafka.utils.Implicits._
+import kafka.utils._
import org.apache.kafka.clients._
+import org.apache.kafka.common._
import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState
import org.apache.kafka.common.message.StopReplicaRequestData.{StopReplicaPartitionState, StopReplicaTopicState}
+import org.apache.kafka.common.message.UpdateMetadataRequestData.{UpdateMetadataBroker, UpdateMetadataEndpoint, UpdateMetadataPartitionState}
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.network._
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
-import org.apache.kafka.common.message.UpdateMetadataRequestData.{UpdateMetadataBroker, UpdateMetadataEndpoint, UpdateMetadataPartitionState}
import org.apache.kafka.common.requests._
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.common.{KafkaException, Node, Reconfigurable, TopicPartition, Uuid}
+import org.apache.kafka.server.common.MetadataVersion
import org.apache.kafka.server.common.MetadataVersion._
-import scala.jdk.CollectionConverters._
-import scala.collection.mutable.HashMap
+import java.net.SocketTimeoutException
+import java.util.concurrent.{BlockingQueue, LinkedBlockingQueue, TimeUnit}
import scala.collection.{Seq, Set, mutable}
+import scala.jdk.CollectionConverters._
object ControllerChannelManager {
val QueueSizeMetricName = "QueueSize"
val RequestRateAndQueueTimeMetricName = "RequestRateAndQueueTimeMs"
}
-class ControllerChannelManager(controllerContext: ControllerContext,
+class ControllerChannelManager(controllerEpoch: () => Int,
config: KafkaConfig,
time: Time,
metrics: Metrics,
@@ -57,7 +56,7 @@ class ControllerChannelManager(controllerContext: ControllerContext,
threadNamePrefix: Option[String] = None) extends Logging with KafkaMetricsGroup {
import ControllerChannelManager._
- protected val brokerStateInfo = new HashMap[Int, ControllerBrokerStateInfo]
+ protected val brokerStateInfo = new mutable.HashMap[Int, ControllerBrokerStateInfo]
private val brokerLock = new Object
this.logIdent = "[Channel manager on controller " + config.brokerId + "]: "
@@ -67,15 +66,15 @@ class ControllerChannelManager(controllerContext: ControllerContext,
}
)
- def startup() = {
- controllerContext.liveOrShuttingDownBrokers.foreach(addNewBroker)
+ def startup(initialBrokers: Set[Broker]):Unit = {
+ initialBrokers.foreach(addNewBroker)
brokerLock synchronized {
brokerStateInfo.foreach(brokerState => startRequestSendThread(brokerState._1))
}
}
- def shutdown() = {
+ def shutdown():Unit = {
brokerLock synchronized {
brokerStateInfo.values.toList.foreach(removeExistingBroker)
}
@@ -173,7 +172,7 @@ class ControllerChannelManager(controllerContext: ControllerContext,
RequestRateAndQueueTimeMetricName, TimeUnit.MILLISECONDS, TimeUnit.SECONDS, brokerMetricTags(broker.id)
)
- val requestThread = new RequestSendThread(config.brokerId, controllerContext, messageQueue, networkClient,
+ val requestThread = new RequestSendThread(config.brokerId, controllerEpoch, messageQueue, networkClient,
brokerNode, config, time, requestRateAndQueueTimeMetrics, stateChangeLogger, threadName)
requestThread.setDaemon(false)
@@ -214,7 +213,7 @@ case class QueueItem(apiKey: ApiKeys, request: AbstractControlRequest.Builder[_
callback: AbstractResponse => Unit, enqueueTimeMs: Long)
class RequestSendThread(val controllerId: Int,
- val controllerContext: ControllerContext,
+ controllerEpoch: () => Int,
val queue: BlockingQueue[QueueItem],
val networkClient: NetworkClient,
val brokerNode: Node,
@@ -255,7 +254,8 @@ class RequestSendThread(val controllerId: Int,
}
} catch {
case e: Throwable => // if the send was not successful, reconnect to broker and resend the message
- warn(s"Controller $controllerId epoch ${controllerContext.epoch} fails to send request $requestBuilder " +
+ warn(s"Controller $controllerId epoch ${controllerEpoch()} fails to send request " +
+ s"$requestBuilder " +
s"to broker $brokerNode. Reconnecting to broker.", e)
networkClient.close(brokerNode.idString)
isSendSuccessful = false
@@ -270,7 +270,7 @@ class RequestSendThread(val controllerId: Int,
val response = clientResponse.responseBody
- stateChangeLogger.withControllerEpoch(controllerContext.epoch).trace(s"Received response " +
+ stateChangeLogger.withControllerEpoch(controllerEpoch()).trace(s"Received response " +
s"$response for request $api with correlation id " +
s"${requestHeader.correlationId} sent to broker $brokerNode")
@@ -313,35 +313,65 @@ class RequestSendThread(val controllerId: Int,
}
}
-class ControllerBrokerRequestBatch(config: KafkaConfig,
- controllerChannelManager: ControllerChannelManager,
- controllerEventManager: ControllerEventManager,
- controllerContext: ControllerContext,
- stateChangeLogger: StateChangeLogger)
- extends AbstractControllerBrokerRequestBatch(config, controllerContext, stateChangeLogger) {
+class ControllerBrokerRequestBatch(
+ config: KafkaConfig,
+ controllerChannelManager: ControllerChannelManager,
+ controllerEventManager: ControllerEventManager,
+ controllerContext: ControllerContext,
+ stateChangeLogger: StateChangeLogger
+) extends AbstractControllerBrokerRequestBatch(
+ config,
+ () => controllerContext,
+ () => config.interBrokerProtocolVersion,
+ stateChangeLogger
+) {
def sendEvent(event: ControllerEvent): Unit = {
controllerEventManager.put(event)
}
-
def sendRequest(brokerId: Int,
request: AbstractControlRequest.Builder[_ <: AbstractControlRequest],
callback: AbstractResponse => Unit = null): Unit = {
controllerChannelManager.sendRequest(brokerId, request, callback)
}
+ override def handleLeaderAndIsrResponse(response: LeaderAndIsrResponse, broker: Int): Unit = {
+ sendEvent(LeaderAndIsrResponseReceived(response, broker))
+ }
+
+ override def handleUpdateMetadataResponse(response: UpdateMetadataResponse, broker: Int): Unit = {
+ sendEvent(UpdateMetadataResponseReceived(response, broker))
+ }
+
+ override def handleStopReplicaResponse(stopReplicaResponse: StopReplicaResponse, brokerId: Int,
+ partitionErrorsForDeletingTopics: Map[TopicPartition, Errors]): Unit = {
+ if (partitionErrorsForDeletingTopics.nonEmpty)
+ sendEvent(TopicDeletionStopReplicaResponseReceived(brokerId, stopReplicaResponse.error, partitionErrorsForDeletingTopics))
+ }
}
+/**
+ * Structure to send RPCs from controller to broker to inform about the metadata and leadership
+ * changes in the system.
+ * @param config Kafka config present in the controller.
+ * @param metadataProvider Provider to provide the relevant metadata to build the state needed to
+ * send RPCs
+ * @param metadataVersionProvider Provider to provide the metadata version used by the controller.
+ * @param stateChangeLogger logger to log the various events while sending requests and receving
+ * responses from the brokers
+ * @param kraftController whether the controller is KRaft controller
+ */
abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
- controllerContext: ControllerContext,
- stateChangeLogger: StateChangeLogger) extends Logging {
+ metadataProvider: () => ControllerChannelContext,
+ metadataVersionProvider: () => MetadataVersion,
+ stateChangeLogger: StateChangeLogger,
+ kraftController: Boolean = false) extends Logging {
val controllerId: Int = config.brokerId
val leaderAndIsrRequestMap = mutable.Map.empty[Int, mutable.Map[TopicPartition, LeaderAndIsrPartitionState]]
val stopReplicaRequestMap = mutable.Map.empty[Int, mutable.Map[TopicPartition, StopReplicaPartitionState]]
val updateMetadataRequestBrokerSet = mutable.Set.empty[Int]
val updateMetadataRequestPartitionInfoMap = mutable.Map.empty[TopicPartition, UpdateMetadataPartitionState]
-
- def sendEvent(event: ControllerEvent): Unit
+ private var metadataInstance: ControllerChannelContext = _
def sendRequest(brokerId: Int,
request: AbstractControlRequest.Builder[_ <: AbstractControlRequest],
@@ -359,6 +389,7 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
throw new IllegalStateException("Controller to broker state change requests batch is not empty while creating a " +
s"new one. Some UpdateMetadata state changes to brokers $updateMetadataRequestBrokerSet with partition info " +
s"$updateMetadataRequestPartitionInfoMap might be lost ")
+ metadataInstance = metadataProvider()
}
def clear(): Unit = {
@@ -366,6 +397,7 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
stopReplicaRequestMap.clear()
updateMetadataRequestBrokerSet.clear()
updateMetadataRequestPartitionInfoMap.clear()
+ metadataInstance = null
}
def addLeaderAndIsrRequestForBrokers(brokerIds: Seq[Int],
@@ -391,14 +423,14 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
.setRemovingReplicas(replicaAssignment.removingReplicas.map(Integer.valueOf).asJava)
.setIsNew(isNew || alreadyNew)
- if (config.interBrokerProtocolVersion.isAtLeast(IBP_3_2_IV0)) {
+ if (metadataVersionProvider.apply().isAtLeast(IBP_3_2_IV0)) {
partitionState.setLeaderRecoveryState(leaderAndIsr.leaderRecoveryState.value)
}
result.put(topicPartition, partitionState)
}
- addUpdateMetadataRequestForBrokers(controllerContext.liveOrShuttingDownBrokerIds.toSeq, Set(topicPartition))
+ addUpdateMetadataRequestForBrokers(metadataInstance.liveOrShuttingDownBrokerIds.toSeq, Set(topicPartition))
}
def addStopReplicaRequestForBrokers(brokerIds: Seq[Int],
@@ -406,13 +438,7 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
deletePartition: Boolean): Unit = {
// A sentinel (-2) is used as an epoch if the topic is queued for deletion. It overrides
// any existing epoch.
- val leaderEpoch = if (controllerContext.isTopicQueuedUpForDeletion(topicPartition.topic)) {
- LeaderAndIsr.EpochDuringDelete
- } else {
- controllerContext.partitionLeadershipInfo(topicPartition)
- .map(_.leaderAndIsr.leaderEpoch)
- .getOrElse(LeaderAndIsr.NoEpoch)
- }
+ val leaderEpoch = metadataInstance.leaderEpoch(topicPartition)
brokerIds.filter(_ >= 0).foreach { brokerId =>
val result = stopReplicaRequestMap.getOrElseUpdate(brokerId, mutable.Map.empty)
@@ -427,51 +453,66 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
/** Send UpdateMetadataRequest to the given brokers for the given partitions and partitions that are being deleted */
def addUpdateMetadataRequestForBrokers(brokerIds: Seq[Int],
partitions: collection.Set[TopicPartition]): Unit = {
-
- def updateMetadataRequestPartitionInfo(partition: TopicPartition, beingDeleted: Boolean): Unit = {
- controllerContext.partitionLeadershipInfo(partition) match {
+ updateMetadataRequestBrokerSet ++= brokerIds.filter(_ >= 0)
+ partitions.foreach { partition =>
+ val beingDeleted = metadataInstance.isTopicQueuedUpForDeletion(partition.topic())
+ metadataInstance.partitionLeadershipInfo(partition) match {
case Some(LeaderIsrAndControllerEpoch(leaderAndIsr, controllerEpoch)) =>
- val replicas = controllerContext.partitionReplicaAssignment(partition)
- val offlineReplicas = replicas.filterNot(controllerContext.isReplicaOnline(_, partition))
+ val replicas = metadataInstance.partitionReplicaAssignment(partition)
+ val offlineReplicas = replicas.filter(!metadataInstance.isReplicaOnline(_, partition))
val updatedLeaderAndIsr =
if (beingDeleted) LeaderAndIsr.duringDelete(leaderAndIsr.isr)
else leaderAndIsr
-
- val partitionStateInfo = new UpdateMetadataPartitionState()
- .setTopicName(partition.topic)
- .setPartitionIndex(partition.partition)
- .setControllerEpoch(controllerEpoch)
- .setLeader(updatedLeaderAndIsr.leader)
- .setLeaderEpoch(updatedLeaderAndIsr.leaderEpoch)
- .setIsr(updatedLeaderAndIsr.isr.map(Integer.valueOf).asJava)
- .setZkVersion(updatedLeaderAndIsr.partitionEpoch)
- .setReplicas(replicas.map(Integer.valueOf).asJava)
- .setOfflineReplicas(offlineReplicas.map(Integer.valueOf).asJava)
- updateMetadataRequestPartitionInfoMap.put(partition, partitionStateInfo)
-
+ addUpdateMetadataRequestForBrokers(brokerIds, controllerEpoch, partition,
+ updatedLeaderAndIsr.leader, updatedLeaderAndIsr.leaderEpoch, updatedLeaderAndIsr.partitionEpoch,
+ updatedLeaderAndIsr.isr, replicas, offlineReplicas)
case None =>
info(s"Leader not yet assigned for partition $partition. Skip sending UpdateMetadataRequest.")
}
}
+ }
+
+ def addUpdateMetadataRequestForBrokers(brokerIds: Seq[Int]): Unit = {
+ updateMetadataRequestBrokerSet ++= brokerIds.filter(_ >= 0)
+ }
+ def addUpdateMetadataRequestForBrokers(brokerIds: Seq[Int],
+ controllerEpoch: Int,
+ partition: TopicPartition,
+ leader: Int,
+ leaderEpoch: Int,
+ partitionEpoch: Int,
+ isrs: List[Int],
+ replicas: Seq[Int],
+ offlineReplicas: Seq[Int]): Unit = {
updateMetadataRequestBrokerSet ++= brokerIds.filter(_ >= 0)
- partitions.foreach(partition => updateMetadataRequestPartitionInfo(partition,
- beingDeleted = controllerContext.topicsToBeDeleted.contains(partition.topic)))
+ val partitionStateInfo = new UpdateMetadataPartitionState()
+ .setTopicName(partition.topic)
+ .setPartitionIndex(partition.partition)
+ .setControllerEpoch(controllerEpoch)
+ .setLeader(leader)
+ .setLeaderEpoch(leaderEpoch)
+ .setIsr(isrs.map(Integer.valueOf).asJava)
+ .setZkVersion(partitionEpoch)
+ .setReplicas(replicas.map(Integer.valueOf).asJava)
+ .setOfflineReplicas(offlineReplicas.map(Integer.valueOf).asJava)
+ updateMetadataRequestPartitionInfoMap.put(partition, partitionStateInfo)
}
private def sendLeaderAndIsrRequest(controllerEpoch: Int, stateChangeLog: StateChangeLogger): Unit = {
+ val metadataVersion = metadataVersionProvider.apply()
val leaderAndIsrRequestVersion: Short =
- if (config.interBrokerProtocolVersion.isAtLeast(IBP_3_4_IV0)) 7
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_3_2_IV0)) 6
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_2_8_IV1)) 5
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_2_4_IV1)) 4
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_2_4_IV0)) 3
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_2_2_IV0)) 2
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_1_0_IV0)) 1
+ if (metadataVersion.isAtLeast(IBP_3_4_IV0)) 7
+ else if (metadataVersion.isAtLeast(IBP_3_2_IV0)) 6
+ else if (metadataVersion.isAtLeast(IBP_2_8_IV1)) 5
+ else if (metadataVersion.isAtLeast(IBP_2_4_IV1)) 4
+ else if (metadataVersion.isAtLeast(IBP_2_4_IV0)) 3
+ else if (metadataVersion.isAtLeast(IBP_2_2_IV0)) 2
+ else if (metadataVersion.isAtLeast(IBP_1_0_IV0)) 1
else 0
leaderAndIsrRequestMap.forKeyValue { (broker, leaderAndIsrPartitionStates) =>
- if (controllerContext.liveOrShuttingDownBrokerIds.contains(broker)) {
+ if (metadataInstance.liveOrShuttingDownBrokerIds.contains(broker)) {
val leaderIds = mutable.Set.empty[Int]
var numBecomeLeaders = 0
leaderAndIsrPartitionStates.forKeyValue { (topicPartition, state) =>
@@ -487,43 +528,46 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
}
stateChangeLog.info(s"Sending LeaderAndIsr request to broker $broker with $numBecomeLeaders become-leader " +
s"and ${leaderAndIsrPartitionStates.size - numBecomeLeaders} become-follower partitions")
- val leaders = controllerContext.liveOrShuttingDownBrokers.filter(b => leaderIds.contains(b.id)).map {
+ val leaders = metadataInstance.liveOrShuttingDownBrokers.filter(b => leaderIds.contains(b.id)).map {
_.node(config.interBrokerListenerName)
}
- val brokerEpoch = controllerContext.liveBrokerIdAndEpochs(broker)
+ val brokerEpoch = metadataInstance.liveBrokerIdAndEpochs(broker)
val topicIds = leaderAndIsrPartitionStates.keys
.map(_.topic)
.toSet[String]
- .map(topic => (topic, controllerContext.topicIds.getOrElse(topic, Uuid.ZERO_UUID)))
+ .map(topic => (topic, metadataInstance.topicIds.getOrElse(topic, Uuid.ZERO_UUID)))
.toMap
val leaderAndIsrRequestBuilder = new LeaderAndIsrRequest.Builder(leaderAndIsrRequestVersion, controllerId,
- controllerEpoch, brokerEpoch, leaderAndIsrPartitionStates.values.toBuffer.asJava, topicIds.asJava, leaders.asJava)
+ controllerEpoch, brokerEpoch, leaderAndIsrPartitionStates.values.toBuffer.asJava, topicIds.asJava, leaders.asJava, kraftController)
sendRequest(broker, leaderAndIsrRequestBuilder, (r: AbstractResponse) => {
val leaderAndIsrResponse = r.asInstanceOf[LeaderAndIsrResponse]
- sendEvent(LeaderAndIsrResponseReceived(leaderAndIsrResponse, broker))
+ handleLeaderAndIsrResponse(leaderAndIsrResponse, broker)
})
}
}
leaderAndIsrRequestMap.clear()
}
+ def handleLeaderAndIsrResponse(response: LeaderAndIsrResponse, broker: Int): Unit
+
private def sendUpdateMetadataRequests(controllerEpoch: Int, stateChangeLog: StateChangeLogger): Unit = {
stateChangeLog.info(s"Sending UpdateMetadata request to brokers $updateMetadataRequestBrokerSet " +
s"for ${updateMetadataRequestPartitionInfoMap.size} partitions")
val partitionStates = updateMetadataRequestPartitionInfoMap.values.toBuffer
+ val metadataVersion = metadataVersionProvider.apply()
val updateMetadataRequestVersion: Short =
- if (config.interBrokerProtocolVersion.isAtLeast(IBP_3_4_IV0)) 8
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_2_8_IV1)) 7
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_2_4_IV1)) 6
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_2_2_IV0)) 5
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_1_0_IV0)) 4
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_0_10_2_IV0)) 3
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_0_10_0_IV1)) 2
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_0_9_0)) 1
+ if (metadataVersion.isAtLeast(IBP_3_4_IV0)) 8
+ else if (metadataVersion.isAtLeast(IBP_2_8_IV1)) 7
+ else if (metadataVersion.isAtLeast(IBP_2_4_IV1)) 6
+ else if (metadataVersion.isAtLeast(IBP_2_2_IV0)) 5
+ else if (metadataVersion.isAtLeast(IBP_1_0_IV0)) 4
+ else if (metadataVersion.isAtLeast(IBP_0_10_2_IV0)) 3
+ else if (metadataVersion.isAtLeast(IBP_0_10_0_IV1)) 2
+ else if (metadataVersion.isAtLeast(IBP_0_9_0)) 1
else 0
- val liveBrokers = controllerContext.liveOrShuttingDownBrokers.iterator.map { broker =>
+ val liveBrokers = metadataInstance.liveOrShuttingDownBrokers.iterator.map { broker =>
val endpoints = if (updateMetadataRequestVersion == 0) {
// Version 0 of UpdateMetadataRequest only supports PLAINTEXT
val securityProtocol = SecurityProtocol.PLAINTEXT
@@ -549,17 +593,17 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
.setRack(broker.rack.orNull)
}.toBuffer
- updateMetadataRequestBrokerSet.intersect(controllerContext.liveOrShuttingDownBrokerIds).foreach { broker =>
- val brokerEpoch = controllerContext.liveBrokerIdAndEpochs(broker)
+ updateMetadataRequestBrokerSet.intersect(metadataInstance.liveOrShuttingDownBrokerIds).foreach { broker =>
+ val brokerEpoch = metadataInstance.liveBrokerIdAndEpochs(broker)
val topicIds = partitionStates.map(_.topicName())
.distinct
- .filter(controllerContext.topicIds.contains)
- .map(topic => (topic, controllerContext.topicIds(topic))).toMap
+ .filter(metadataInstance.topicIds.contains)
+ .map(topic => (topic, metadataInstance.topicIds(topic))).toMap
val updateMetadataRequestBuilder = new UpdateMetadataRequest.Builder(updateMetadataRequestVersion,
- controllerId, controllerEpoch, brokerEpoch, partitionStates.asJava, liveBrokers.asJava, topicIds.asJava)
+ controllerId, controllerEpoch, brokerEpoch, partitionStates.asJava, liveBrokers.asJava, topicIds.asJava, kraftController)
sendRequest(broker, updateMetadataRequestBuilder, (r: AbstractResponse) => {
val updateMetadataResponse = r.asInstanceOf[UpdateMetadataResponse]
- sendEvent(UpdateMetadataResponseReceived(updateMetadataResponse, broker))
+ handleUpdateMetadataResponse(updateMetadataResponse, broker)
})
}
@@ -567,13 +611,16 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
updateMetadataRequestPartitionInfoMap.clear()
}
+ def handleUpdateMetadataResponse(response: UpdateMetadataResponse, broker: Int): Unit
+
private def sendStopReplicaRequests(controllerEpoch: Int, stateChangeLog: StateChangeLogger): Unit = {
val traceEnabled = stateChangeLog.isTraceEnabled
+ val metadataVersion = metadataVersionProvider.apply()
val stopReplicaRequestVersion: Short =
- if (config.interBrokerProtocolVersion.isAtLeast(IBP_3_4_IV0)) 4
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_2_6_IV0)) 3
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_2_4_IV1)) 2
- else if (config.interBrokerProtocolVersion.isAtLeast(IBP_2_2_IV0)) 1
+ if (metadataVersion.isAtLeast(IBP_3_4_IV0)) 4
+ else if (metadataVersion.isAtLeast(IBP_2_6_IV0)) 3
+ else if (metadataVersion.isAtLeast(IBP_2_4_IV1)) 2
+ else if (metadataVersion.isAtLeast(IBP_2_2_IV0)) 1
else 0
def responseCallback(brokerId: Int, isPartitionDeleted: TopicPartition => Boolean)
@@ -582,25 +629,24 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
val partitionErrorsForDeletingTopics = mutable.Map.empty[TopicPartition, Errors]
stopReplicaResponse.partitionErrors.forEach { pe =>
val tp = new TopicPartition(pe.topicName, pe.partitionIndex)
- if (controllerContext.isTopicDeletionInProgress(pe.topicName) &&
+ if (metadataInstance.isTopicDeletionInProgress(pe.topicName) &&
isPartitionDeleted(tp)) {
partitionErrorsForDeletingTopics += tp -> Errors.forCode(pe.errorCode)
}
}
if (partitionErrorsForDeletingTopics.nonEmpty)
- sendEvent(TopicDeletionStopReplicaResponseReceived(brokerId, stopReplicaResponse.error,
- partitionErrorsForDeletingTopics))
+ handleStopReplicaResponse(stopReplicaResponse, brokerId, partitionErrorsForDeletingTopics.toMap)
}
stopReplicaRequestMap.forKeyValue { (brokerId, partitionStates) =>
- if (controllerContext.liveOrShuttingDownBrokerIds.contains(brokerId)) {
+ if (metadataInstance.liveOrShuttingDownBrokerIds.contains(brokerId)) {
if (traceEnabled)
partitionStates.forKeyValue { (topicPartition, partitionState) =>
stateChangeLog.trace(s"Sending StopReplica request $partitionState to " +
s"broker $brokerId for partition $topicPartition")
}
- val brokerEpoch = controllerContext.liveBrokerIdAndEpochs(brokerId)
+ val brokerEpoch = metadataInstance.liveBrokerIdAndEpochs(brokerId)
if (stopReplicaRequestVersion >= 3) {
val stopReplicaTopicState = mutable.Map.empty[String, StopReplicaTopicState]
partitionStates.forKeyValue { (topicPartition, partitionState) =>
@@ -613,7 +659,7 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
s"replicas to broker $brokerId")
val stopReplicaRequestBuilder = new StopReplicaRequest.Builder(
stopReplicaRequestVersion, controllerId, controllerEpoch, brokerEpoch,
- false, stopReplicaTopicState.values.toBuffer.asJava)
+ false, stopReplicaTopicState.values.toBuffer.asJava, kraftController)
sendRequest(brokerId, stopReplicaRequestBuilder,
responseCallback(brokerId, tp => partitionStates.get(tp).exists(_.deletePartition)))
} else {
@@ -640,7 +686,7 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
s"$numPartitionStateWithDelete replicas to broker $brokerId")
val stopReplicaRequestBuilder = new StopReplicaRequest.Builder(
stopReplicaRequestVersion, controllerId, controllerEpoch, brokerEpoch,
- true, topicStatesWithDelete.values.toBuffer.asJava)
+ true, topicStatesWithDelete.values.toBuffer.asJava, kraftController)
sendRequest(brokerId, stopReplicaRequestBuilder, responseCallback(brokerId, _ => true))
}
@@ -649,7 +695,7 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
s"$numPartitionStateWithoutDelete replicas to broker $brokerId")
val stopReplicaRequestBuilder = new StopReplicaRequest.Builder(
stopReplicaRequestVersion, controllerId, controllerEpoch, brokerEpoch,
- false, topicStatesWithoutDelete.values.toBuffer.asJava)
+ false, topicStatesWithoutDelete.values.toBuffer.asJava, kraftController)
sendRequest(brokerId, stopReplicaRequestBuilder)
}
}
@@ -659,6 +705,9 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
stopReplicaRequestMap.clear()
}
+ def handleStopReplicaResponse(stopReplicaResponse: StopReplicaResponse, brokerId: Int,
+ partitionErrorsForDeletingTopics: Map[TopicPartition, Errors]): Unit
+
def sendRequestsToBrokers(controllerEpoch: Int): Unit = {
try {
val stateChangeLog = stateChangeLogger.withControllerEpoch(controllerEpoch)
diff --git a/core/src/main/scala/kafka/controller/ControllerContext.scala b/core/src/main/scala/kafka/controller/ControllerContext.scala
index 7065d87c4c6..8c99a40c41f 100644
--- a/core/src/main/scala/kafka/controller/ControllerContext.scala
+++ b/core/src/main/scala/kafka/controller/ControllerContext.scala
@@ -17,6 +17,7 @@
package kafka.controller
+import kafka.api.LeaderAndIsr
import kafka.cluster.Broker
import kafka.utils.Implicits._
import org.apache.kafka.common.{TopicPartition, Uuid}
@@ -72,7 +73,7 @@ case class ReplicaAssignment private (replicas: Seq[Int],
s"removingReplicas=${removingReplicas.mkString(",")})"
}
-class ControllerContext {
+class ControllerContext extends ControllerChannelContext {
val stats = new ControllerStats
var offlinePartitionCount = 0
var preferredReplicaImbalanceCount = 0
@@ -230,7 +231,11 @@ class ControllerContext {
}.toSet
}
- def isReplicaOnline(brokerId: Int, topicPartition: TopicPartition, includeShuttingDownBrokers: Boolean = false): Boolean = {
+ def isReplicaOnline(brokerId: Int, topicPartition: TopicPartition): Boolean = {
+ isReplicaOnline(brokerId, topicPartition, includeShuttingDownBrokers = false)
+ }
+
+ def isReplicaOnline(brokerId: Int, topicPartition: TopicPartition, includeShuttingDownBrokers: Boolean): Boolean = {
val brokerOnline = {
if (includeShuttingDownBrokers) liveOrShuttingDownBrokerIds.contains(brokerId)
else liveBrokerIds.contains(brokerId)
@@ -445,6 +450,22 @@ class ControllerContext {
Some(replicaAssignment), Some(leaderIsrAndControllerEpoch))
}
+ def partitionLeaderAndIsr(partition: TopicPartition): Option[LeaderAndIsr] = {
+ partitionLeadershipInfo.get(partition).map(_.leaderAndIsr)
+ }
+
+ def leaderEpoch(partition: TopicPartition): Int = {
+ // A sentinel (-2) is used as an epoch if the topic is queued for deletion. It overrides
+ // any existing epoch.
+ if (isTopicQueuedUpForDeletion(partition.topic)) {
+ LeaderAndIsr.EpochDuringDelete
+ } else {
+ partitionLeadershipInfo.get(partition)
+ .map(_.leaderAndIsr.leaderEpoch)
+ .getOrElse(LeaderAndIsr.NoEpoch)
+ }
+ }
+
def partitionLeadershipInfo(partition: TopicPartition): Option[LeaderIsrAndControllerEpoch] = {
partitionLeadershipInfo.get(partition)
}
@@ -524,5 +545,4 @@ class ControllerContext {
private def isValidPartitionStateTransition(partition: TopicPartition, targetState: PartitionState): Boolean =
targetState.validPreviousStates.contains(partitionStates(partition))
-
}
diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala
index b473cd7afe6..c7b4277705a 100644
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@ -90,8 +90,14 @@ class KafkaController(val config: KafkaConfig,
private val isAlterPartitionEnabled = config.interBrokerProtocolVersion.isAlterPartitionSupported
private val stateChangeLogger = new StateChangeLogger(config.brokerId, inControllerContext = true, None)
val controllerContext = new ControllerContext
- var controllerChannelManager = new ControllerChannelManager(controllerContext, config, time, metrics,
- stateChangeLogger, threadNamePrefix)
+ var controllerChannelManager = new ControllerChannelManager(
+ () => controllerContext.epoch,
+ config,
+ time,
+ metrics,
+ stateChangeLogger,
+ threadNamePrefix
+ )
// have a separate scheduler for the controller to be able to start and stop independently of the kafka server
// visible for testing
@@ -928,7 +934,7 @@ class KafkaController(val config: KafkaConfig,
// update the leader and isr cache for all existing partitions from Zookeeper
updateLeaderAndIsrCache()
// start the channel manager
- controllerChannelManager.startup()
+ controllerChannelManager.startup(controllerContext.liveOrShuttingDownBrokers)
info(s"Currently active brokers in the cluster: ${controllerContext.liveBrokerIds}")
info(s"Currently shutting brokers in the cluster: ${controllerContext.shuttingDownBrokerIds}")
info(s"Current list of topics in the cluster: ${controllerContext.allTopics}")
diff --git a/core/src/main/scala/kafka/migration/MigrationControllerChannelContext.scala b/core/src/main/scala/kafka/migration/MigrationControllerChannelContext.scala
new file mode 100644
index 00000000000..cb7b9fb4f16
--- /dev/null
+++ b/core/src/main/scala/kafka/migration/MigrationControllerChannelContext.scala
@@ -0,0 +1,115 @@
+/*
+ * 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 kafka.migration
+
+import kafka.api.LeaderAndIsr
+import kafka.cluster.Broker
+import kafka.controller.{ControllerChannelContext, LeaderIsrAndControllerEpoch}
+import org.apache.kafka.common.{TopicPartition, Uuid}
+import org.apache.kafka.image.MetadataImage
+
+import scala.jdk.CollectionConverters._
+
+object MigrationControllerChannelContext {
+ def isReplicaOnline(image: MetadataImage, brokerId: Int, replicaAssignment: Set[Int]): Boolean = {
+ val brokerOnline = image.cluster().containsBroker(brokerId)
+ brokerOnline && replicaAssignment.contains(brokerId)
+ }
+
+ def partitionReplicaAssignment(image: MetadataImage, tp: TopicPartition): collection.Seq[Int] = {
+ image.topics().topicsByName().asScala.get(tp.topic()) match {
+ case Some(topic) => topic.partitions().asScala.get(tp.partition()) match {
+ case Some(partition) => partition.replicas.toSeq
+ case None => collection.Seq.empty
+ }
+ case None => collection.Seq.empty
+ }
+ }
+
+ def partitionLeadershipInfo(image: MetadataImage, topicPartition: TopicPartition): Option[LeaderIsrAndControllerEpoch] = {
+ image.topics().topicsByName().asScala.get(topicPartition.topic()) match {
+ case Some(topic) => topic.partitions().asScala.get(topicPartition.partition()) match {
+ case Some(partition) =>
+ val leaderAndIsr = LeaderAndIsr(partition.leader, partition.leaderEpoch, partition.isr.toList,
+ partition.leaderRecoveryState, partition.partitionEpoch)
+ Some(LeaderIsrAndControllerEpoch(leaderAndIsr, image.highestOffsetAndEpoch().epoch()))
+ case None => None
+ }
+ case None => None
+ }
+ }
+}
+
+sealed class MigrationControllerChannelContext(
+ val image: MetadataImage
+) extends ControllerChannelContext {
+ override def isTopicDeletionInProgress(topicName: String): Boolean = {
+ !image.topics().topicsByName().containsKey(topicName)
+ }
+
+ override val topicIds: collection.Map[String, Uuid] = {
+ image.topics().topicsByName().asScala.map {
+ case (name, topic) => name -> topic.id()
+ }.toMap
+ }
+
+ override val liveBrokerIdAndEpochs: collection.Map[Int, Long] = {
+ image.cluster().zkBrokers().asScala.map {
+ case (brokerId, broker) => brokerId.intValue() -> broker.epoch()
+ }
+ }
+
+ override val liveOrShuttingDownBrokers: collection.Set[Broker] = {
+ image.cluster().zkBrokers().asScala.values.map { registration =>
+ Broker.fromBrokerRegistration(registration)
+ }.toSet
+ }
+
+ override def isTopicQueuedUpForDeletion(topic: String): Boolean = {
+ !image.topics().topicsByName().containsKey(topic)
+ }
+
+ override def isReplicaOnline(brokerId: Int, partition: TopicPartition): Boolean = {
+ MigrationControllerChannelContext.isReplicaOnline(
+ image, brokerId, partitionReplicaAssignment(partition).toSet)
+ }
+
+ override def partitionReplicaAssignment(tp: TopicPartition): collection.Seq[Int] = {
+ MigrationControllerChannelContext.partitionReplicaAssignment(image, tp)
+ }
+
+ override def leaderEpoch(topicPartition: TopicPartition): Int = {
+ // Topic is deleted use a special sentinel -2 to the indicate the same.
+ if (isTopicQueuedUpForDeletion(topicPartition.topic())) {
+ LeaderAndIsr.EpochDuringDelete
+ } else {
+ image.topics().topicsByName.asScala.get(topicPartition.topic()) match {
+ case Some(topic) => topic.partitions().asScala.get(topicPartition.partition()) match {
+ case Some(partition) => partition.leaderEpoch
+ case None => LeaderAndIsr.NoEpoch
+ }
+ case None => LeaderAndIsr.NoEpoch
+ }
+ }
+ }
+
+ override val liveOrShuttingDownBrokerIds: collection.Set[Int] = liveBrokerIdAndEpochs.keySet
+
+ override def partitionLeadershipInfo(topicPartition: TopicPartition): Option[LeaderIsrAndControllerEpoch] = {
+ MigrationControllerChannelContext.partitionLeadershipInfo(image, topicPartition)
+ }
+}
\ No newline at end of file
diff --git a/core/src/main/scala/kafka/migration/MigrationPropagator.scala b/core/src/main/scala/kafka/migration/MigrationPropagator.scala
new file mode 100644
index 00000000000..9c61ffd7aef
--- /dev/null
+++ b/core/src/main/scala/kafka/migration/MigrationPropagator.scala
@@ -0,0 +1,204 @@
+/*
+ * 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 kafka.migration
+
+import kafka.api.LeaderAndIsr
+import kafka.cluster.Broker
+import kafka.controller.{ControllerChannelContext, ControllerChannelManager, ReplicaAssignment, StateChangeLogger}
+import kafka.server.KafkaConfig
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.metrics.Metrics
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.image.{MetadataDelta, MetadataImage}
+import org.apache.kafka.metadata.migration.LegacyPropagator
+import org.apache.kafka.server.common.MetadataVersion
+
+import java.util
+import scala.jdk.CollectionConverters._
+
+class MigrationPropagator(
+ nodeId: Int,
+ config: KafkaConfig,
+ metadataVersionProvider: () => MetadataVersion,
+) extends LegacyPropagator {
+ @volatile private var _image = MetadataImage.EMPTY
+ val stateChangeLogger = new StateChangeLogger(nodeId, inControllerContext = true, None)
+ val channelManager = new ControllerChannelManager(
+ () => _image.highestOffsetAndEpoch().epoch(),
+ config,
+ Time.SYSTEM,
+ new Metrics(),
+ stateChangeLogger
+ )
+
+ val requestBatch = new MigrationPropagatorBatch(
+ config,
+ metadataProvider,
+ metadataVersionProvider,
+ channelManager,
+ stateChangeLogger
+ )
+
+ private def metadataProvider(): ControllerChannelContext = {
+ new MigrationControllerChannelContext(_image)
+ }
+
+ def startup(): Unit = {
+ channelManager.startup(Set.empty)
+ }
+
+ def shutdown(): Unit = {
+ clear()
+ channelManager.shutdown()
+ }
+
+ override def publishMetadata(image: MetadataImage): Unit = {
+ val oldImage = _image
+ val addedBrokers = new util.HashSet[Integer](image.cluster().brokers().keySet())
+ addedBrokers.removeAll(oldImage.cluster().brokers().keySet())
+ val removedBrokers = new util.HashSet[Integer](oldImage.cluster().brokers().keySet())
+ removedBrokers.removeAll(image.cluster().brokers().keySet())
+
+ removedBrokers.asScala.foreach(id => channelManager.removeBroker(id))
+ addedBrokers.asScala.foreach(id =>
+ channelManager.addBroker(Broker.fromBrokerRegistration(image.cluster().broker(id))))
+ _image = image
+ }
+
+ override def sendRPCsToBrokersFromMetadataDelta(delta: MetadataDelta, image: MetadataImage,
+ zkControllerEpoch: Int): Unit = {
+ publishMetadata(image)
+ requestBatch.newBatch()
+
+ delta.getOrCreateTopicsDelta()
+ delta.getOrCreateClusterDelta()
+
+ val changedZkBrokers = delta.clusterDelta().liveZkBrokerIdChanges().asScala.map(_.toInt).toSet
+ val zkBrokers = image.cluster().zkBrokers().keySet().asScala.map(_.toInt).toSet
+ val oldZkBrokers = zkBrokers -- changedZkBrokers
+ val brokersChanged = !delta.clusterDelta().changedBrokers().isEmpty
+
+ if (changedZkBrokers.nonEmpty) {
+ // Update new Zk brokers about all the metadata.
+ requestBatch.addUpdateMetadataRequestForBrokers(changedZkBrokers.toSeq, image.topics().partitions().keySet().asScala)
+ // Send these requests first to make sure, we don't add all the partition metadata to the
+ // old brokers as well.
+ requestBatch.sendRequestsToBrokers(zkControllerEpoch)
+ requestBatch.newBatch()
+
+ // For new the brokers, check if there are partition assignments and add LISR appropriately.
+ image.topics().partitions().asScala.foreach { case (tp, partitionRegistration) =>
+ val replicas = partitionRegistration.replicas.toSet
+ val leaderIsrAndControllerEpochOpt = MigrationControllerChannelContext.partitionLeadershipInfo(image, tp)
+ val newBrokersWithReplicas = replicas.intersect(changedZkBrokers)
+ if (newBrokersWithReplicas.nonEmpty) {
+ leaderIsrAndControllerEpochOpt match {
+ case Some(leaderIsrAndControllerEpoch) =>
+ val replicaAssignment = ReplicaAssignment(partitionRegistration.replicas,
+ partitionRegistration.addingReplicas, partitionRegistration.removingReplicas)
+ requestBatch.addLeaderAndIsrRequestForBrokers(newBrokersWithReplicas.toSeq, tp,
+ leaderIsrAndControllerEpoch, replicaAssignment, isNew = true)
+ case None =>
+ }
+ }
+ }
+ }
+
+ // If there are new brokers (including KRaft brokers) or if there are changes in topic
+ // metadata, let's send UMR about the changes to the old Zk brokers.
+ if (brokersChanged || !delta.topicsDelta().deletedTopicIds().isEmpty || !delta.topicsDelta().changedTopics().isEmpty) {
+ requestBatch.addUpdateMetadataRequestForBrokers(oldZkBrokers.toSeq)
+ }
+
+ // Handle deleted topics by sending appropriate StopReplica and UMR requests to the brokers.
+ delta.topicsDelta().deletedTopicIds().asScala.foreach { deletedTopicId =>
+ val deletedTopic = delta.image().topics().getTopic(deletedTopicId)
+ deletedTopic.partitions().asScala.foreach { case (partition, partitionRegistration) =>
+ val tp = new TopicPartition(deletedTopic.name(), partition)
+ val offlineReplicas = partitionRegistration.replicas.filter {
+ MigrationControllerChannelContext.isReplicaOnline(image, _, partitionRegistration.replicas.toSet)
+ }
+ val deletedLeaderAndIsr = LeaderAndIsr.duringDelete(partitionRegistration.isr.toList)
+ requestBatch.addStopReplicaRequestForBrokers(partitionRegistration.replicas, tp, deletePartition = true)
+ requestBatch.addUpdateMetadataRequestForBrokers(
+ oldZkBrokers.toSeq, zkControllerEpoch, tp, deletedLeaderAndIsr.leader, deletedLeaderAndIsr.leaderEpoch,
+ deletedLeaderAndIsr.partitionEpoch, deletedLeaderAndIsr.isr, partitionRegistration.replicas, offlineReplicas)
+ }
+ }
+
+ // Handle changes in other topics and send appropriate LeaderAndIsr and UMR requests to the
+ // brokers.
+ delta.topicsDelta().changedTopics().asScala.foreach { case (_, topicDelta) =>
+ topicDelta.partitionChanges().asScala.foreach { case (partition, partitionRegistration) =>
+ val tp = new TopicPartition(topicDelta.name(), partition)
+
+ // Check for replica leadership changes.
+ val leaderIsrAndControllerEpochOpt = MigrationControllerChannelContext.partitionLeadershipInfo(image, tp)
+ leaderIsrAndControllerEpochOpt match {
+ case Some(leaderIsrAndControllerEpoch) =>
+ val replicaAssignment = ReplicaAssignment(partitionRegistration.replicas,
+ partitionRegistration.addingReplicas, partitionRegistration.removingReplicas)
+ requestBatch.addLeaderAndIsrRequestForBrokers(replicaAssignment.replicas, tp,
+ leaderIsrAndControllerEpoch, replicaAssignment, isNew = true)
+ case None =>
+ }
+
+ // Check for removed replicas.
+ val oldReplicas =
+ Option(delta.image().topics().getPartition(topicDelta.id(), tp.partition()))
+ .map(_.replicas.toSet)
+ .getOrElse(Set.empty)
+ val newReplicas = partitionRegistration.replicas.toSet
+ val removedReplicas = oldReplicas -- newReplicas
+ if (removedReplicas.nonEmpty) {
+ requestBatch.addStopReplicaRequestForBrokers(removedReplicas.toSeq, tp, deletePartition = false)
+ }
+ }
+ }
+ // Send all the accumulated requests to the broker.
+ requestBatch.sendRequestsToBrokers(zkControllerEpoch)
+ }
+
+ override def sendRPCsToBrokersFromMetadataImage(image: MetadataImage, zkControllerEpoch: Int): Unit = {
+ publishMetadata(image)
+ requestBatch.newBatch()
+
+ // When we need to send RPCs from the image, we're sending 'full' requests meaning we let
+ // every broker know about all the metadata and all the LISR requests it needs to handle.
+ // Note that we cannot send StopReplica requests from the image. We don't have any state
+ // about brokers that host a replica but are not part of the replica set known by the Controller.
+ val zkBrokers = image.cluster().zkBrokers().keySet().asScala.map(_.toInt).toSeq
+ val partitions = image.topics().partitions()
+ partitions.asScala.foreach{ case (tp, partitionRegistration) =>
+ val leaderIsrAndControllerEpochOpt = MigrationControllerChannelContext.partitionLeadershipInfo(image, tp)
+ leaderIsrAndControllerEpochOpt match {
+ case Some(leaderIsrAndControllerEpoch) =>
+ val replicaAssignment = ReplicaAssignment(partitionRegistration.replicas,
+ partitionRegistration.addingReplicas, partitionRegistration.removingReplicas)
+ requestBatch.addLeaderAndIsrRequestForBrokers(replicaAssignment.replicas, tp,
+ leaderIsrAndControllerEpoch, replicaAssignment, isNew = true)
+ case None => None
+ }
+ }
+ requestBatch.addUpdateMetadataRequestForBrokers(zkBrokers, partitions.keySet().asScala)
+ requestBatch.sendRequestsToBrokers(zkControllerEpoch)
+ }
+
+ override def clear(): Unit = {
+ requestBatch.clear()
+ }
+}
diff --git a/core/src/main/scala/kafka/migration/MigrationPropagatorBatch.scala b/core/src/main/scala/kafka/migration/MigrationPropagatorBatch.scala
new file mode 100644
index 00000000000..7c4952c9382
--- /dev/null
+++ b/core/src/main/scala/kafka/migration/MigrationPropagatorBatch.scala
@@ -0,0 +1,85 @@
+/*
+ * 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 kafka.migration
+
+import kafka.controller.{AbstractControllerBrokerRequestBatch, ControllerChannelContext, ControllerChannelManager, StateChangeLogger}
+import kafka.server.KafkaConfig
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests._
+import org.apache.kafka.server.common.MetadataVersion
+
+import scala.collection.mutable.ArrayBuffer
+import scala.jdk.CollectionConverters._
+
+sealed class MigrationPropagatorBatch(
+ config: KafkaConfig,
+ metadataProvider: () => ControllerChannelContext,
+ metadataVersionProvider: () => MetadataVersion,
+ controllerChannelManager: ControllerChannelManager,
+ stateChangeLogger: StateChangeLogger
+) extends AbstractControllerBrokerRequestBatch(
+ config,
+ metadataProvider,
+ metadataVersionProvider,
+ stateChangeLogger,
+ kraftController = true,
+) {
+ override def sendRequest(brokerId: Int, request: AbstractControlRequest.Builder[_ <: AbstractControlRequest], callback: AbstractResponse => Unit): Unit = {
+ controllerChannelManager.sendRequest(brokerId, request, callback)
+ }
+
+ override def handleLeaderAndIsrResponse(response: LeaderAndIsrResponse, broker: Int): Unit = {
+ if (response.error != Errors.NONE) {
+ stateChangeLogger.error(s"Received error ${response.error} in LeaderAndIsr " +
+ s"response $response from broker $broker")
+ return
+ }
+ val partitionErrors = response.partitionErrors(
+ metadataProvider().topicIds.map { case (id, string) => (string, id) }.asJava)
+ val offlineReplicas = new ArrayBuffer[TopicPartition]()
+ partitionErrors.forEach{ case(tp, error) =>
+ if (error == Errors.KAFKA_STORAGE_ERROR) {
+ offlineReplicas += tp
+ }
+ }
+ if (offlineReplicas.nonEmpty) {
+ stateChangeLogger.error(s"Found ${offlineReplicas.mkString(",")} on broker $broker as offline")
+ }
+ }
+
+ override def handleUpdateMetadataResponse(response: UpdateMetadataResponse, broker: Int): Unit = {
+ if (response.error != Errors.NONE) {
+ stateChangeLogger.error(s"Received error ${response.error} in UpdateMetadata " +
+ s"response $response from broker $broker")
+ }
+ }
+
+ override def handleStopReplicaResponse(response: StopReplicaResponse, broker: Int,
+ partitionErrorsForDeletingTopics: Map[TopicPartition, Errors]): Unit = {
+ if (response.error() != Errors.NONE) {
+ stateChangeLogger.error(s"Received error ${response.error} in StopReplica " +
+ s"response $response from broker $broker")
+ }
+ partitionErrorsForDeletingTopics.foreach{ case(tp, error) =>
+ if (error != Errors.NONE) {
+ stateChangeLogger.error(s"Received error $error in StopReplica request for partition $tp " +
+ s"from broker $broker")
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/core/src/test/scala/unit/kafka/controller/ControllerChannelManagerTest.scala b/core/src/test/scala/unit/kafka/controller/ControllerChannelManagerTest.scala
index 8e9176af2e9..44a14386f7f 100644
--- a/core/src/test/scala/unit/kafka/controller/ControllerChannelManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/controller/ControllerChannelManagerTest.scala
@@ -941,14 +941,11 @@ class ControllerChannelManagerTest {
private case class SentRequest(request: ControlRequest, responseCallback: AbstractResponse => Unit)
private class MockControllerBrokerRequestBatch(context: ControllerContext, config: KafkaConfig = config)
- extends AbstractControllerBrokerRequestBatch(config, context, logger) {
+ extends AbstractControllerBrokerRequestBatch(config, () => context, () => config.interBrokerProtocolVersion, logger) {
val sentEvents = ListBuffer.empty[ControllerEvent]
val sentRequests = mutable.Map.empty[Int, ListBuffer[SentRequest]]
- override def sendEvent(event: ControllerEvent): Unit = {
- sentEvents.append(event)
- }
override def sendRequest(brokerId: Int, request: ControlRequest, callback: AbstractResponse => Unit): Unit = {
sentRequests.getOrElseUpdate(brokerId, ListBuffer.empty)
sentRequests(brokerId).append(SentRequest(request, callback))
@@ -980,6 +977,21 @@ class ControllerChannelManagerTest {
case None => List.empty[LeaderAndIsrRequest]
}
}
+
+ override def handleLeaderAndIsrResponse(response: LeaderAndIsrResponse, broker: Int): Unit = {
+ sentEvents.append(LeaderAndIsrResponseReceived(response, broker))
+ }
+
+ override def handleUpdateMetadataResponse(response: UpdateMetadataResponse, broker: Int)
+ : Unit = {
+ sentEvents.append(UpdateMetadataResponseReceived(response, broker))
+ }
+
+ override def handleStopReplicaResponse(stopReplicaResponse: StopReplicaResponse,
+ brokerId: Int, partitionErrorsForDeletingTopics: Map[TopicPartition, Errors]): Unit = {
+ if (partitionErrorsForDeletingTopics.nonEmpty)
+ sentEvents.append(TopicDeletionStopReplicaResponseReceived(brokerId, stopReplicaResponse.error, partitionErrorsForDeletingTopics))
+ }
}
}
diff --git a/core/src/test/scala/unit/kafka/server/BrokerEpochIntegrationTest.scala b/core/src/test/scala/unit/kafka/server/BrokerEpochIntegrationTest.scala
index 250f87cbd73..02fd74a5aa0 100755
--- a/core/src/test/scala/unit/kafka/server/BrokerEpochIntegrationTest.scala
+++ b/core/src/test/scala/unit/kafka/server/BrokerEpochIntegrationTest.scala
@@ -131,9 +131,14 @@ class BrokerEpochIntegrationTest extends QuorumTestHarness {
val controllerContext = new ControllerContext
controllerContext.setLiveBrokers(brokerAndEpochs)
val metrics = new Metrics
- val controllerChannelManager = new ControllerChannelManager(controllerContext, controllerConfig, Time.SYSTEM,
- metrics, new StateChangeLogger(controllerId, inControllerContext = true, None))
- controllerChannelManager.startup()
+ val controllerChannelManager = new ControllerChannelManager(
+ () => controllerContext.epoch,
+ controllerConfig,
+ Time.SYSTEM,
+ metrics,
+ new StateChangeLogger(controllerId, inControllerContext = true, None)
+ )
+ controllerChannelManager.startup(controllerContext.liveOrShuttingDownBrokers)
val broker2 = servers(brokerId2)
val epochInRequest = broker2.kafkaController.brokerEpoch + epochInRequestDiffFromCurrentEpoch
diff --git a/core/src/test/scala/unit/kafka/server/LeaderElectionTest.scala b/core/src/test/scala/unit/kafka/server/LeaderElectionTest.scala
index f0dea91335e..259ea91df87 100755
--- a/core/src/test/scala/unit/kafka/server/LeaderElectionTest.scala
+++ b/core/src/test/scala/unit/kafka/server/LeaderElectionTest.scala
@@ -138,9 +138,14 @@ class LeaderElectionTest extends QuorumTestHarness {
val controllerContext = new ControllerContext
controllerContext.setLiveBrokers(brokerAndEpochs)
val metrics = new Metrics
- val controllerChannelManager = new ControllerChannelManager(controllerContext, controllerConfig, Time.SYSTEM,
- metrics, new StateChangeLogger(controllerId, inControllerContext = true, None))
- controllerChannelManager.startup()
+ val controllerChannelManager = new ControllerChannelManager(
+ () => controllerContext.epoch,
+ controllerConfig,
+ Time.SYSTEM,
+ metrics,
+ new StateChangeLogger(controllerId, inControllerContext = true, None)
+ )
+ controllerChannelManager.startup(controllerContext.liveOrShuttingDownBrokers)
try {
val staleControllerEpoch = 0
val partitionStates = Seq(
diff --git a/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala b/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala
index 70554d9427c..50b16498037 100644
--- a/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala
@@ -280,9 +280,13 @@ class ServerShutdownTest extends KafkaServerTestHarness {
val controllerConfig = KafkaConfig.fromProps(TestUtils.createBrokerConfig(controllerId, zkConnect))
val controllerContext = new ControllerContext
controllerContext.setLiveBrokers(brokerAndEpochs)
- controllerChannelManager = new ControllerChannelManager(controllerContext, controllerConfig, Time.SYSTEM,
- metrics, new StateChangeLogger(controllerId, inControllerContext = true, None))
- controllerChannelManager.startup()
+ controllerChannelManager = new ControllerChannelManager(
+ () => controllerContext.epoch,
+ controllerConfig,
+ Time.SYSTEM,
+ metrics,
+ new StateChangeLogger(controllerId, inControllerContext = true, None))
+ controllerChannelManager.startup(controllerContext.liveOrShuttingDownBrokers)
// Initiate a sendRequest and wait until connection is established and one byte is received by the peer
val requestBuilder = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion,
diff --git a/metadata/src/main/java/org/apache/kafka/image/ClusterDelta.java b/metadata/src/main/java/org/apache/kafka/image/ClusterDelta.java
index 39d6fdb3d74..fd50b1ba3e2 100644
--- a/metadata/src/main/java/org/apache/kafka/image/ClusterDelta.java
+++ b/metadata/src/main/java/org/apache/kafka/image/ClusterDelta.java
@@ -31,6 +31,8 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
/**
@@ -56,6 +58,16 @@ public final class ClusterDelta {
return image.broker(nodeId);
}
+ public Set<Integer> liveZkBrokerIdChanges() {
+ return changedBrokers
+ .values()
+ .stream()
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .map(BrokerRegistration::id)
+ .collect(Collectors.toSet());
+ }
+
public void finishSnapshot() {
for (Integer brokerId : image.brokers().keySet()) {
if (!changedBrokers.containsKey(brokerId)) {
diff --git a/metadata/src/main/java/org/apache/kafka/image/ClusterImage.java b/metadata/src/main/java/org/apache/kafka/image/ClusterImage.java
index 05ba74412ba..b39a88725f0 100644
--- a/metadata/src/main/java/org/apache/kafka/image/ClusterImage.java
+++ b/metadata/src/main/java/org/apache/kafka/image/ClusterImage.java
@@ -47,10 +47,25 @@ public final class ClusterImage {
return brokers;
}
+ public Map<Integer, BrokerRegistration> zkBrokers() {
+ return Collections.unmodifiableMap(
+ brokers
+ .entrySet()
+ .stream()
+ .filter(x -> x.getValue().isMigratingZkBroker())
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
+ }
+
public BrokerRegistration broker(int nodeId) {
return brokers.get(nodeId);
}
+
+
+ public boolean containsBroker(int brokerId) {
+ return brokers.containsKey(brokerId);
+ }
+
public void write(ImageWriter writer, ImageWriterOptions options) {
for (BrokerRegistration broker : brokers.values()) {
writer.write(broker.toRecord(options));
diff --git a/metadata/src/main/java/org/apache/kafka/image/TopicsDelta.java b/metadata/src/main/java/org/apache/kafka/image/TopicsDelta.java
index 66f12102a59..d3c5888fa7a 100644
--- a/metadata/src/main/java/org/apache/kafka/image/TopicsDelta.java
+++ b/metadata/src/main/java/org/apache/kafka/image/TopicsDelta.java
@@ -53,6 +53,8 @@ public final class TopicsDelta {
*/
private final Set<Uuid> deletedTopicIds = new HashSet<>();
+ private final Set<Uuid> createdTopicIds = new HashSet<>();
+
public TopicsDelta(TopicsImage image) {
this.image = image;
}
@@ -69,6 +71,7 @@ public final class TopicsDelta {
TopicDelta delta = new TopicDelta(
new TopicImage(record.name(), record.topicId(), Collections.emptyMap()));
changedTopics.put(record.topicId(), delta);
+ createdTopicIds.add(record.topicId());
}
TopicDelta getOrCreateTopicDelta(Uuid id) {
@@ -170,6 +173,10 @@ public final class TopicsDelta {
return deletedTopicIds;
}
+ public Set<Uuid> createdTopicIds() {
+ return createdTopicIds;
+ }
+
/**
* Find the topic partitions that have change based on the replica given.
*
@@ -212,6 +219,7 @@ public final class TopicsDelta {
return "TopicsDelta(" +
"changedTopics=" + changedTopics +
", deletedTopicIds=" + deletedTopicIds +
+ ", createdTopicIds=" + createdTopicIds +
')';
}
}
diff --git a/metadata/src/main/java/org/apache/kafka/image/TopicsImage.java b/metadata/src/main/java/org/apache/kafka/image/TopicsImage.java
index 5f7db112f0a..392bea4119b 100644
--- a/metadata/src/main/java/org/apache/kafka/image/TopicsImage.java
+++ b/metadata/src/main/java/org/apache/kafka/image/TopicsImage.java
@@ -17,6 +17,7 @@
package org.apache.kafka.image;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.image.writer.ImageWriter;
import org.apache.kafka.image.writer.ImageWriterOptions;
@@ -24,6 +25,7 @@ import org.apache.kafka.metadata.PartitionRegistration;
import org.apache.kafka.server.util.TranslatedValueMapView;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
@@ -110,6 +112,14 @@ public final class TopicsImage {
return new TranslatedValueMapView<>(topicsById, image -> image.name());
}
+ public Map<TopicPartition, PartitionRegistration> partitions() {
+ Map<TopicPartition, PartitionRegistration> partitions = new HashMap<>();
+ topicsById.values().forEach(topic -> {
+ topic.partitions().forEach((key, value) -> partitions.put(new TopicPartition(topic.name(), key), value));
+ });
+ return partitions;
+ }
+
@Override
public String toString() {
return "TopicsImage(topicsById=" + topicsById.entrySet().stream().
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/migration/LegacyPropagator.java b/metadata/src/main/java/org/apache/kafka/metadata/migration/LegacyPropagator.java
new file mode 100644
index 00000000000..c2bd4e6f3d0
--- /dev/null
+++ b/metadata/src/main/java/org/apache/kafka/metadata/migration/LegacyPropagator.java
@@ -0,0 +1,37 @@
+/*
+ * 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.metadata.migration;
+
+import org.apache.kafka.image.MetadataDelta;
+import org.apache.kafka.image.MetadataImage;
+
+public interface LegacyPropagator {
+
+ void startup();
+
+ void shutdown();
+
+ void publishMetadata(MetadataImage image);
+
+ void sendRPCsToBrokersFromMetadataDelta(MetadataDelta delta,
+ MetadataImage image,
+ int zkControllerEpoch);
+
+ void sendRPCsToBrokersFromMetadataImage(MetadataImage image, int zkControllerEpoch);
+
+ void clear();
+}