From bfc99c32233eb7dfa25d9e041792af3e025b6d20 Mon Sep 17 00:00:00 2001 From: dengziming Date: Mon, 1 Mar 2021 21:15:05 +0800 Subject: [PATCH] KAFKA-12388: WIP share broker channel --- .../scala/kafka/server/AlterIsrManager.scala | 36 +++++-------------- .../kafka/server/BrokerLifecycleManager.scala | 2 -- .../scala/kafka/server/BrokerServer.scala | 22 +++++++----- .../main/scala/kafka/server/KafkaServer.scala | 30 ++++++++++++---- .../kafka/server/AlterIsrManagerTest.scala | 21 ++++------- 5 files changed, 54 insertions(+), 57 deletions(-) diff --git a/core/src/main/scala/kafka/server/AlterIsrManager.scala b/core/src/main/scala/kafka/server/AlterIsrManager.scala index e6a934df61423..d95e1b5a9baa5 100644 --- a/core/src/main/scala/kafka/server/AlterIsrManager.scala +++ b/core/src/main/scala/kafka/server/AlterIsrManager.scala @@ -17,20 +17,20 @@ package kafka.server import java.util +import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.{ConcurrentHashMap, TimeUnit} + import kafka.api.LeaderAndIsr import kafka.metrics.KafkaMetricsGroup -import kafka.utils.{KafkaScheduler, Logging, Scheduler} +import kafka.utils.{Logging, Scheduler} import kafka.zk.KafkaZkClient import org.apache.kafka.clients.ClientResponse import org.apache.kafka.common.TopicPartition import org.apache.kafka.common.message.{AlterIsrRequestData, AlterIsrResponseData} -import org.apache.kafka.common.metrics.Metrics import org.apache.kafka.common.protocol.Errors import org.apache.kafka.common.requests.{AlterIsrRequest, AlterIsrResponse} import org.apache.kafka.common.utils.Time -import java.util.concurrent.atomic.AtomicBoolean import scala.collection.mutable import scala.collection.mutable.ListBuffer import scala.jdk.CollectionConverters._ @@ -64,26 +64,12 @@ object AlterIsrManager { * Factory to AlterIsr based implementation, used when IBP >= 2.7-IV2 */ def apply( - config: KafkaConfig, - metadataCache: MetadataCache, - scheduler: KafkaScheduler, + channelManager: BrokerToControllerChannelManager, + scheduler: Scheduler, time: Time, - metrics: Metrics, - threadNamePrefix: Option[String], - brokerEpochSupplier: () => Long, - brokerId: Int + brokerId: Int, + brokerEpochSupplier: () => Long ): AlterIsrManager = { - val nodeProvider = MetadataCacheControllerNodeProvider(config, metadataCache) - - val channelManager = BrokerToControllerChannelManager( - controllerNodeProvider = nodeProvider, - time = time, - metrics = metrics, - config = config, - channelName = "alterIsr", - threadNamePrefix = threadNamePrefix, - retryTimeoutMs = Long.MaxValue - ) new DefaultAlterIsrManager( controllerChannelManager = channelManager, scheduler = scheduler, @@ -120,13 +106,9 @@ class DefaultAlterIsrManager( // Used to allow only one in-flight request at a time private val inflightRequest: AtomicBoolean = new AtomicBoolean(false) - override def start(): Unit = { - controllerChannelManager.start() - } + override def start(): Unit = {} - override def shutdown(): Unit = { - controllerChannelManager.shutdown() - } + override def shutdown(): Unit = {} override def submit(alterIsrItem: AlterIsrItem): Boolean = { val enqueued = unsentIsrUpdates.putIfAbsent(alterIsrItem.topicPartition, alterIsrItem) == null diff --git a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala index b15fc1c481a6d..a0ce18e3c1ef8 100644 --- a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala +++ b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala @@ -251,7 +251,6 @@ class BrokerLifecycleManager(val config: KafkaConfig, override def run(): Unit = { _highestMetadataOffsetProvider = highestMetadataOffsetProvider _channelManager = channelManager - _channelManager.start() _state = BrokerState.STARTING _clusterId = clusterId _advertisedListeners = advertisedListeners.duplicate() @@ -473,7 +472,6 @@ class BrokerLifecycleManager(val config: KafkaConfig, controlledShutdownFuture.complete(null) initialCatchUpFuture.cancel(false) if (_channelManager != null) { - _channelManager.shutdown() _channelManager = null } } diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala index aecb71ea4b045..4f0bfdf78f41d 100644 --- a/core/src/main/scala/kafka/server/BrokerServer.scala +++ b/core/src/main/scala/kafka/server/BrokerServer.scala @@ -106,6 +106,8 @@ class BrokerServer( var forwardingManager: ForwardingManager = null + var brokerToControllerChannelManager: BrokerToControllerChannelManager = null + var alterIsrManager: AlterIsrManager = null var autoTopicCreationManager: AutoTopicCreationManager = null @@ -186,7 +188,7 @@ class BrokerServer( time, metrics, config, - channelName = "forwarding", + channelName = "clientForwarding", threadNamePrefix, retryTimeoutMs = 60000 ) @@ -207,20 +209,22 @@ class BrokerServer( socketServer = new SocketServer(config, metrics, time, credentialProvider, apiVersionManager) socketServer.startup(startProcessingRequests = false) - val alterIsrChannelManager = BrokerToControllerChannelManager( + brokerToControllerChannelManager = BrokerToControllerChannelManager( controllerNodeProvider, time, metrics, config, - channelName = "alterIsr", + channelName = "brokerForwarding", threadNamePrefix, retryTimeoutMs = Long.MaxValue ) - alterIsrManager = new DefaultAlterIsrManager( - controllerChannelManager = alterIsrChannelManager, + brokerToControllerChannelManager.start() + + alterIsrManager = AlterIsrManager( + brokerToControllerChannelManager, scheduler = kafkaScheduler, time = time, - brokerId = config.nodeId, + brokerId = config.brokerId, brokerEpochSupplier = () => lifecycleManager.brokerEpoch() ) alterIsrManager.start() @@ -277,8 +281,7 @@ class BrokerServer( setSecurityProtocol(ep.securityProtocol.id)) } lifecycleManager.start(() => brokerMetadataListener.highestMetadataOffset(), - BrokerToControllerChannelManager(controllerNodeProvider, time, metrics, config, - "heartbeat", threadNamePrefix, config.brokerSessionTimeoutMs.toLong), + brokerToControllerChannelManager, metaProps.clusterId, networkListeners, supportedFeatures) // Register a listener with the Raft layer to receive metadata event notifications @@ -450,6 +453,9 @@ class BrokerServer( if (clientToControllerChannelManager != null) CoreUtils.swallow(clientToControllerChannelManager.shutdown(), this) + if (brokerToControllerChannelManager != null) + CoreUtils.swallow(brokerToControllerChannelManager.shutdown(), this) + if (logManager != null) CoreUtils.swallow(logManager.shutdown(), this) diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala index 947c23bf8c749..7e78cf3cce29a 100755 --- a/core/src/main/scala/kafka/server/KafkaServer.scala +++ b/core/src/main/scala/kafka/server/KafkaServer.scala @@ -135,8 +135,12 @@ class KafkaServer( var autoTopicCreationManager: AutoTopicCreationManager = null + /** Channel to send client request to controller, such as auto topic creation and forwarding */ var clientToControllerChannelManager: Option[BrokerToControllerChannelManager] = None + /** Channel to send broker request to controller, such as AlterIsr(KIP-497) and AlterReplicaState(KIP-589) */ + var brokerToControllerChannelManager: Option[BrokerToControllerChannelManager] = None + var alterIsrManager: AlterIsrManager = null var kafkaScheduler: KafkaScheduler = null @@ -265,7 +269,7 @@ class KafkaServer( time = time, metrics = metrics, config = config, - channelName = "forwarding", + channelName = "clientForwarding", threadNamePrefix = threadNamePrefix, retryTimeoutMs = config.requestTimeoutMs.longValue) brokerToControllerManager.start() @@ -290,17 +294,29 @@ class KafkaServer( socketServer = new SocketServer(config, metrics, time, credentialProvider, apiVersionManager) socketServer.startup(startProcessingRequests = false) + // Currently we only support alterIsr, maybe alterReplicaState in the future + if (config.interBrokerProtocolVersion.isAlterIsrSupported) { + val nodeProvider = MetadataCacheControllerNodeProvider(config, metadataCache) + val channelManager = BrokerToControllerChannelManager( + controllerNodeProvider = nodeProvider, + time = time, + metrics = metrics, + config = config, + channelName = "brokerToControllerForwardingChannel", + threadNamePrefix = threadNamePrefix, + retryTimeoutMs = Long.MaxValue) + channelManager.start() + brokerToControllerChannelManager = Some(channelManager) + } + /* start replica manager */ alterIsrManager = if (config.interBrokerProtocolVersion.isAlterIsrSupported) { AlterIsrManager( - config = config, - metadataCache = metadataCache, + channelManager = brokerToControllerChannelManager.get, scheduler = kafkaScheduler, time = time, - metrics = metrics, - threadNamePrefix = threadNamePrefix, brokerEpochSupplier = () => kafkaController.brokerEpoch, - config.brokerId + brokerId = config.brokerId ) } else { AlterIsrManager(kafkaScheduler, time, zkClient) @@ -707,6 +723,8 @@ class KafkaServer( CoreUtils.swallow(clientToControllerChannelManager.foreach(_.shutdown()), this) + CoreUtils.swallow(brokerToControllerChannelManager.foreach(_.shutdown()), this) + if (logManager != null) CoreUtils.swallow(logManager.shutdown(), this) diff --git a/core/src/test/scala/unit/kafka/server/AlterIsrManagerTest.scala b/core/src/test/scala/unit/kafka/server/AlterIsrManagerTest.scala index 1074fd3157c2f..1e3554be48251 100644 --- a/core/src/test/scala/unit/kafka/server/AlterIsrManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/AlterIsrManagerTest.scala @@ -56,12 +56,11 @@ class AlterIsrManagerTest { @Test def testBasic(): Unit = { - EasyMock.expect(brokerToController.start()) EasyMock.expect(brokerToController.sendRequest(EasyMock.anyObject(), EasyMock.anyObject())).once() EasyMock.replay(brokerToController) val scheduler = new MockScheduler(time) - val alterIsrManager = new DefaultAlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2) + val alterIsrManager = AlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2) alterIsrManager.start() alterIsrManager.submit(AlterIsrItem(tp0, new LeaderAndIsr(1, 1, List(1,2,3), 10), _ => {}, 0)) EasyMock.verify(brokerToController) @@ -70,12 +69,11 @@ class AlterIsrManagerTest { @Test def testOverwriteWithinBatch(): Unit = { val capture = EasyMock.newCapture[AbstractRequest.Builder[AlterIsrRequest]]() - EasyMock.expect(brokerToController.start()) EasyMock.expect(brokerToController.sendRequest(EasyMock.capture(capture), EasyMock.anyObject())).once() EasyMock.replay(brokerToController) val scheduler = new MockScheduler(time) - val alterIsrManager = new DefaultAlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2) + val alterIsrManager = AlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2) alterIsrManager.start() // Only send one ISR update for a given topic+partition @@ -93,12 +91,11 @@ class AlterIsrManagerTest { val capture = EasyMock.newCapture[AbstractRequest.Builder[AlterIsrRequest]]() val callbackCapture = EasyMock.newCapture[ControllerRequestCompletionHandler]() - EasyMock.expect(brokerToController.start()) EasyMock.expect(brokerToController.sendRequest(EasyMock.capture(capture), EasyMock.capture(callbackCapture))).times(2) EasyMock.replay(brokerToController) val scheduler = new MockScheduler(time) - val alterIsrManager = new DefaultAlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2) + val alterIsrManager = AlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2) alterIsrManager.start() // First request will send batch of one @@ -166,7 +163,6 @@ class AlterIsrManagerTest { val isrs = Seq(AlterIsrItem(tp0, leaderAndIsr, _ => { }, 0)) val callbackCapture = EasyMock.newCapture[ControllerRequestCompletionHandler]() - EasyMock.expect(brokerToController.start()) EasyMock.expect(brokerToController.sendRequest(EasyMock.anyObject(), EasyMock.capture(callbackCapture))).times(1) EasyMock.replay(brokerToController) @@ -225,12 +221,11 @@ class AlterIsrManagerTest { private def testPartitionError(tp: TopicPartition, error: Errors): AlterIsrManager = { val callbackCapture = EasyMock.newCapture[ControllerRequestCompletionHandler]() EasyMock.reset(brokerToController) - EasyMock.expect(brokerToController.start()) EasyMock.expect(brokerToController.sendRequest(EasyMock.anyObject(), EasyMock.capture(callbackCapture))).once() EasyMock.replay(brokerToController) val scheduler = new MockScheduler(time) - val alterIsrManager = new DefaultAlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2) + val alterIsrManager = AlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2) alterIsrManager.start() var capturedError: Option[Errors] = None @@ -259,12 +254,11 @@ class AlterIsrManagerTest { def testOneInFlight(): Unit = { val callbackCapture = EasyMock.newCapture[ControllerRequestCompletionHandler]() EasyMock.reset(brokerToController) - EasyMock.expect(brokerToController.start()) EasyMock.expect(brokerToController.sendRequest(EasyMock.anyObject(), EasyMock.capture(callbackCapture))).once() EasyMock.replay(brokerToController) val scheduler = new MockScheduler(time) - val alterIsrManager = new DefaultAlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2) + val alterIsrManager = AlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2) alterIsrManager.start() // First submit will send the request @@ -291,16 +285,15 @@ class AlterIsrManagerTest { def testPartitionMissingInResponse(): Unit = { val callbackCapture = EasyMock.newCapture[ControllerRequestCompletionHandler]() EasyMock.reset(brokerToController) - EasyMock.expect(brokerToController.start()) EasyMock.expect(brokerToController.sendRequest(EasyMock.anyObject(), EasyMock.capture(callbackCapture))).once() EasyMock.replay(brokerToController) val scheduler = new MockScheduler(time) - val alterIsrManager = new DefaultAlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2) + val alterIsrManager = AlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2) alterIsrManager.start() val count = new AtomicInteger(0) - val callback = (result: Either[Errors, LeaderAndIsr]) => { + val callback = (_: Either[Errors, LeaderAndIsr]) => { count.incrementAndGet() return }