You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/02/10 01:36:51 UTC
[GitHub] [kafka] cmccabe opened a new pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
cmccabe opened a new pull request #10095:
URL: https://github.com/apache/kafka/pull/10095
The broker lifecycle manager handles registering the broker and sending periodic heartbeats, as described in KIP-631. Based on the responses it receives from the controller, it transitions the broker through the states described in BrokerState.java: STARTING, RECOVERY, RUNNING, PENDING_CONTROLLED_SHUTDOWN, etc.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573849465
##########
File path: core/src/main/scala/kafka/server/KafkaConfig.scala
##########
@@ -1497,6 +1509,9 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
var brokerId: Int = getInt(KafkaConfig.BrokerIdProp)
val nodeId: Int = getInt(KafkaConfig.NodeIdProp)
val processRoles: Set[ProcessRole] = parseProcessRoles()
+ val initialRegistrationTimeoutMs = getInt(KafkaConfig.InitialBrokerRegistrationTimeoutMs)
+ val brokerHeartbeatIntervalMs = getInt(KafkaConfig.BrokerHeartbeatIntervalMsProp)
+ val brokerSessionTimeoutMs = getInt(KafkaConfig.BrokerSessionTimeoutMsProp)
Review comment:
It's worth including the explicit type here since it's a public `val`. Are these meant to be nullable? It may be better to make then `Option[Int]` if so. Otherwise, we should make them `Int` so that they're easier to use (they're currently being inferred as `Integer`).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573953538
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
Review comment:
20% jitter is kind of a lot for the heartbeat, I think. 20% means we could be heartbeating every 2.4 s rather than 2 s.
Thinking about it more I'd be more comfortable with 2%. The point of jitter is just to avoid having a lot of heartbeats come in all at once, and I think 2% jitter is more than sufficient to accomplish that. With a very small injection of randomness each time the heartbeat times will drift apart naturally, without the need for big divergences.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mumrah commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
mumrah commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573877692
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ _state match {
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ info(s"Attempted to enter controlled shutdown state, but we are already in " +
+ s"that state.")
+ case BrokerState.RUNNING =>
+ info(s"Beginning controlled shutdown.")
+ _state = BrokerState.PENDING_CONTROLLED_SHUTDOWN
+ case _ =>
+ info(s"Skipping controlled shutdown because we are in state ${_state}.")
+ beginShutdown()
+ }
+ }
+ }
+
+ /**
+ * Enter the controlled shutdown state if we are in RUNNING state.
+ * Or, if we're not running, shut down immediately.
+ */
+ def beginControlledShutdown(): Unit = {
+ eventQueue.append(new BeginControlledShutdownEvent())
+ }
+
+ /**
+ * Start shutting down the BrokerLifecycleManager, but do not block.
+ */
+ def beginShutdown(): Unit = {
+ eventQueue.beginShutdown("beginShutdown", new ShutdownEvent())
+ }
+
+ /**
+ * Shut down the BrokerLifecycleManager and block until all threads are joined.
+ */
+ def close(): Unit = {
+ beginShutdown()
+ eventQueue.close()
+ }
+
+ class SetReadyToUnfenceEvent() extends EventQueue.Event {
+ override def run(): Unit = {
+ readyToUnfence = true
+ scheduleNextCommunicationImmediately()
+ }
+ }
+
+ class StartupEvent(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]) extends EventQueue.Event {
+ override def run(): Unit = {
+ _highestMetadataOffsetProvider = highestMetadataOffsetProvider
+ _channelManager = channelManager
+ _channelManager.start()
+ _state = BrokerState.STARTING
+ _clusterId = clusterId
+ _advertisedListeners = advertisedListeners
+ _supportedFeatures = supportedFeatures
Review comment:
Should we be copying from this map rather than assigning to it? Do we want things outside of this class to mutate the data in this map?
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ _state match {
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ info(s"Attempted to enter controlled shutdown state, but we are already in " +
+ s"that state.")
+ case BrokerState.RUNNING =>
+ info(s"Beginning controlled shutdown.")
+ _state = BrokerState.PENDING_CONTROLLED_SHUTDOWN
+ case _ =>
+ info(s"Skipping controlled shutdown because we are in state ${_state}.")
+ beginShutdown()
+ }
+ }
+ }
+
+ /**
+ * Enter the controlled shutdown state if we are in RUNNING state.
+ * Or, if we're not running, shut down immediately.
+ */
+ def beginControlledShutdown(): Unit = {
+ eventQueue.append(new BeginControlledShutdownEvent())
+ }
+
+ /**
+ * Start shutting down the BrokerLifecycleManager, but do not block.
+ */
+ def beginShutdown(): Unit = {
+ eventQueue.beginShutdown("beginShutdown", new ShutdownEvent())
+ }
+
+ /**
+ * Shut down the BrokerLifecycleManager and block until all threads are joined.
+ */
+ def close(): Unit = {
+ beginShutdown()
+ eventQueue.close()
+ }
+
+ class SetReadyToUnfenceEvent() extends EventQueue.Event {
+ override def run(): Unit = {
+ readyToUnfence = true
+ scheduleNextCommunicationImmediately()
+ }
+ }
+
+ class StartupEvent(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]) extends EventQueue.Event {
+ override def run(): Unit = {
+ _highestMetadataOffsetProvider = highestMetadataOffsetProvider
+ _channelManager = channelManager
+ _channelManager.start()
+ _state = BrokerState.STARTING
+ _clusterId = clusterId
+ _advertisedListeners = advertisedListeners
+ _supportedFeatures = supportedFeatures
Review comment:
Should we be copying from this map rather than assigning to it? Do we want things outside of this class to mutate the data in this map after start is called?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573953538
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
Review comment:
20% jitter is kind of a lot for the heartbeat, I think. 20% means we could be heartbeating every 2.4 s rather than 2 s.
Thinking about it more I'd be more comfortable with 2%. The point of jitter is just to avoid having a lot of heartbeats come in all at once, and I think 2% jitter is more than sufficient to accomplish that...
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573994221
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ _state match {
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ info(s"Attempted to enter controlled shutdown state, but we are already in " +
+ s"that state.")
+ case BrokerState.RUNNING =>
+ info(s"Beginning controlled shutdown.")
+ _state = BrokerState.PENDING_CONTROLLED_SHUTDOWN
+ case _ =>
+ info(s"Skipping controlled shutdown because we are in state ${_state}.")
+ beginShutdown()
+ }
+ }
+ }
+
+ /**
+ * Enter the controlled shutdown state if we are in RUNNING state.
+ * Or, if we're not running, shut down immediately.
+ */
+ def beginControlledShutdown(): Unit = {
+ eventQueue.append(new BeginControlledShutdownEvent())
+ }
+
+ /**
+ * Start shutting down the BrokerLifecycleManager, but do not block.
+ */
+ def beginShutdown(): Unit = {
+ eventQueue.beginShutdown("beginShutdown", new ShutdownEvent())
+ }
+
+ /**
+ * Shut down the BrokerLifecycleManager and block until all threads are joined.
+ */
+ def close(): Unit = {
+ beginShutdown()
+ eventQueue.close()
+ }
+
+ class SetReadyToUnfenceEvent() extends EventQueue.Event {
+ override def run(): Unit = {
+ readyToUnfence = true
+ scheduleNextCommunicationImmediately()
+ }
+ }
+
+ class StartupEvent(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]) extends EventQueue.Event {
+ override def run(): Unit = {
+ _highestMetadataOffsetProvider = highestMetadataOffsetProvider
+ _channelManager = channelManager
+ _channelManager.start()
Review comment:
I guess we could have the responsibility for closing the channel manager lie with the external callers. Would that be better here? It would be clearer, I guess.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573981156
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ _state match {
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ info(s"Attempted to enter controlled shutdown state, but we are already in " +
+ s"that state.")
+ case BrokerState.RUNNING =>
+ info(s"Beginning controlled shutdown.")
+ _state = BrokerState.PENDING_CONTROLLED_SHUTDOWN
+ case _ =>
+ info(s"Skipping controlled shutdown because we are in state ${_state}.")
+ beginShutdown()
+ }
+ }
+ }
+
+ /**
+ * Enter the controlled shutdown state if we are in RUNNING state.
+ * Or, if we're not running, shut down immediately.
+ */
+ def beginControlledShutdown(): Unit = {
+ eventQueue.append(new BeginControlledShutdownEvent())
+ }
+
+ /**
+ * Start shutting down the BrokerLifecycleManager, but do not block.
+ */
+ def beginShutdown(): Unit = {
+ eventQueue.beginShutdown("beginShutdown", new ShutdownEvent())
+ }
+
+ /**
+ * Shut down the BrokerLifecycleManager and block until all threads are joined.
+ */
+ def close(): Unit = {
+ beginShutdown()
+ eventQueue.close()
+ }
+
+ class SetReadyToUnfenceEvent() extends EventQueue.Event {
+ override def run(): Unit = {
+ readyToUnfence = true
+ scheduleNextCommunicationImmediately()
+ }
+ }
+
+ class StartupEvent(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]) extends EventQueue.Event {
+ override def run(): Unit = {
+ _highestMetadataOffsetProvider = highestMetadataOffsetProvider
+ _channelManager = channelManager
+ _channelManager.start()
+ _state = BrokerState.STARTING
+ _clusterId = clusterId
+ _advertisedListeners = advertisedListeners
+ _supportedFeatures = supportedFeatures
+ eventQueue.scheduleDeferred("initialRegistrationTimeout",
+ new DeadlineFunction(time.nanoseconds() + initialTimeoutNs),
+ new RegistrationTimeoutEvent())
+ sendBrokerRegistration()
+ info(s"Incarnation ${incarnationId} of broker ${nodeId} in cluster ${clusterId} " +
+ "is now STARTING.")
+ }
+ }
+
+ private def sendBrokerRegistration(): Unit = {
+ val features = new BrokerRegistrationRequestData.FeatureCollection()
+ _supportedFeatures.asScala.foreach {
+ case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ setName(name).
+ setMinSupportedVersion(range.min()).
+ setMaxSupportedVersion(range.max()))
+ }
+ val data = new BrokerRegistrationRequestData().
+ setBrokerId(nodeId).
+ setClusterId(_clusterId).
+ setFeatures(features).
+ setIncarnationId(incarnationId).
+ setListeners(_advertisedListeners).
+ setRack(rack)
+ if (isTraceEnabled) {
+ trace(s"Sending broker registration ${data}")
+ }
+ _channelManager.sendRequest(new BrokerRegistrationRequest.Builder(data),
+ new BrokerRegistrationResponseHandler())
+ }
+
+ class BrokerRegistrationResponseHandler extends ControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ if (response.authenticationException() != null) {
+ error(s"Unable to register broker ${nodeId} because of an authentication exception.",
+ response.authenticationException());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.versionMismatch() != null) {
+ error(s"Unable to register broker ${nodeId} because of an API version problem.",
+ response.versionMismatch());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.responseBody() == null) {
+ warn(s"Unable to register broker ${nodeId}.")
+ scheduleNextCommunicationAfterFailure()
+ } else if (!response.responseBody().isInstanceOf[BrokerRegistrationResponse]) {
+ error(s"Unable to register broker ${nodeId} because the controller returned an " +
+ "invalid response type.")
+ scheduleNextCommunicationAfterFailure()
+ } else {
+ val message = response.responseBody().asInstanceOf[BrokerRegistrationResponse]
+ val errorCode = Errors.forCode(message.data().errorCode())
+ if (errorCode == Errors.NONE) {
+ failedAttempts = 0
+ _brokerEpoch = message.data().brokerEpoch()
+ registered = true
+ initialRegistrationSucceeded = true
+ info(s"Successfully registered broker ${nodeId} with broker epoch ${_brokerEpoch}")
+ scheduleNextCommunicationImmediately() // Immediately send a heartbeat
+ } else {
+ info(s"Unable to register broker ${nodeId} because the controller returned " +
+ s"error ${errorCode}")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+ }
+
+ override def onTimeout(): Unit = {
+ info(s"Unable to register the broker because the RPC got timed out before it could be sent.")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+
+ private def sendBrokerHeartbeat(): Unit = {
+ val metadataOffset = _highestMetadataOffsetProvider()
+ val data = new BrokerHeartbeatRequestData().
+ setBrokerEpoch(_brokerEpoch).
+ setBrokerId(nodeId).
+ setCurrentMetadataOffset(metadataOffset).
+ setWantFence(!readyToUnfence).
+ setWantShutDown(_state == BrokerState.PENDING_CONTROLLED_SHUTDOWN)
+ if (isTraceEnabled) {
+ trace(s"Sending broker heartbeat ${data}")
+ }
+ _channelManager.sendRequest(new BrokerHeartbeatRequest.Builder(data),
+ new BrokerHeartbeatResponseHandler())
+ }
+
+ class BrokerHeartbeatResponseHandler extends ControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ if (response.authenticationException() != null) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because of an " +
+ "authentication exception.", response.authenticationException());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.versionMismatch() != null) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because of an API " +
+ "version problem.", response.versionMismatch());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.responseBody() == null) {
+ warn(s"Unable to send broker heartbeat for ${nodeId}. Retrying.")
+ scheduleNextCommunicationAfterFailure()
+ } else if (!response.responseBody().isInstanceOf[BrokerHeartbeatResponse]) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because the controller " +
+ "returned an invalid response type.")
+ scheduleNextCommunicationAfterFailure()
+ } else {
+ val message = response.responseBody().asInstanceOf[BrokerHeartbeatResponse]
+ val errorCode = Errors.forCode(message.data().errorCode())
+ if (errorCode == Errors.NONE) {
+ failedAttempts = 0
+ _state match {
+ case BrokerState.STARTING =>
+ if (message.data().isCaughtUp()) {
+ info(s"The broker has caught up. Transitioning from STARTING to RECOVERY.")
+ _state = BrokerState.RECOVERY
+ initialCatchUpFuture.complete(null)
+ } else {
+ info(s"The broker is STARTING. Still waiting to catch up with cluster metadata.")
+ }
+ // Schedule the heartbeat after only 10 ms so that in the case where
+ //there is no recovery work to be done, we start up a bit quicker.
+ scheduleNextCommunication(NANOSECONDS.convert(10, MILLISECONDS))
+ case BrokerState.RECOVERY =>
+ if (!message.data().isFenced()) {
+ info(s"The broker has been unfenced. Transitioning from RECOVERY to RUNNING.")
+ _state = BrokerState.RUNNING
+ } else {
+ info(s"The broker is in RECOVERY.")
+ }
+ scheduleNextCommunicationAfterSuccess()
+ case BrokerState.RUNNING =>
+ debug(s"The broker is RUNNING. Processing heartbeat response.")
+ scheduleNextCommunicationAfterSuccess()
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ if (!message.data().shouldShutDown()) {
+ info(s"The broker is in PENDING_CONTROLLED_SHUTDOWN state, still waiting " +
+ "for the active controller.")
+ if (!gotControlledShutdownResponse) {
+ // If this is the first pending controlled shutdown response we got,
+ // schedule our next heartbeat a little bit sooner than we usually would.
+ // In the case where controlled shutdown completes quickly, this will
+ // speed things up a little bit.
+ scheduleNextCommunication(NANOSECONDS.convert(50, MILLISECONDS))
+ } else {
+ scheduleNextCommunicationAfterSuccess()
+ }
+ } else {
+ info(s"The controlled has asked us to exit controlled shutdown.")
+ beginShutdown()
+ }
+ gotControlledShutdownResponse = true
+ case BrokerState.SHUTTING_DOWN =>
+ info(s"The broker is SHUTTING_DOWN. Ignoring heartbeat response.")
+ case _ =>
+ error(s"Unexpected broker state ${_state}")
+ scheduleNextCommunicationAfterSuccess()
+ }
+ } else {
+ warn(s"Broker ${nodeId} sent a heartbeat request but received error ${errorCode}.")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+ }
+
+ override def onTimeout(): Unit = {
+ info("Unable to send a heartbeat because the RPC got timed out before it could be sent.")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+
+ private def scheduleNextCommunicationImmediately(): Unit = scheduleNextCommunication(0)
+
+ private def scheduleNextCommunicationAfterFailure(): Unit = {
+ val delayMs = resendExponentialBackoff.backoff(failedAttempts)
+ failedAttempts = failedAttempts + 1
+ scheduleNextCommunication(NANOSECONDS.convert(delayMs, MILLISECONDS))
+ }
+
+ private def scheduleNextCommunicationAfterSuccess(): Unit = {
+ scheduleNextCommunication(NANOSECONDS.convert(
+ config.brokerHeartbeatIntervalMs.longValue() , MILLISECONDS))
+ }
+
+ private def scheduleNextCommunication(intervalNs: Long): Unit = {
+ trace(s"Scheduling next communication at ${MILLISECONDS.convert(intervalNs, NANOSECONDS)} " +
+ "ms from now.")
+ val deadlineNs = time.nanoseconds() + intervalNs
+ eventQueue.scheduleDeferred("communication",
+ new DeadlineFunction(deadlineNs),
+ new CommunicationEvent())
+ }
+
+ class RegistrationTimeoutEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ if (!initialRegistrationSucceeded) {
+ error("Shutting down because we were unable to register with the controller quorum.")
+ eventQueue.beginShutdown("registrationTimeout", new ShutdownEvent())
+ }
+ }
+ }
+
+ class CommunicationEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ if (registered) {
+ sendBrokerHeartbeat()
+ } else {
+ sendBrokerRegistration()
+ }
+ }
+ }
+
+ class ShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ info(s"Transitioning from ${_state} to ${BrokerState.SHUTTING_DOWN}.")
+ _state = BrokerState.SHUTTING_DOWN
+ controlledShutdownFuture.complete(null)
+ initialCatchUpFuture.cancel(false)
+ _channelManager.shutdown()
Review comment:
good point. we should check.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573984736
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ _state match {
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ info(s"Attempted to enter controlled shutdown state, but we are already in " +
+ s"that state.")
+ case BrokerState.RUNNING =>
+ info(s"Beginning controlled shutdown.")
+ _state = BrokerState.PENDING_CONTROLLED_SHUTDOWN
+ case _ =>
+ info(s"Skipping controlled shutdown because we are in state ${_state}.")
+ beginShutdown()
+ }
+ }
+ }
+
+ /**
+ * Enter the controlled shutdown state if we are in RUNNING state.
+ * Or, if we're not running, shut down immediately.
+ */
+ def beginControlledShutdown(): Unit = {
+ eventQueue.append(new BeginControlledShutdownEvent())
+ }
+
+ /**
+ * Start shutting down the BrokerLifecycleManager, but do not block.
+ */
+ def beginShutdown(): Unit = {
+ eventQueue.beginShutdown("beginShutdown", new ShutdownEvent())
+ }
+
+ /**
+ * Shut down the BrokerLifecycleManager and block until all threads are joined.
+ */
+ def close(): Unit = {
+ beginShutdown()
+ eventQueue.close()
+ }
+
+ class SetReadyToUnfenceEvent() extends EventQueue.Event {
+ override def run(): Unit = {
+ readyToUnfence = true
+ scheduleNextCommunicationImmediately()
+ }
+ }
+
+ class StartupEvent(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]) extends EventQueue.Event {
+ override def run(): Unit = {
+ _highestMetadataOffsetProvider = highestMetadataOffsetProvider
+ _channelManager = channelManager
+ _channelManager.start()
+ _state = BrokerState.STARTING
+ _clusterId = clusterId
+ _advertisedListeners = advertisedListeners
+ _supportedFeatures = supportedFeatures
+ eventQueue.scheduleDeferred("initialRegistrationTimeout",
+ new DeadlineFunction(time.nanoseconds() + initialTimeoutNs),
+ new RegistrationTimeoutEvent())
+ sendBrokerRegistration()
+ info(s"Incarnation ${incarnationId} of broker ${nodeId} in cluster ${clusterId} " +
+ "is now STARTING.")
+ }
+ }
+
+ private def sendBrokerRegistration(): Unit = {
+ val features = new BrokerRegistrationRequestData.FeatureCollection()
+ _supportedFeatures.asScala.foreach {
+ case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ setName(name).
+ setMinSupportedVersion(range.min()).
+ setMaxSupportedVersion(range.max()))
+ }
+ val data = new BrokerRegistrationRequestData().
+ setBrokerId(nodeId).
+ setClusterId(_clusterId).
+ setFeatures(features).
+ setIncarnationId(incarnationId).
+ setListeners(_advertisedListeners).
+ setRack(rack)
+ if (isTraceEnabled) {
+ trace(s"Sending broker registration ${data}")
+ }
+ _channelManager.sendRequest(new BrokerRegistrationRequest.Builder(data),
+ new BrokerRegistrationResponseHandler())
+ }
+
+ class BrokerRegistrationResponseHandler extends ControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ if (response.authenticationException() != null) {
+ error(s"Unable to register broker ${nodeId} because of an authentication exception.",
+ response.authenticationException());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.versionMismatch() != null) {
+ error(s"Unable to register broker ${nodeId} because of an API version problem.",
+ response.versionMismatch());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.responseBody() == null) {
+ warn(s"Unable to register broker ${nodeId}.")
+ scheduleNextCommunicationAfterFailure()
+ } else if (!response.responseBody().isInstanceOf[BrokerRegistrationResponse]) {
+ error(s"Unable to register broker ${nodeId} because the controller returned an " +
+ "invalid response type.")
+ scheduleNextCommunicationAfterFailure()
+ } else {
+ val message = response.responseBody().asInstanceOf[BrokerRegistrationResponse]
+ val errorCode = Errors.forCode(message.data().errorCode())
+ if (errorCode == Errors.NONE) {
+ failedAttempts = 0
+ _brokerEpoch = message.data().brokerEpoch()
+ registered = true
+ initialRegistrationSucceeded = true
+ info(s"Successfully registered broker ${nodeId} with broker epoch ${_brokerEpoch}")
+ scheduleNextCommunicationImmediately() // Immediately send a heartbeat
+ } else {
+ info(s"Unable to register broker ${nodeId} because the controller returned " +
+ s"error ${errorCode}")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+ }
+
+ override def onTimeout(): Unit = {
+ info(s"Unable to register the broker because the RPC got timed out before it could be sent.")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+
+ private def sendBrokerHeartbeat(): Unit = {
+ val metadataOffset = _highestMetadataOffsetProvider()
+ val data = new BrokerHeartbeatRequestData().
+ setBrokerEpoch(_brokerEpoch).
+ setBrokerId(nodeId).
+ setCurrentMetadataOffset(metadataOffset).
+ setWantFence(!readyToUnfence).
+ setWantShutDown(_state == BrokerState.PENDING_CONTROLLED_SHUTDOWN)
+ if (isTraceEnabled) {
+ trace(s"Sending broker heartbeat ${data}")
+ }
+ _channelManager.sendRequest(new BrokerHeartbeatRequest.Builder(data),
+ new BrokerHeartbeatResponseHandler())
+ }
+
+ class BrokerHeartbeatResponseHandler extends ControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ if (response.authenticationException() != null) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because of an " +
+ "authentication exception.", response.authenticationException());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.versionMismatch() != null) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because of an API " +
+ "version problem.", response.versionMismatch());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.responseBody() == null) {
+ warn(s"Unable to send broker heartbeat for ${nodeId}. Retrying.")
+ scheduleNextCommunicationAfterFailure()
+ } else if (!response.responseBody().isInstanceOf[BrokerHeartbeatResponse]) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because the controller " +
+ "returned an invalid response type.")
+ scheduleNextCommunicationAfterFailure()
+ } else {
+ val message = response.responseBody().asInstanceOf[BrokerHeartbeatResponse]
+ val errorCode = Errors.forCode(message.data().errorCode())
+ if (errorCode == Errors.NONE) {
+ failedAttempts = 0
+ _state match {
+ case BrokerState.STARTING =>
+ if (message.data().isCaughtUp()) {
+ info(s"The broker has caught up. Transitioning from STARTING to RECOVERY.")
+ _state = BrokerState.RECOVERY
+ initialCatchUpFuture.complete(null)
+ } else {
+ info(s"The broker is STARTING. Still waiting to catch up with cluster metadata.")
+ }
+ // Schedule the heartbeat after only 10 ms so that in the case where
+ //there is no recovery work to be done, we start up a bit quicker.
+ scheduleNextCommunication(NANOSECONDS.convert(10, MILLISECONDS))
+ case BrokerState.RECOVERY =>
+ if (!message.data().isFenced()) {
+ info(s"The broker has been unfenced. Transitioning from RECOVERY to RUNNING.")
+ _state = BrokerState.RUNNING
+ } else {
+ info(s"The broker is in RECOVERY.")
+ }
+ scheduleNextCommunicationAfterSuccess()
+ case BrokerState.RUNNING =>
+ debug(s"The broker is RUNNING. Processing heartbeat response.")
+ scheduleNextCommunicationAfterSuccess()
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ if (!message.data().shouldShutDown()) {
+ info(s"The broker is in PENDING_CONTROLLED_SHUTDOWN state, still waiting " +
+ "for the active controller.")
+ if (!gotControlledShutdownResponse) {
+ // If this is the first pending controlled shutdown response we got,
+ // schedule our next heartbeat a little bit sooner than we usually would.
+ // In the case where controlled shutdown completes quickly, this will
+ // speed things up a little bit.
+ scheduleNextCommunication(NANOSECONDS.convert(50, MILLISECONDS))
Review comment:
Hmm. I'm not sure. Even if the heartbeat interval was 1 second or 3 seconds (just for example) I don't think we'd want this interval to change.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe merged pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe merged pull request #10095:
URL: https://github.com/apache/kafka/pull/10095
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573992730
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ _state match {
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ info(s"Attempted to enter controlled shutdown state, but we are already in " +
+ s"that state.")
+ case BrokerState.RUNNING =>
+ info(s"Beginning controlled shutdown.")
+ _state = BrokerState.PENDING_CONTROLLED_SHUTDOWN
+ case _ =>
+ info(s"Skipping controlled shutdown because we are in state ${_state}.")
+ beginShutdown()
+ }
+ }
+ }
+
+ /**
+ * Enter the controlled shutdown state if we are in RUNNING state.
+ * Or, if we're not running, shut down immediately.
+ */
+ def beginControlledShutdown(): Unit = {
+ eventQueue.append(new BeginControlledShutdownEvent())
+ }
+
+ /**
+ * Start shutting down the BrokerLifecycleManager, but do not block.
+ */
+ def beginShutdown(): Unit = {
+ eventQueue.beginShutdown("beginShutdown", new ShutdownEvent())
+ }
+
+ /**
+ * Shut down the BrokerLifecycleManager and block until all threads are joined.
+ */
+ def close(): Unit = {
+ beginShutdown()
+ eventQueue.close()
+ }
+
+ class SetReadyToUnfenceEvent() extends EventQueue.Event {
+ override def run(): Unit = {
+ readyToUnfence = true
+ scheduleNextCommunicationImmediately()
+ }
+ }
+
+ class StartupEvent(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]) extends EventQueue.Event {
+ override def run(): Unit = {
+ _highestMetadataOffsetProvider = highestMetadataOffsetProvider
+ _channelManager = channelManager
+ _channelManager.start()
+ _state = BrokerState.STARTING
+ _clusterId = clusterId
+ _advertisedListeners = advertisedListeners
+ _supportedFeatures = supportedFeatures
Review comment:
I'll make a defensive copy
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573972180
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
Review comment:
supportedFeatures is a static final map which basically explains what features the software supports. So the intention is for it to be defined in common code which happens to be written in Java (not included in this PR). It seemed simpler not to translate the map into scala, although I guess it's not too difficult either way.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573980461
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
Review comment:
I guess you already know this but unlike Java, Scala doesn't support an accessor function having the same name as an instance variable. So I kind of got into the habit of prefixing private variables with an underscore. In this case I guess it's not needed since we don't have a clusterId accessor (yet?) but maybe it's easier just to be consistent.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573980461
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
Review comment:
Unlike Java, Scala doesn't support an accessor function having the same name as an instance variable. So we kind of got into the habit of prefixing private variables with an underscore. In this case I guess it's not needed since we don't have a clusterId accessor (yet?) but maybe it's easier just to be consistent.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573839484
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
Review comment:
Nit: Isn't `TimeUnit.MILLISECONDS.toNanos(config.initialRegistrationTimeoutMs)` more readable?
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
Review comment:
We seem to use `0.2` as jitter elsewhere. Any reason to deviate? Maybe we should have an `ExponentialBackoff` static factory method that only takes the initial interval and the max interval. It seems that the other two parameters can be the same most of the time.
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
Review comment:
Why are we using a Java collection?
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
Review comment:
Worth adding a comment explaining the thread safety aspects of this class and also its general purpose. I see a few volatile fields, but `failedAttempts` is not volatile (for example).
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
Review comment:
Are these events meant to be public?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573972923
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
Review comment:
that's a good point-- these can all be private. Will fix.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573975606
##########
File path: core/src/main/scala/kafka/server/KafkaConfig.scala
##########
@@ -1497,6 +1509,9 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
var brokerId: Int = getInt(KafkaConfig.BrokerIdProp)
val nodeId: Int = getInt(KafkaConfig.NodeIdProp)
val processRoles: Set[ProcessRole] = parseProcessRoles()
+ val initialRegistrationTimeoutMs = getInt(KafkaConfig.InitialBrokerRegistrationTimeoutMs)
+ val brokerHeartbeatIntervalMs = getInt(KafkaConfig.BrokerHeartbeatIntervalMsProp)
+ val brokerSessionTimeoutMs = getInt(KafkaConfig.BrokerSessionTimeoutMsProp)
Review comment:
Good point. I added the explicit type of `Int`.
I don't think they need to be nullable... they have reasonable defaults as usual in KafkaConfig. And they will be ignored altogether when in ZK mode, of course.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573992299
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ _state match {
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ info(s"Attempted to enter controlled shutdown state, but we are already in " +
+ s"that state.")
+ case BrokerState.RUNNING =>
+ info(s"Beginning controlled shutdown.")
+ _state = BrokerState.PENDING_CONTROLLED_SHUTDOWN
+ case _ =>
+ info(s"Skipping controlled shutdown because we are in state ${_state}.")
+ beginShutdown()
+ }
+ }
+ }
+
+ /**
+ * Enter the controlled shutdown state if we are in RUNNING state.
+ * Or, if we're not running, shut down immediately.
+ */
+ def beginControlledShutdown(): Unit = {
+ eventQueue.append(new BeginControlledShutdownEvent())
+ }
+
+ /**
+ * Start shutting down the BrokerLifecycleManager, but do not block.
+ */
+ def beginShutdown(): Unit = {
+ eventQueue.beginShutdown("beginShutdown", new ShutdownEvent())
+ }
+
+ /**
+ * Shut down the BrokerLifecycleManager and block until all threads are joined.
+ */
+ def close(): Unit = {
+ beginShutdown()
+ eventQueue.close()
+ }
+
+ class SetReadyToUnfenceEvent() extends EventQueue.Event {
+ override def run(): Unit = {
+ readyToUnfence = true
+ scheduleNextCommunicationImmediately()
+ }
+ }
+
+ class StartupEvent(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]) extends EventQueue.Event {
+ override def run(): Unit = {
+ _highestMetadataOffsetProvider = highestMetadataOffsetProvider
+ _channelManager = channelManager
+ _channelManager.start()
Review comment:
We do it this way to make unit tests easier, since we need to supply a mock channel manager at times
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573975606
##########
File path: core/src/main/scala/kafka/server/KafkaConfig.scala
##########
@@ -1497,6 +1509,9 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
var brokerId: Int = getInt(KafkaConfig.BrokerIdProp)
val nodeId: Int = getInt(KafkaConfig.NodeIdProp)
val processRoles: Set[ProcessRole] = parseProcessRoles()
+ val initialRegistrationTimeoutMs = getInt(KafkaConfig.InitialBrokerRegistrationTimeoutMs)
+ val brokerHeartbeatIntervalMs = getInt(KafkaConfig.BrokerHeartbeatIntervalMsProp)
+ val brokerSessionTimeoutMs = getInt(KafkaConfig.BrokerSessionTimeoutMsProp)
Review comment:
I added the explicit type.
I don't think they need to be nullable... they have reasonable defaults as usual in KafkaConfig. And they will be ignored altogether when in ZK mode, of course.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573983344
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ _state match {
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ info(s"Attempted to enter controlled shutdown state, but we are already in " +
+ s"that state.")
+ case BrokerState.RUNNING =>
+ info(s"Beginning controlled shutdown.")
+ _state = BrokerState.PENDING_CONTROLLED_SHUTDOWN
+ case _ =>
+ info(s"Skipping controlled shutdown because we are in state ${_state}.")
+ beginShutdown()
+ }
+ }
+ }
+
+ /**
+ * Enter the controlled shutdown state if we are in RUNNING state.
+ * Or, if we're not running, shut down immediately.
+ */
+ def beginControlledShutdown(): Unit = {
+ eventQueue.append(new BeginControlledShutdownEvent())
+ }
+
+ /**
+ * Start shutting down the BrokerLifecycleManager, but do not block.
+ */
+ def beginShutdown(): Unit = {
+ eventQueue.beginShutdown("beginShutdown", new ShutdownEvent())
+ }
+
+ /**
+ * Shut down the BrokerLifecycleManager and block until all threads are joined.
+ */
+ def close(): Unit = {
+ beginShutdown()
+ eventQueue.close()
+ }
+
+ class SetReadyToUnfenceEvent() extends EventQueue.Event {
+ override def run(): Unit = {
+ readyToUnfence = true
+ scheduleNextCommunicationImmediately()
+ }
+ }
+
+ class StartupEvent(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]) extends EventQueue.Event {
+ override def run(): Unit = {
+ _highestMetadataOffsetProvider = highestMetadataOffsetProvider
+ _channelManager = channelManager
+ _channelManager.start()
+ _state = BrokerState.STARTING
+ _clusterId = clusterId
+ _advertisedListeners = advertisedListeners
+ _supportedFeatures = supportedFeatures
+ eventQueue.scheduleDeferred("initialRegistrationTimeout",
+ new DeadlineFunction(time.nanoseconds() + initialTimeoutNs),
+ new RegistrationTimeoutEvent())
+ sendBrokerRegistration()
+ info(s"Incarnation ${incarnationId} of broker ${nodeId} in cluster ${clusterId} " +
+ "is now STARTING.")
+ }
+ }
+
+ private def sendBrokerRegistration(): Unit = {
+ val features = new BrokerRegistrationRequestData.FeatureCollection()
+ _supportedFeatures.asScala.foreach {
+ case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ setName(name).
+ setMinSupportedVersion(range.min()).
+ setMaxSupportedVersion(range.max()))
+ }
+ val data = new BrokerRegistrationRequestData().
+ setBrokerId(nodeId).
+ setClusterId(_clusterId).
+ setFeatures(features).
+ setIncarnationId(incarnationId).
+ setListeners(_advertisedListeners).
+ setRack(rack)
+ if (isTraceEnabled) {
+ trace(s"Sending broker registration ${data}")
+ }
+ _channelManager.sendRequest(new BrokerRegistrationRequest.Builder(data),
+ new BrokerRegistrationResponseHandler())
+ }
+
+ class BrokerRegistrationResponseHandler extends ControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ if (response.authenticationException() != null) {
+ error(s"Unable to register broker ${nodeId} because of an authentication exception.",
+ response.authenticationException());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.versionMismatch() != null) {
+ error(s"Unable to register broker ${nodeId} because of an API version problem.",
+ response.versionMismatch());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.responseBody() == null) {
+ warn(s"Unable to register broker ${nodeId}.")
+ scheduleNextCommunicationAfterFailure()
+ } else if (!response.responseBody().isInstanceOf[BrokerRegistrationResponse]) {
+ error(s"Unable to register broker ${nodeId} because the controller returned an " +
+ "invalid response type.")
+ scheduleNextCommunicationAfterFailure()
+ } else {
+ val message = response.responseBody().asInstanceOf[BrokerRegistrationResponse]
+ val errorCode = Errors.forCode(message.data().errorCode())
+ if (errorCode == Errors.NONE) {
+ failedAttempts = 0
+ _brokerEpoch = message.data().brokerEpoch()
+ registered = true
+ initialRegistrationSucceeded = true
+ info(s"Successfully registered broker ${nodeId} with broker epoch ${_brokerEpoch}")
+ scheduleNextCommunicationImmediately() // Immediately send a heartbeat
+ } else {
+ info(s"Unable to register broker ${nodeId} because the controller returned " +
+ s"error ${errorCode}")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+ }
+
+ override def onTimeout(): Unit = {
+ info(s"Unable to register the broker because the RPC got timed out before it could be sent.")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+
+ private def sendBrokerHeartbeat(): Unit = {
+ val metadataOffset = _highestMetadataOffsetProvider()
+ val data = new BrokerHeartbeatRequestData().
+ setBrokerEpoch(_brokerEpoch).
+ setBrokerId(nodeId).
+ setCurrentMetadataOffset(metadataOffset).
+ setWantFence(!readyToUnfence).
+ setWantShutDown(_state == BrokerState.PENDING_CONTROLLED_SHUTDOWN)
+ if (isTraceEnabled) {
+ trace(s"Sending broker heartbeat ${data}")
+ }
+ _channelManager.sendRequest(new BrokerHeartbeatRequest.Builder(data),
+ new BrokerHeartbeatResponseHandler())
+ }
+
+ class BrokerHeartbeatResponseHandler extends ControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ if (response.authenticationException() != null) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because of an " +
+ "authentication exception.", response.authenticationException());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.versionMismatch() != null) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because of an API " +
+ "version problem.", response.versionMismatch());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.responseBody() == null) {
+ warn(s"Unable to send broker heartbeat for ${nodeId}. Retrying.")
+ scheduleNextCommunicationAfterFailure()
+ } else if (!response.responseBody().isInstanceOf[BrokerHeartbeatResponse]) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because the controller " +
+ "returned an invalid response type.")
+ scheduleNextCommunicationAfterFailure()
+ } else {
+ val message = response.responseBody().asInstanceOf[BrokerHeartbeatResponse]
+ val errorCode = Errors.forCode(message.data().errorCode())
+ if (errorCode == Errors.NONE) {
+ failedAttempts = 0
+ _state match {
+ case BrokerState.STARTING =>
+ if (message.data().isCaughtUp()) {
+ info(s"The broker has caught up. Transitioning from STARTING to RECOVERY.")
+ _state = BrokerState.RECOVERY
+ initialCatchUpFuture.complete(null)
+ } else {
+ info(s"The broker is STARTING. Still waiting to catch up with cluster metadata.")
+ }
+ // Schedule the heartbeat after only 10 ms so that in the case where
+ //there is no recovery work to be done, we start up a bit quicker.
+ scheduleNextCommunication(NANOSECONDS.convert(10, MILLISECONDS))
Review comment:
If there is no log recovery work to be done, a SetReadyToUnfenceEvent will probably be handled in this time. Basically BrokerServer will just continue on and tell us it's ready to go.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mumrah commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
mumrah commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r574644427
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ _state match {
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ info(s"Attempted to enter controlled shutdown state, but we are already in " +
+ s"that state.")
+ case BrokerState.RUNNING =>
+ info(s"Beginning controlled shutdown.")
+ _state = BrokerState.PENDING_CONTROLLED_SHUTDOWN
+ case _ =>
+ info(s"Skipping controlled shutdown because we are in state ${_state}.")
+ beginShutdown()
+ }
+ }
+ }
+
+ /**
+ * Enter the controlled shutdown state if we are in RUNNING state.
+ * Or, if we're not running, shut down immediately.
+ */
+ def beginControlledShutdown(): Unit = {
+ eventQueue.append(new BeginControlledShutdownEvent())
+ }
+
+ /**
+ * Start shutting down the BrokerLifecycleManager, but do not block.
+ */
+ def beginShutdown(): Unit = {
+ eventQueue.beginShutdown("beginShutdown", new ShutdownEvent())
+ }
+
+ /**
+ * Shut down the BrokerLifecycleManager and block until all threads are joined.
+ */
+ def close(): Unit = {
+ beginShutdown()
+ eventQueue.close()
+ }
+
+ class SetReadyToUnfenceEvent() extends EventQueue.Event {
+ override def run(): Unit = {
+ readyToUnfence = true
+ scheduleNextCommunicationImmediately()
+ }
+ }
+
+ class StartupEvent(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]) extends EventQueue.Event {
+ override def run(): Unit = {
+ _highestMetadataOffsetProvider = highestMetadataOffsetProvider
+ _channelManager = channelManager
+ _channelManager.start()
+ _state = BrokerState.STARTING
+ _clusterId = clusterId
+ _advertisedListeners = advertisedListeners
+ _supportedFeatures = supportedFeatures
+ eventQueue.scheduleDeferred("initialRegistrationTimeout",
+ new DeadlineFunction(time.nanoseconds() + initialTimeoutNs),
+ new RegistrationTimeoutEvent())
+ sendBrokerRegistration()
+ info(s"Incarnation ${incarnationId} of broker ${nodeId} in cluster ${clusterId} " +
+ "is now STARTING.")
+ }
+ }
+
+ private def sendBrokerRegistration(): Unit = {
+ val features = new BrokerRegistrationRequestData.FeatureCollection()
+ _supportedFeatures.asScala.foreach {
+ case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ setName(name).
+ setMinSupportedVersion(range.min()).
+ setMaxSupportedVersion(range.max()))
+ }
+ val data = new BrokerRegistrationRequestData().
+ setBrokerId(nodeId).
+ setClusterId(_clusterId).
+ setFeatures(features).
+ setIncarnationId(incarnationId).
+ setListeners(_advertisedListeners).
+ setRack(rack)
+ if (isTraceEnabled) {
+ trace(s"Sending broker registration ${data}")
+ }
+ _channelManager.sendRequest(new BrokerRegistrationRequest.Builder(data),
+ new BrokerRegistrationResponseHandler())
+ }
+
+ class BrokerRegistrationResponseHandler extends ControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ if (response.authenticationException() != null) {
+ error(s"Unable to register broker ${nodeId} because of an authentication exception.",
+ response.authenticationException());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.versionMismatch() != null) {
+ error(s"Unable to register broker ${nodeId} because of an API version problem.",
+ response.versionMismatch());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.responseBody() == null) {
+ warn(s"Unable to register broker ${nodeId}.")
+ scheduleNextCommunicationAfterFailure()
+ } else if (!response.responseBody().isInstanceOf[BrokerRegistrationResponse]) {
+ error(s"Unable to register broker ${nodeId} because the controller returned an " +
+ "invalid response type.")
+ scheduleNextCommunicationAfterFailure()
+ } else {
+ val message = response.responseBody().asInstanceOf[BrokerRegistrationResponse]
+ val errorCode = Errors.forCode(message.data().errorCode())
+ if (errorCode == Errors.NONE) {
+ failedAttempts = 0
+ _brokerEpoch = message.data().brokerEpoch()
+ registered = true
+ initialRegistrationSucceeded = true
+ info(s"Successfully registered broker ${nodeId} with broker epoch ${_brokerEpoch}")
+ scheduleNextCommunicationImmediately() // Immediately send a heartbeat
+ } else {
+ info(s"Unable to register broker ${nodeId} because the controller returned " +
+ s"error ${errorCode}")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+ }
+
+ override def onTimeout(): Unit = {
+ info(s"Unable to register the broker because the RPC got timed out before it could be sent.")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+
+ private def sendBrokerHeartbeat(): Unit = {
+ val metadataOffset = _highestMetadataOffsetProvider()
+ val data = new BrokerHeartbeatRequestData().
+ setBrokerEpoch(_brokerEpoch).
+ setBrokerId(nodeId).
+ setCurrentMetadataOffset(metadataOffset).
+ setWantFence(!readyToUnfence).
+ setWantShutDown(_state == BrokerState.PENDING_CONTROLLED_SHUTDOWN)
+ if (isTraceEnabled) {
+ trace(s"Sending broker heartbeat ${data}")
+ }
+ _channelManager.sendRequest(new BrokerHeartbeatRequest.Builder(data),
+ new BrokerHeartbeatResponseHandler())
+ }
+
+ class BrokerHeartbeatResponseHandler extends ControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ if (response.authenticationException() != null) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because of an " +
+ "authentication exception.", response.authenticationException());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.versionMismatch() != null) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because of an API " +
+ "version problem.", response.versionMismatch());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.responseBody() == null) {
+ warn(s"Unable to send broker heartbeat for ${nodeId}. Retrying.")
+ scheduleNextCommunicationAfterFailure()
+ } else if (!response.responseBody().isInstanceOf[BrokerHeartbeatResponse]) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because the controller " +
+ "returned an invalid response type.")
+ scheduleNextCommunicationAfterFailure()
+ } else {
+ val message = response.responseBody().asInstanceOf[BrokerHeartbeatResponse]
+ val errorCode = Errors.forCode(message.data().errorCode())
+ if (errorCode == Errors.NONE) {
+ failedAttempts = 0
+ _state match {
+ case BrokerState.STARTING =>
+ if (message.data().isCaughtUp()) {
+ info(s"The broker has caught up. Transitioning from STARTING to RECOVERY.")
+ _state = BrokerState.RECOVERY
+ initialCatchUpFuture.complete(null)
+ } else {
+ info(s"The broker is STARTING. Still waiting to catch up with cluster metadata.")
+ }
+ // Schedule the heartbeat after only 10 ms so that in the case where
+ //there is no recovery work to be done, we start up a bit quicker.
+ scheduleNextCommunication(NANOSECONDS.convert(10, MILLISECONDS))
+ case BrokerState.RECOVERY =>
+ if (!message.data().isFenced()) {
+ info(s"The broker has been unfenced. Transitioning from RECOVERY to RUNNING.")
+ _state = BrokerState.RUNNING
+ } else {
+ info(s"The broker is in RECOVERY.")
+ }
+ scheduleNextCommunicationAfterSuccess()
+ case BrokerState.RUNNING =>
+ debug(s"The broker is RUNNING. Processing heartbeat response.")
+ scheduleNextCommunicationAfterSuccess()
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ if (!message.data().shouldShutDown()) {
+ info(s"The broker is in PENDING_CONTROLLED_SHUTDOWN state, still waiting " +
+ "for the active controller.")
+ if (!gotControlledShutdownResponse) {
+ // If this is the first pending controlled shutdown response we got,
+ // schedule our next heartbeat a little bit sooner than we usually would.
+ // In the case where controlled shutdown completes quickly, this will
+ // speed things up a little bit.
+ scheduleNextCommunication(NANOSECONDS.convert(50, MILLISECONDS))
Review comment:
Yea, fair enough. Maybe we can at least consolidate on one magic "short" time and make it a constant? Now we have 10ms and 50ms hard coded.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573980461
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
Review comment:
Unlike Java, Scala doesn't support an accessor function having the same name as an instance variable. So I kind of got into the habit of prefixing private variables with an underscore. In this case I guess it's not needed since we don't have a clusterId accessor (yet?) but maybe it's easier just to be consistent.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#issuecomment-776924564
Thanks for the reviews, @mumrah !
> One thing I don't love is how BrokerLifecycleManager can be instantiated but some of the members are not initialized until start is called. Is there no way we can defer initialization until all the dependencies of this class are available?
I think the next PR in the series makes things clearer. The `BrokerServer` uses `BrokerLifecycleManager` to manage the broker state at all times, not just after the broker has been started or before it has been stopped. Although there is a little bit of extra complexity in the lifecycle manager to support this usage, it makes the surrounding code much simpler because we can rely on this component to always have the state. Therefore we don't need more mutable state and if statements in `BrokerServer`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573862043
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
Review comment:
nit: append " and failed" or `s/tried/failed/`
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
Review comment:
nit: `s/we this/this/`
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
Review comment:
I wonder if it might be helpful to assign each `var` into one of two buckets: stuff that can only be written to from the event queue thread but that can be read from any thread (must be `@volatile`), and stuff that is only used from within the event queue (doesn't need to be). At a minimum put these into different commented sections, but maybe even create a single container object for the `@volatile` ones:
```
private case class EventQueueThreadOwnedVars(@volatile var _brokerEpoch: Long = -1L,
@volatile var _state: BrokerState = BrokerState.NOT_RUNNING)
val eventQueueThreadOwnedVars = EventQueueThreadOwnedVars()
```
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
Review comment:
Make an `Option[String]`? No need to propagate the unfortunate fact that rack can be null elsewhere?
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ _state match {
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ info(s"Attempted to enter controlled shutdown state, but we are already in " +
Review comment:
nit: `s/controlled/pending controlled/`
##########
File path: core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala
##########
@@ -0,0 +1,217 @@
+/**
+ * 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.server
+
+import java.util.{Collections, Properties}
+import java.util.concurrent.atomic.{AtomicLong, AtomicReference}
+
+import kafka.utils.{MockTime, TestUtils}
+import org.apache.kafka.clients.{Metadata, MockClient, NodeApiVersions}
+import org.apache.kafka.common.{Node, Uuid}
+import org.apache.kafka.common.internals.ClusterResourceListeners
+import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.{Listener, ListenerCollection}
+import org.apache.kafka.common.message.{BrokerHeartbeatResponseData, BrokerRegistrationResponseData}
+import org.apache.kafka.common.network.ListenerName
+import org.apache.kafka.common.protocol.ApiKeys.{BROKER_HEARTBEAT, BROKER_REGISTRATION}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatResponse, BrokerRegistrationResponse}
+import org.apache.kafka.common.security.auth.SecurityProtocol
+import org.apache.kafka.common.utils.LogContext
+import org.apache.kafka.metadata.BrokerState
+import org.junit.jupiter.api.{Assertions, Test, Timeout}
+
+import scala.jdk.CollectionConverters._
+
+
+@Timeout(value = 12)
+class BrokerLifecycleManagerTest {
+ def configProperties = {
+ val properties = new Properties()
+ properties.setProperty(KafkaConfig.LogDirsProp, "/tmp/foo")
+ properties.setProperty(KafkaConfig.ProcessRolesProp, "broker")
+ properties.setProperty(KafkaConfig.NodeIdProp, "1")
+ properties.setProperty(KafkaConfig.InitialBrokerRegistrationTimeoutMs, "300000")
+ properties
+ }
+
+ class SimpleControllerNodeProvider extends ControllerNodeProvider {
+ val node = new AtomicReference[Node](null)
+
+ override def get(): Option[Node] = Option(node.get())
+
+ override def listenerName: ListenerName = new ListenerName("PLAINTEXT")
+
+ override def securityProtocol: SecurityProtocol = SecurityProtocol.PLAINTEXT;
+ }
+
+ class BrokerLifecycleManagerTestContext(properties: Properties) {
+ val config = new KafkaConfig(properties)
+ val time = new MockTime(1, 1)
+ val highestMetadataOffset = new AtomicLong(0)
+ val metadata = new Metadata(1000, 1000, new LogContext(), new ClusterResourceListeners())
+ val mockClient = new MockClient(time, metadata)
+ val controllerNodeProvider = new SimpleControllerNodeProvider()
+ val nodeApiVersions = new NodeApiVersions(Seq(BROKER_REGISTRATION, BROKER_HEARTBEAT).map {
+ apiKey => new ApiVersion().setApiKey(apiKey.id).
+ setMinVersion(apiKey.oldestVersion()).setMaxVersion(apiKey.latestVersion())
+ }.toList.asJava)
+ val mockChannelManager = new MockBrokerToControllerChannelManager(mockClient,
+ time, controllerNodeProvider, nodeApiVersions)
+ val clusterId = Uuid.fromString("x4AJGXQSRnephtTZzujw4w")
+ val advertisedListeners = new ListenerCollection()
+ config.advertisedListeners.foreach { ep =>
+ advertisedListeners.add(new Listener().setHost(ep.host).
+ setName(ep.listenerName.value()).
+ setPort(ep.port.shortValue()).
+ setSecurityProtocol(ep.securityProtocol.id))
+ }
+
+ def poll(): Unit = {
+ mockClient.wakeup()
+ mockChannelManager.poll()
+ }
+ }
+
+ @Test
+ def testCreateAndClose(): Unit = {
+ val context = new BrokerLifecycleManagerTestContext(configProperties)
+ val manager = new BrokerLifecycleManager(context.config, context.time, None)
+ manager.close()
+ }
+
+ @Test
+ def testCreateStartAndClose(): Unit = {
+ val context = new BrokerLifecycleManagerTestContext(configProperties)
+ val manager = new BrokerLifecycleManager(context.config, context.time, None)
+ Assertions.assertEquals(BrokerState.NOT_RUNNING, manager.state())
+ manager.start(() => context.highestMetadataOffset.get(),
+ context.mockChannelManager, context.clusterId, context.advertisedListeners,
+ Collections.emptyMap())
+ TestUtils.retry(60000) {
+ Assertions.assertEquals(BrokerState.STARTING, manager.state())
+ }
+ manager.close()
+ Assertions.assertEquals(BrokerState.SHUTTING_DOWN, manager.state())
+ }
+
+ @Test
+ def testSuccessfulRegistration(): Unit = {
+ val context = new BrokerLifecycleManagerTestContext(configProperties)
+ val manager = new BrokerLifecycleManager(context.config, context.time, None)
+ val controllerNode = new Node(3000, "localhost", 8021)
+ context.controllerNodeProvider.node.set(controllerNode)
+ context.mockClient.prepareResponseFrom(new BrokerRegistrationResponse(
+ new BrokerRegistrationResponseData().setBrokerEpoch(1000)), controllerNode)
+ manager.start(() => context.highestMetadataOffset.get(),
+ context.mockChannelManager, context.clusterId, context.advertisedListeners,
+ Collections.emptyMap())
+ TestUtils.retry(10000) {
+ context.poll()
+ Assertions.assertEquals(1000L, manager.brokerEpoch())
+ }
+ manager.close()
+
+ }
+
+ @Test
+ def testRegistrationTimeout(): Unit = {
+ val context = new BrokerLifecycleManagerTestContext(configProperties)
+ val controllerNode = new Node(3000, "localhost", 8021)
+ val manager = new BrokerLifecycleManager(context.config, context.time, None)
+ context.controllerNodeProvider.node.set(controllerNode)
+ def newDuplicateRegistrationResponse(): Unit = {
+ context.mockClient.prepareResponseFrom(new BrokerRegistrationResponse(
+ new BrokerRegistrationResponseData().
+ setErrorCode(Errors.DUPLICATE_BROKER_REGISTRATION.code())), controllerNode)
+ context.mockChannelManager.poll()
+ }
+ newDuplicateRegistrationResponse()
+ Assertions.assertEquals(1, context.mockClient.futureResponses().size)
+ manager.start(() => context.highestMetadataOffset.get(),
+ context.mockChannelManager, context.clusterId, context.advertisedListeners,
+ Collections.emptyMap())
+ // We should send the first registration request and get a failure immediately
+ TestUtils.retry(60000) {
+ context.poll()
+ Assertions.assertEquals(0, context.mockClient.futureResponses().size)
+ }
+ // Verify that we resend the registration request.
+ newDuplicateRegistrationResponse()
+ TestUtils.retry(60000) {
+ context.time.sleep(100)
+ context.poll()
+ manager.eventQueue.wakeup()
+ Assertions.assertEquals(0, context.mockClient.futureResponses().size)
+ }
+ // Verify that we time out eventually.
+ context.time.sleep(300000)
+ TestUtils.retry(60000) {
+ context.poll()
+ manager.eventQueue.wakeup()
+ Assertions.assertEquals(BrokerState.SHUTTING_DOWN, manager.state())
+ Assertions.assertTrue(manager.initialCatchUpFuture.isCompletedExceptionally())
+ Assertions.assertEquals(-1L, manager.brokerEpoch())
+ }
+ manager.close()
+ }
+
+ @Test
+ def testControlledShutdown(): Unit = {
+ val context = new BrokerLifecycleManagerTestContext(configProperties)
+ val manager = new BrokerLifecycleManager(context.config, context.time, None)
+ val controllerNode = new Node(3000, "localhost", 8021)
+ context.controllerNodeProvider.node.set(controllerNode)
+ context.mockClient.prepareResponseFrom(new BrokerRegistrationResponse(
+ new BrokerRegistrationResponseData().setBrokerEpoch(1000)), controllerNode)
+ context.mockClient.prepareResponseFrom(new BrokerHeartbeatResponse(
+ new BrokerHeartbeatResponseData().setIsCaughtUp(true)), controllerNode)
+ manager.start(() => context.highestMetadataOffset.get(),
+ context.mockChannelManager, context.clusterId, context.advertisedListeners,
+ Collections.emptyMap())
+ TestUtils.retry(10000) {
+ context.poll()
+ manager.eventQueue.wakeup()
+ Assertions.assertEquals(BrokerState.RECOVERY, manager.state())
+ }
+ context.mockClient.prepareResponseFrom(new BrokerHeartbeatResponse(
+ new BrokerHeartbeatResponseData().setIsFenced(false)), controllerNode)
+ context.time.sleep(20)
+ TestUtils.retry(10000) {
+ context.poll()
+ manager.eventQueue.wakeup()
+ Assertions.assertEquals(BrokerState.RUNNING, manager.state())
+ }
+ manager.beginControlledShutdown()
+ TestUtils.retry(10000) {
+ context.poll()
+ manager.eventQueue.wakeup()
+ Assertions.assertEquals(BrokerState.PENDING_CONTROLLED_SHUTDOWN, manager.state())
+ }
+ context.mockClient.prepareResponseFrom(new BrokerHeartbeatResponse(
+ new BrokerHeartbeatResponseData().setShouldShutDown(true)), controllerNode)
+ context.time.sleep(3000)
+ TestUtils.retry(10000) {
+ context.poll()
+ manager.eventQueue.wakeup()
+ Assertions.assertEquals(BrokerState.SHUTTING_DOWN, manager.state())
+ }
+ manager.controlledShutdownFuture.get()
+ manager.close()
+ }
+}
Review comment:
nit: missing newline
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573970076
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
Review comment:
I added some JavaDoc for the class as a whole explaining the overall paradigm.
I think putting all the mutable state into a separate class is a bit too extreme. I added notes about how each mutable variable should be used, however.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mumrah commented on a change in pull request #10095: MINOR: Introduce the KIP-500 Broker lifecycle manager
Posted by GitBox <gi...@apache.org>.
mumrah commented on a change in pull request #10095:
URL: https://github.com/apache/kafka/pull/10095#discussion_r573845682
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
Review comment:
Why are some members prefixed with an underscore?
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ _state match {
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ info(s"Attempted to enter controlled shutdown state, but we are already in " +
+ s"that state.")
+ case BrokerState.RUNNING =>
+ info(s"Beginning controlled shutdown.")
+ _state = BrokerState.PENDING_CONTROLLED_SHUTDOWN
+ case _ =>
+ info(s"Skipping controlled shutdown because we are in state ${_state}.")
+ beginShutdown()
+ }
+ }
+ }
+
+ /**
+ * Enter the controlled shutdown state if we are in RUNNING state.
+ * Or, if we're not running, shut down immediately.
+ */
+ def beginControlledShutdown(): Unit = {
+ eventQueue.append(new BeginControlledShutdownEvent())
+ }
+
+ /**
+ * Start shutting down the BrokerLifecycleManager, but do not block.
+ */
+ def beginShutdown(): Unit = {
+ eventQueue.beginShutdown("beginShutdown", new ShutdownEvent())
+ }
+
+ /**
+ * Shut down the BrokerLifecycleManager and block until all threads are joined.
+ */
+ def close(): Unit = {
+ beginShutdown()
+ eventQueue.close()
+ }
+
+ class SetReadyToUnfenceEvent() extends EventQueue.Event {
+ override def run(): Unit = {
+ readyToUnfence = true
+ scheduleNextCommunicationImmediately()
+ }
+ }
+
+ class StartupEvent(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]) extends EventQueue.Event {
+ override def run(): Unit = {
+ _highestMetadataOffsetProvider = highestMetadataOffsetProvider
+ _channelManager = channelManager
+ _channelManager.start()
Review comment:
If this class is owning the lifecycle for `_channelManager`, should it not also instantiate it?
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ _state match {
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ info(s"Attempted to enter controlled shutdown state, but we are already in " +
+ s"that state.")
+ case BrokerState.RUNNING =>
+ info(s"Beginning controlled shutdown.")
+ _state = BrokerState.PENDING_CONTROLLED_SHUTDOWN
+ case _ =>
+ info(s"Skipping controlled shutdown because we are in state ${_state}.")
+ beginShutdown()
+ }
+ }
+ }
+
+ /**
+ * Enter the controlled shutdown state if we are in RUNNING state.
+ * Or, if we're not running, shut down immediately.
+ */
+ def beginControlledShutdown(): Unit = {
+ eventQueue.append(new BeginControlledShutdownEvent())
+ }
+
+ /**
+ * Start shutting down the BrokerLifecycleManager, but do not block.
+ */
+ def beginShutdown(): Unit = {
+ eventQueue.beginShutdown("beginShutdown", new ShutdownEvent())
+ }
+
+ /**
+ * Shut down the BrokerLifecycleManager and block until all threads are joined.
+ */
+ def close(): Unit = {
+ beginShutdown()
+ eventQueue.close()
+ }
+
+ class SetReadyToUnfenceEvent() extends EventQueue.Event {
+ override def run(): Unit = {
+ readyToUnfence = true
+ scheduleNextCommunicationImmediately()
+ }
+ }
+
+ class StartupEvent(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]) extends EventQueue.Event {
+ override def run(): Unit = {
+ _highestMetadataOffsetProvider = highestMetadataOffsetProvider
+ _channelManager = channelManager
+ _channelManager.start()
+ _state = BrokerState.STARTING
+ _clusterId = clusterId
+ _advertisedListeners = advertisedListeners
+ _supportedFeatures = supportedFeatures
+ eventQueue.scheduleDeferred("initialRegistrationTimeout",
+ new DeadlineFunction(time.nanoseconds() + initialTimeoutNs),
+ new RegistrationTimeoutEvent())
+ sendBrokerRegistration()
+ info(s"Incarnation ${incarnationId} of broker ${nodeId} in cluster ${clusterId} " +
+ "is now STARTING.")
+ }
+ }
+
+ private def sendBrokerRegistration(): Unit = {
+ val features = new BrokerRegistrationRequestData.FeatureCollection()
+ _supportedFeatures.asScala.foreach {
+ case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ setName(name).
+ setMinSupportedVersion(range.min()).
+ setMaxSupportedVersion(range.max()))
+ }
+ val data = new BrokerRegistrationRequestData().
+ setBrokerId(nodeId).
+ setClusterId(_clusterId).
+ setFeatures(features).
+ setIncarnationId(incarnationId).
+ setListeners(_advertisedListeners).
+ setRack(rack)
+ if (isTraceEnabled) {
+ trace(s"Sending broker registration ${data}")
+ }
+ _channelManager.sendRequest(new BrokerRegistrationRequest.Builder(data),
+ new BrokerRegistrationResponseHandler())
+ }
+
+ class BrokerRegistrationResponseHandler extends ControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ if (response.authenticationException() != null) {
+ error(s"Unable to register broker ${nodeId} because of an authentication exception.",
+ response.authenticationException());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.versionMismatch() != null) {
+ error(s"Unable to register broker ${nodeId} because of an API version problem.",
+ response.versionMismatch());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.responseBody() == null) {
+ warn(s"Unable to register broker ${nodeId}.")
+ scheduleNextCommunicationAfterFailure()
+ } else if (!response.responseBody().isInstanceOf[BrokerRegistrationResponse]) {
+ error(s"Unable to register broker ${nodeId} because the controller returned an " +
+ "invalid response type.")
+ scheduleNextCommunicationAfterFailure()
+ } else {
+ val message = response.responseBody().asInstanceOf[BrokerRegistrationResponse]
+ val errorCode = Errors.forCode(message.data().errorCode())
+ if (errorCode == Errors.NONE) {
+ failedAttempts = 0
+ _brokerEpoch = message.data().brokerEpoch()
+ registered = true
+ initialRegistrationSucceeded = true
+ info(s"Successfully registered broker ${nodeId} with broker epoch ${_brokerEpoch}")
+ scheduleNextCommunicationImmediately() // Immediately send a heartbeat
+ } else {
+ info(s"Unable to register broker ${nodeId} because the controller returned " +
+ s"error ${errorCode}")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+ }
+
+ override def onTimeout(): Unit = {
+ info(s"Unable to register the broker because the RPC got timed out before it could be sent.")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+
+ private def sendBrokerHeartbeat(): Unit = {
+ val metadataOffset = _highestMetadataOffsetProvider()
+ val data = new BrokerHeartbeatRequestData().
+ setBrokerEpoch(_brokerEpoch).
+ setBrokerId(nodeId).
+ setCurrentMetadataOffset(metadataOffset).
+ setWantFence(!readyToUnfence).
+ setWantShutDown(_state == BrokerState.PENDING_CONTROLLED_SHUTDOWN)
+ if (isTraceEnabled) {
+ trace(s"Sending broker heartbeat ${data}")
+ }
+ _channelManager.sendRequest(new BrokerHeartbeatRequest.Builder(data),
+ new BrokerHeartbeatResponseHandler())
+ }
+
+ class BrokerHeartbeatResponseHandler extends ControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ if (response.authenticationException() != null) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because of an " +
+ "authentication exception.", response.authenticationException());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.versionMismatch() != null) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because of an API " +
+ "version problem.", response.versionMismatch());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.responseBody() == null) {
+ warn(s"Unable to send broker heartbeat for ${nodeId}. Retrying.")
+ scheduleNextCommunicationAfterFailure()
+ } else if (!response.responseBody().isInstanceOf[BrokerHeartbeatResponse]) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because the controller " +
+ "returned an invalid response type.")
+ scheduleNextCommunicationAfterFailure()
+ } else {
+ val message = response.responseBody().asInstanceOf[BrokerHeartbeatResponse]
+ val errorCode = Errors.forCode(message.data().errorCode())
+ if (errorCode == Errors.NONE) {
+ failedAttempts = 0
+ _state match {
+ case BrokerState.STARTING =>
+ if (message.data().isCaughtUp()) {
+ info(s"The broker has caught up. Transitioning from STARTING to RECOVERY.")
+ _state = BrokerState.RECOVERY
+ initialCatchUpFuture.complete(null)
+ } else {
+ info(s"The broker is STARTING. Still waiting to catch up with cluster metadata.")
+ }
+ // Schedule the heartbeat after only 10 ms so that in the case where
+ //there is no recovery work to be done, we start up a bit quicker.
+ scheduleNextCommunication(NANOSECONDS.convert(10, MILLISECONDS))
Review comment:
We are scheduling the next heartbeat in the near future here so we don't have to wait for the full heartbeat interval in order to make it to RUNNING. But i think this will only happen if a SetReadyToUnfenceEvent is handled between the broker entering RECOVERY and the next heartbeat request, in which case that event handler will send a heartbeat right away. If that's the case, do we really need this 10ms here or should we just wait the normal heartbeat interval?
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ _state match {
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ info(s"Attempted to enter controlled shutdown state, but we are already in " +
+ s"that state.")
+ case BrokerState.RUNNING =>
+ info(s"Beginning controlled shutdown.")
+ _state = BrokerState.PENDING_CONTROLLED_SHUTDOWN
+ case _ =>
+ info(s"Skipping controlled shutdown because we are in state ${_state}.")
+ beginShutdown()
+ }
+ }
+ }
+
+ /**
+ * Enter the controlled shutdown state if we are in RUNNING state.
+ * Or, if we're not running, shut down immediately.
+ */
+ def beginControlledShutdown(): Unit = {
+ eventQueue.append(new BeginControlledShutdownEvent())
+ }
+
+ /**
+ * Start shutting down the BrokerLifecycleManager, but do not block.
+ */
+ def beginShutdown(): Unit = {
+ eventQueue.beginShutdown("beginShutdown", new ShutdownEvent())
+ }
+
+ /**
+ * Shut down the BrokerLifecycleManager and block until all threads are joined.
+ */
+ def close(): Unit = {
+ beginShutdown()
+ eventQueue.close()
+ }
+
+ class SetReadyToUnfenceEvent() extends EventQueue.Event {
+ override def run(): Unit = {
+ readyToUnfence = true
+ scheduleNextCommunicationImmediately()
+ }
+ }
+
+ class StartupEvent(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]) extends EventQueue.Event {
+ override def run(): Unit = {
+ _highestMetadataOffsetProvider = highestMetadataOffsetProvider
+ _channelManager = channelManager
+ _channelManager.start()
+ _state = BrokerState.STARTING
+ _clusterId = clusterId
+ _advertisedListeners = advertisedListeners
+ _supportedFeatures = supportedFeatures
Review comment:
Should we be copying this map rather than assigning to it? Do we want things outside of this class to mutate the data in this map?
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ _state match {
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ info(s"Attempted to enter controlled shutdown state, but we are already in " +
+ s"that state.")
+ case BrokerState.RUNNING =>
+ info(s"Beginning controlled shutdown.")
+ _state = BrokerState.PENDING_CONTROLLED_SHUTDOWN
+ case _ =>
+ info(s"Skipping controlled shutdown because we are in state ${_state}.")
+ beginShutdown()
+ }
+ }
+ }
+
+ /**
+ * Enter the controlled shutdown state if we are in RUNNING state.
+ * Or, if we're not running, shut down immediately.
+ */
+ def beginControlledShutdown(): Unit = {
+ eventQueue.append(new BeginControlledShutdownEvent())
+ }
+
+ /**
+ * Start shutting down the BrokerLifecycleManager, but do not block.
+ */
+ def beginShutdown(): Unit = {
+ eventQueue.beginShutdown("beginShutdown", new ShutdownEvent())
+ }
+
+ /**
+ * Shut down the BrokerLifecycleManager and block until all threads are joined.
+ */
+ def close(): Unit = {
+ beginShutdown()
+ eventQueue.close()
+ }
+
+ class SetReadyToUnfenceEvent() extends EventQueue.Event {
+ override def run(): Unit = {
+ readyToUnfence = true
+ scheduleNextCommunicationImmediately()
+ }
+ }
+
+ class StartupEvent(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]) extends EventQueue.Event {
+ override def run(): Unit = {
+ _highestMetadataOffsetProvider = highestMetadataOffsetProvider
+ _channelManager = channelManager
+ _channelManager.start()
+ _state = BrokerState.STARTING
+ _clusterId = clusterId
+ _advertisedListeners = advertisedListeners
+ _supportedFeatures = supportedFeatures
+ eventQueue.scheduleDeferred("initialRegistrationTimeout",
+ new DeadlineFunction(time.nanoseconds() + initialTimeoutNs),
+ new RegistrationTimeoutEvent())
+ sendBrokerRegistration()
+ info(s"Incarnation ${incarnationId} of broker ${nodeId} in cluster ${clusterId} " +
+ "is now STARTING.")
+ }
+ }
+
+ private def sendBrokerRegistration(): Unit = {
+ val features = new BrokerRegistrationRequestData.FeatureCollection()
+ _supportedFeatures.asScala.foreach {
+ case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ setName(name).
+ setMinSupportedVersion(range.min()).
+ setMaxSupportedVersion(range.max()))
+ }
+ val data = new BrokerRegistrationRequestData().
+ setBrokerId(nodeId).
+ setClusterId(_clusterId).
+ setFeatures(features).
+ setIncarnationId(incarnationId).
+ setListeners(_advertisedListeners).
+ setRack(rack)
+ if (isTraceEnabled) {
+ trace(s"Sending broker registration ${data}")
+ }
+ _channelManager.sendRequest(new BrokerRegistrationRequest.Builder(data),
+ new BrokerRegistrationResponseHandler())
+ }
+
+ class BrokerRegistrationResponseHandler extends ControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ if (response.authenticationException() != null) {
+ error(s"Unable to register broker ${nodeId} because of an authentication exception.",
+ response.authenticationException());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.versionMismatch() != null) {
+ error(s"Unable to register broker ${nodeId} because of an API version problem.",
+ response.versionMismatch());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.responseBody() == null) {
+ warn(s"Unable to register broker ${nodeId}.")
+ scheduleNextCommunicationAfterFailure()
+ } else if (!response.responseBody().isInstanceOf[BrokerRegistrationResponse]) {
+ error(s"Unable to register broker ${nodeId} because the controller returned an " +
+ "invalid response type.")
+ scheduleNextCommunicationAfterFailure()
+ } else {
+ val message = response.responseBody().asInstanceOf[BrokerRegistrationResponse]
+ val errorCode = Errors.forCode(message.data().errorCode())
+ if (errorCode == Errors.NONE) {
+ failedAttempts = 0
+ _brokerEpoch = message.data().brokerEpoch()
+ registered = true
+ initialRegistrationSucceeded = true
+ info(s"Successfully registered broker ${nodeId} with broker epoch ${_brokerEpoch}")
+ scheduleNextCommunicationImmediately() // Immediately send a heartbeat
+ } else {
+ info(s"Unable to register broker ${nodeId} because the controller returned " +
+ s"error ${errorCode}")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+ }
+
+ override def onTimeout(): Unit = {
+ info(s"Unable to register the broker because the RPC got timed out before it could be sent.")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+
+ private def sendBrokerHeartbeat(): Unit = {
+ val metadataOffset = _highestMetadataOffsetProvider()
+ val data = new BrokerHeartbeatRequestData().
+ setBrokerEpoch(_brokerEpoch).
+ setBrokerId(nodeId).
+ setCurrentMetadataOffset(metadataOffset).
+ setWantFence(!readyToUnfence).
+ setWantShutDown(_state == BrokerState.PENDING_CONTROLLED_SHUTDOWN)
+ if (isTraceEnabled) {
+ trace(s"Sending broker heartbeat ${data}")
+ }
+ _channelManager.sendRequest(new BrokerHeartbeatRequest.Builder(data),
+ new BrokerHeartbeatResponseHandler())
+ }
+
+ class BrokerHeartbeatResponseHandler extends ControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ if (response.authenticationException() != null) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because of an " +
+ "authentication exception.", response.authenticationException());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.versionMismatch() != null) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because of an API " +
+ "version problem.", response.versionMismatch());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.responseBody() == null) {
+ warn(s"Unable to send broker heartbeat for ${nodeId}. Retrying.")
+ scheduleNextCommunicationAfterFailure()
+ } else if (!response.responseBody().isInstanceOf[BrokerHeartbeatResponse]) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because the controller " +
+ "returned an invalid response type.")
+ scheduleNextCommunicationAfterFailure()
+ } else {
+ val message = response.responseBody().asInstanceOf[BrokerHeartbeatResponse]
+ val errorCode = Errors.forCode(message.data().errorCode())
+ if (errorCode == Errors.NONE) {
+ failedAttempts = 0
+ _state match {
+ case BrokerState.STARTING =>
+ if (message.data().isCaughtUp()) {
+ info(s"The broker has caught up. Transitioning from STARTING to RECOVERY.")
+ _state = BrokerState.RECOVERY
+ initialCatchUpFuture.complete(null)
+ } else {
+ info(s"The broker is STARTING. Still waiting to catch up with cluster metadata.")
+ }
+ // Schedule the heartbeat after only 10 ms so that in the case where
+ //there is no recovery work to be done, we start up a bit quicker.
+ scheduleNextCommunication(NANOSECONDS.convert(10, MILLISECONDS))
+ case BrokerState.RECOVERY =>
+ if (!message.data().isFenced()) {
+ info(s"The broker has been unfenced. Transitioning from RECOVERY to RUNNING.")
+ _state = BrokerState.RUNNING
+ } else {
+ info(s"The broker is in RECOVERY.")
+ }
+ scheduleNextCommunicationAfterSuccess()
+ case BrokerState.RUNNING =>
+ debug(s"The broker is RUNNING. Processing heartbeat response.")
+ scheduleNextCommunicationAfterSuccess()
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ if (!message.data().shouldShutDown()) {
+ info(s"The broker is in PENDING_CONTROLLED_SHUTDOWN state, still waiting " +
+ "for the active controller.")
+ if (!gotControlledShutdownResponse) {
+ // If this is the first pending controlled shutdown response we got,
+ // schedule our next heartbeat a little bit sooner than we usually would.
+ // In the case where controlled shutdown completes quickly, this will
+ // speed things up a little bit.
+ scheduleNextCommunication(NANOSECONDS.convert(50, MILLISECONDS))
+ } else {
+ scheduleNextCommunicationAfterSuccess()
+ }
+ } else {
+ info(s"The controlled has asked us to exit controlled shutdown.")
+ beginShutdown()
+ }
+ gotControlledShutdownResponse = true
+ case BrokerState.SHUTTING_DOWN =>
+ info(s"The broker is SHUTTING_DOWN. Ignoring heartbeat response.")
+ case _ =>
+ error(s"Unexpected broker state ${_state}")
+ scheduleNextCommunicationAfterSuccess()
+ }
+ } else {
+ warn(s"Broker ${nodeId} sent a heartbeat request but received error ${errorCode}.")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+ }
+
+ override def onTimeout(): Unit = {
+ info("Unable to send a heartbeat because the RPC got timed out before it could be sent.")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+
+ private def scheduleNextCommunicationImmediately(): Unit = scheduleNextCommunication(0)
+
+ private def scheduleNextCommunicationAfterFailure(): Unit = {
+ val delayMs = resendExponentialBackoff.backoff(failedAttempts)
+ failedAttempts = failedAttempts + 1
+ scheduleNextCommunication(NANOSECONDS.convert(delayMs, MILLISECONDS))
+ }
+
+ private def scheduleNextCommunicationAfterSuccess(): Unit = {
+ scheduleNextCommunication(NANOSECONDS.convert(
+ config.brokerHeartbeatIntervalMs.longValue() , MILLISECONDS))
+ }
+
+ private def scheduleNextCommunication(intervalNs: Long): Unit = {
+ trace(s"Scheduling next communication at ${MILLISECONDS.convert(intervalNs, NANOSECONDS)} " +
+ "ms from now.")
+ val deadlineNs = time.nanoseconds() + intervalNs
+ eventQueue.scheduleDeferred("communication",
+ new DeadlineFunction(deadlineNs),
+ new CommunicationEvent())
+ }
+
+ class RegistrationTimeoutEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ if (!initialRegistrationSucceeded) {
+ error("Shutting down because we were unable to register with the controller quorum.")
+ eventQueue.beginShutdown("registrationTimeout", new ShutdownEvent())
+ }
+ }
+ }
+
+ class CommunicationEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ if (registered) {
+ sendBrokerHeartbeat()
+ } else {
+ sendBrokerRegistration()
+ }
+ }
+ }
+
+ class ShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ info(s"Transitioning from ${_state} to ${BrokerState.SHUTTING_DOWN}.")
+ _state = BrokerState.SHUTTING_DOWN
+ controlledShutdownFuture.complete(null)
+ initialCatchUpFuture.cancel(false)
+ _channelManager.shutdown()
Review comment:
Do we need to guard against `_channelManager` being null here? Is it possible that `close` is called without `start` having been called, maybe in some faulty startup scenario?
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
+
+ /**
+ * The features supported by this broker.
+ */
+ private var _supportedFeatures: util.Map[String, VersionRange] = null
+
+ /**
+ * The channel manager, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ var _channelManager: BrokerToControllerChannelManager = null
+
+ /**
+ * The event queue.
+ */
+ val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
+
+ /**
+ * Start the BrokerLifecycleManager.
+ *
+ * @param highestMetadataOffsetProvider Provides the current highest metadata offset.
+ * @param channelManager The brokerToControllerChannelManager to use.
+ * @param clusterId The cluster ID.
+ */
+ def start(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]): Unit = {
+ eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
+ channelManager, clusterId, advertisedListeners, supportedFeatures))
+ }
+
+ def setReadyToUnfence(): Unit = {
+ eventQueue.append(new SetReadyToUnfenceEvent())
+ }
+
+ def brokerEpoch(): Long = _brokerEpoch
+
+ def state(): BrokerState = _state
+
+ class BeginControlledShutdownEvent extends EventQueue.Event {
+ override def run(): Unit = {
+ _state match {
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ info(s"Attempted to enter controlled shutdown state, but we are already in " +
+ s"that state.")
+ case BrokerState.RUNNING =>
+ info(s"Beginning controlled shutdown.")
+ _state = BrokerState.PENDING_CONTROLLED_SHUTDOWN
+ case _ =>
+ info(s"Skipping controlled shutdown because we are in state ${_state}.")
+ beginShutdown()
+ }
+ }
+ }
+
+ /**
+ * Enter the controlled shutdown state if we are in RUNNING state.
+ * Or, if we're not running, shut down immediately.
+ */
+ def beginControlledShutdown(): Unit = {
+ eventQueue.append(new BeginControlledShutdownEvent())
+ }
+
+ /**
+ * Start shutting down the BrokerLifecycleManager, but do not block.
+ */
+ def beginShutdown(): Unit = {
+ eventQueue.beginShutdown("beginShutdown", new ShutdownEvent())
+ }
+
+ /**
+ * Shut down the BrokerLifecycleManager and block until all threads are joined.
+ */
+ def close(): Unit = {
+ beginShutdown()
+ eventQueue.close()
+ }
+
+ class SetReadyToUnfenceEvent() extends EventQueue.Event {
+ override def run(): Unit = {
+ readyToUnfence = true
+ scheduleNextCommunicationImmediately()
+ }
+ }
+
+ class StartupEvent(highestMetadataOffsetProvider: () => Long,
+ channelManager: BrokerToControllerChannelManager,
+ clusterId: Uuid,
+ advertisedListeners: ListenerCollection,
+ supportedFeatures: util.Map[String, VersionRange]) extends EventQueue.Event {
+ override def run(): Unit = {
+ _highestMetadataOffsetProvider = highestMetadataOffsetProvider
+ _channelManager = channelManager
+ _channelManager.start()
+ _state = BrokerState.STARTING
+ _clusterId = clusterId
+ _advertisedListeners = advertisedListeners
+ _supportedFeatures = supportedFeatures
+ eventQueue.scheduleDeferred("initialRegistrationTimeout",
+ new DeadlineFunction(time.nanoseconds() + initialTimeoutNs),
+ new RegistrationTimeoutEvent())
+ sendBrokerRegistration()
+ info(s"Incarnation ${incarnationId} of broker ${nodeId} in cluster ${clusterId} " +
+ "is now STARTING.")
+ }
+ }
+
+ private def sendBrokerRegistration(): Unit = {
+ val features = new BrokerRegistrationRequestData.FeatureCollection()
+ _supportedFeatures.asScala.foreach {
+ case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ setName(name).
+ setMinSupportedVersion(range.min()).
+ setMaxSupportedVersion(range.max()))
+ }
+ val data = new BrokerRegistrationRequestData().
+ setBrokerId(nodeId).
+ setClusterId(_clusterId).
+ setFeatures(features).
+ setIncarnationId(incarnationId).
+ setListeners(_advertisedListeners).
+ setRack(rack)
+ if (isTraceEnabled) {
+ trace(s"Sending broker registration ${data}")
+ }
+ _channelManager.sendRequest(new BrokerRegistrationRequest.Builder(data),
+ new BrokerRegistrationResponseHandler())
+ }
+
+ class BrokerRegistrationResponseHandler extends ControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ if (response.authenticationException() != null) {
+ error(s"Unable to register broker ${nodeId} because of an authentication exception.",
+ response.authenticationException());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.versionMismatch() != null) {
+ error(s"Unable to register broker ${nodeId} because of an API version problem.",
+ response.versionMismatch());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.responseBody() == null) {
+ warn(s"Unable to register broker ${nodeId}.")
+ scheduleNextCommunicationAfterFailure()
+ } else if (!response.responseBody().isInstanceOf[BrokerRegistrationResponse]) {
+ error(s"Unable to register broker ${nodeId} because the controller returned an " +
+ "invalid response type.")
+ scheduleNextCommunicationAfterFailure()
+ } else {
+ val message = response.responseBody().asInstanceOf[BrokerRegistrationResponse]
+ val errorCode = Errors.forCode(message.data().errorCode())
+ if (errorCode == Errors.NONE) {
+ failedAttempts = 0
+ _brokerEpoch = message.data().brokerEpoch()
+ registered = true
+ initialRegistrationSucceeded = true
+ info(s"Successfully registered broker ${nodeId} with broker epoch ${_brokerEpoch}")
+ scheduleNextCommunicationImmediately() // Immediately send a heartbeat
+ } else {
+ info(s"Unable to register broker ${nodeId} because the controller returned " +
+ s"error ${errorCode}")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+ }
+
+ override def onTimeout(): Unit = {
+ info(s"Unable to register the broker because the RPC got timed out before it could be sent.")
+ scheduleNextCommunicationAfterFailure()
+ }
+ }
+
+ private def sendBrokerHeartbeat(): Unit = {
+ val metadataOffset = _highestMetadataOffsetProvider()
+ val data = new BrokerHeartbeatRequestData().
+ setBrokerEpoch(_brokerEpoch).
+ setBrokerId(nodeId).
+ setCurrentMetadataOffset(metadataOffset).
+ setWantFence(!readyToUnfence).
+ setWantShutDown(_state == BrokerState.PENDING_CONTROLLED_SHUTDOWN)
+ if (isTraceEnabled) {
+ trace(s"Sending broker heartbeat ${data}")
+ }
+ _channelManager.sendRequest(new BrokerHeartbeatRequest.Builder(data),
+ new BrokerHeartbeatResponseHandler())
+ }
+
+ class BrokerHeartbeatResponseHandler extends ControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ if (response.authenticationException() != null) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because of an " +
+ "authentication exception.", response.authenticationException());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.versionMismatch() != null) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because of an API " +
+ "version problem.", response.versionMismatch());
+ scheduleNextCommunicationAfterFailure()
+ } else if (response.responseBody() == null) {
+ warn(s"Unable to send broker heartbeat for ${nodeId}. Retrying.")
+ scheduleNextCommunicationAfterFailure()
+ } else if (!response.responseBody().isInstanceOf[BrokerHeartbeatResponse]) {
+ error(s"Unable to send broker heartbeat for ${nodeId} because the controller " +
+ "returned an invalid response type.")
+ scheduleNextCommunicationAfterFailure()
+ } else {
+ val message = response.responseBody().asInstanceOf[BrokerHeartbeatResponse]
+ val errorCode = Errors.forCode(message.data().errorCode())
+ if (errorCode == Errors.NONE) {
+ failedAttempts = 0
+ _state match {
+ case BrokerState.STARTING =>
+ if (message.data().isCaughtUp()) {
+ info(s"The broker has caught up. Transitioning from STARTING to RECOVERY.")
+ _state = BrokerState.RECOVERY
+ initialCatchUpFuture.complete(null)
+ } else {
+ info(s"The broker is STARTING. Still waiting to catch up with cluster metadata.")
+ }
+ // Schedule the heartbeat after only 10 ms so that in the case where
+ //there is no recovery work to be done, we start up a bit quicker.
+ scheduleNextCommunication(NANOSECONDS.convert(10, MILLISECONDS))
+ case BrokerState.RECOVERY =>
+ if (!message.data().isFenced()) {
+ info(s"The broker has been unfenced. Transitioning from RECOVERY to RUNNING.")
+ _state = BrokerState.RUNNING
+ } else {
+ info(s"The broker is in RECOVERY.")
+ }
+ scheduleNextCommunicationAfterSuccess()
+ case BrokerState.RUNNING =>
+ debug(s"The broker is RUNNING. Processing heartbeat response.")
+ scheduleNextCommunicationAfterSuccess()
+ case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
+ if (!message.data().shouldShutDown()) {
+ info(s"The broker is in PENDING_CONTROLLED_SHUTDOWN state, still waiting " +
+ "for the active controller.")
+ if (!gotControlledShutdownResponse) {
+ // If this is the first pending controlled shutdown response we got,
+ // schedule our next heartbeat a little bit sooner than we usually would.
+ // In the case where controlled shutdown completes quickly, this will
+ // speed things up a little bit.
+ scheduleNextCommunication(NANOSECONDS.convert(50, MILLISECONDS))
Review comment:
Since the heartbeat interval is configurable, maybe we should calculate this from the interval instead of a fixed value. Maybe something like (interval / 2) or sqrt(interval)?
##########
File path: core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
##########
@@ -0,0 +1,457 @@
+/**
+ * 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.server
+
+import java.util
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
+import java.util.concurrent.{CompletableFuture, TimeUnit}
+import kafka.utils.Logging
+import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.Uuid
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection
+import org.apache.kafka.common.message.{BrokerHeartbeatRequestData, BrokerRegistrationRequestData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
+import org.apache.kafka.queue.EventQueue.DeadlineFunction
+import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
+import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+import scala.jdk.CollectionConverters._
+
+
+class BrokerLifecycleManager(val config: KafkaConfig,
+ val time: Time,
+ val threadNamePrefix: Option[String]) extends Logging {
+ val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
+
+ this.logIdent = logContext.logPrefix()
+
+ /**
+ * The broker id.
+ */
+ private val nodeId = config.nodeId
+
+ /**
+ * The broker rack, or null if there is no configured rack.
+ */
+ private val rack = config.rack.orNull
+
+ /**
+ * How long to wait for registration to succeed before failing the startup process.
+ */
+ private val initialTimeoutNs = NANOSECONDS.
+ convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+
+ /**
+ * The exponential backoff to use for resending communication.
+ */
+ private val resendExponentialBackoff =
+ new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+
+ /**
+ * The number of tries we've tried to communicate.
+ */
+ private var failedAttempts = 0L
+
+ /**
+ * The broker incarnation ID. This ID uniquely identifies each time we start the broker
+ */
+ val incarnationId = Uuid.randomUuid()
+
+ /**
+ * A future which is completed just as soon as the broker has caught up with the latest
+ * metadata offset for the first time.
+ */
+ val initialCatchUpFuture = new CompletableFuture[Void]()
+
+ /**
+ * A future which is completed when controlled shutdown is done.
+ */
+ val controlledShutdownFuture = new CompletableFuture[Void]()
+
+ /**
+ * The broker epoch, or -1 if the broker has not yet registered.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _brokerEpoch = -1L
+
+ /**
+ * The current broker state.
+ * This variable can only be written from the event queue thread.
+ */
+ @volatile private var _state = BrokerState.NOT_RUNNING
+
+ /**
+ * A callback function which gives this manager the current highest metadata offset.
+ * This function must be thread-safe.
+ */
+ private var _highestMetadataOffsetProvider: () => Long = null
+
+ /**
+ * True only if we are ready to unfence the broker.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var readyToUnfence = false
+
+ /**
+ * True if we sent a heartbeat to the active controller requesting controlled
+ * shutdown.
+ */
+ private var gotControlledShutdownResponse = false
+
+ /**
+ * Whether or not we this broker is registered with the controller quorum.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var registered = false
+
+ /**
+ * True if the initial registration succeeded.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var initialRegistrationSucceeded = false
+
+ /**
+ * The cluster ID, or null if this manager has not been started yet.
+ * This variable can only be accessed from the event queue thread.
+ */
+ private var _clusterId: Uuid = null
+
+ /**
+ * The listeners which this broker advertises.
+ */
+ private var _advertisedListeners: ListenerCollection = null
Review comment:
nit: I think scala prefers the underscore initializer for cases like this. E.g.,
```scala
private var _advertisedListeners: ListenerCollection = _
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org