From be3b32b1498b66564b922a578cef5780dc6fd397 Mon Sep 17 00:00:00 2001
From: "Colin P. Mccabe" <cmccabe@confluent.io>
Date: Thu, 4 Feb 2021 17:45:18 -0800
Subject: [PATCH 1/7] MINOR: add BrokerLifecycleManager

The BrokerLifecycleManager handles broker state transitions for the
KIP-500 broker.  This includes sending broker registrations, heartbeats,
and controlled shutdown requests.
---
 .../kafka/server/BrokerLifecycleManager.scala | 447 ++++++++++++++++++
 .../main/scala/kafka/server/KafkaConfig.scala |  12 +
 .../server/BrokerLifecycleManagerTest.scala   | 195 ++++++++
 3 files changed, 654 insertions(+)
 create mode 100644 core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
 create mode 100644 core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala

diff --git a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
new file mode 100644
index 0000000000000..efb812a6c898a
--- /dev/null
+++ b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
@@ -0,0 +1,447 @@
+/**
+ * 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, RequestCompletionHandler}
+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.brokerId}] ")
+
+  this.logIdent = logContext.logPrefix()
+
+  /**
+   * The broker id.
+   */
+  private val brokerId = config.brokerId
+
+  /**
+   * 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 ${brokerId} 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(brokerId).
+        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 RequestCompletionHandler {
+    override def onComplete(response: ClientResponse): Unit = {
+      if (response.authenticationException() != null) {
+        error(s"Unable to register broker ${brokerId} because of an authentication exception.",
+          response.authenticationException());
+        scheduleNextCommunicationAfterFailure()
+      } else if (response.versionMismatch() != null) {
+        error(s"Unable to register broker ${brokerId} because of an API version problem.",
+          response.versionMismatch());
+        scheduleNextCommunicationAfterFailure()
+      } else if (response.responseBody() == null) {
+        warn(s"Unable to register broker ${brokerId}.")
+        scheduleNextCommunicationAfterFailure()
+      } else if (!response.responseBody().isInstanceOf[BrokerRegistrationResponse]) {
+        error(s"Unable to register broker ${brokerId} 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 ${brokerId} with broker epoch ${_brokerEpoch}")
+          scheduleNextCommunicationImmediately() // Immediately send a heartbeat
+        } else {
+          info(s"Unable to register broker ${brokerId} because the controller returned " +
+            s"error ${errorCode}")
+          scheduleNextCommunicationAfterFailure()
+        }
+      }
+    }
+  }
+
+  private def sendBrokerHeartbeat(): Unit = {
+    val metadataOffset = _highestMetadataOffsetProvider()
+    val data = new BrokerHeartbeatRequestData().
+      setBrokerEpoch(_brokerEpoch).
+      setBrokerId(brokerId).
+      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 RequestCompletionHandler {
+    override def onComplete(response: ClientResponse): Unit = {
+      if (response.authenticationException() != null) {
+        error(s"Unable to send broker heartbeat for ${brokerId} because of an " +
+          "authentication exception.", response.authenticationException());
+        scheduleNextCommunicationAfterFailure()
+      } else if (response.versionMismatch() != null) {
+        error(s"Unable to send broker heartbeat for ${brokerId} because of an API " +
+          "version problem.", response.versionMismatch());
+        scheduleNextCommunicationAfterFailure()
+      } else if (response.responseBody() == null) {
+        warn(s"Unable to send broker heartbeat for ${brokerId}. Retrying.")
+        scheduleNextCommunicationAfterFailure()
+      } else if (!response.responseBody().isInstanceOf[BrokerHeartbeatResponse]) {
+        error(s"Unable to send broker heartbeat for ${brokerId} 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 ${brokerId} sent a heartbeat request but received error ${errorCode}.")
+          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()
+    }
+  }
+}
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 2db02a1a1933e..03c7818441bff 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -70,6 +70,9 @@ object Defaults {
   val BackgroundThreads = 10
   val QueuedMaxRequests = 500
   val QueuedMaxRequestBytes = -1
+  val InitialBrokerRegistrationTimeoutMs = 60000
+  val BrokerHeartbeatIntervalMs = 2000
+  val BrokerSessionTimeoutMs = 18000
 
   /** KIP-500 Configuration */
   val EmptyNodeId: Int = -1
@@ -369,6 +372,9 @@ object KafkaConfig {
 
   /** KIP-500 Configuration */
   val ProcessRolesProp = "process.roles"
+  val InitialBrokerRegistrationTimeoutMs = "initial.broker.registration.timeout.ms"
+  val BrokerHeartbeatIntervalMsProp = "broker.heartbeat.interval.ms"
+  val BrokerSessionTimeoutMsProp = "broker.session.timeout.ms"
   val NodeIdProp = "node.id"
   val MetadataLogDirProp = "metadata.log.dir"
 
@@ -1064,6 +1070,9 @@ object KafkaConfig {
        */
       .defineInternal(ProcessRolesProp, LIST, Collections.emptyList(), ValidList.in("broker", "controller"), HIGH, ProcessRolesDoc)
       .defineInternal(NodeIdProp, INT, Defaults.EmptyNodeId, null, HIGH, NodeIdDoc)
+      .defineInternal(InitialBrokerRegistrationTimeoutMs, INT, Defaults.InitialBrokerRegistrationTimeoutMs, MEDIUM, InitialBrokerRegistrationTimeoutMsDoc)
+      .defineInternal(BrokerHeartbeatIntervalMsProp, INT, Defaults.BrokerHeartbeatIntervalMs, MEDIUM, BrokerHeartbeatIntervalMsDoc)
+      .defineInternal(BrokerSessionTimeoutMsProp, INT, Defaults.BrokerSessionTimeoutMs, MEDIUM, BrokerSessionTimeoutMsDoc)
       .defineInternal(MetadataLogDirProp, STRING, null, null, HIGH, MetadataLogDirDoc)
 
       /************* Authorizer Configuration ***********/
@@ -1497,6 +1506,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)
 
   def requiresZookeeper: Boolean = processRoles.isEmpty
   def usesSelfManagedQuorum: Boolean = processRoles.nonEmpty
diff --git a/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala b/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala
new file mode 100644
index 0000000000000..89368588ddbb4
--- /dev/null
+++ b/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala
@@ -0,0 +1,195 @@
+/**
+ * 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.{ManualMetadataUpdater, Metadata, MockClient}
+import org.apache.kafka.common.{Node, Uuid}
+import org.apache.kafka.common.internals.ClusterResourceListeners
+import org.apache.kafka.common.message.BrokerRegistrationRequestData.{Listener, ListenerCollection}
+import org.apache.kafka.common.message.{BrokerHeartbeatResponseData, BrokerRegistrationResponseData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{BrokerHeartbeatResponse, BrokerRegistrationResponse}
+import org.apache.kafka.common.utils.LogContext
+import org.apache.kafka.metadata.BrokerState
+import org.junit.rules.Timeout
+import org.junit.{Assert, Rule, Test}
+
+class BrokerLifecycleManagerTest {
+  @Rule
+  def globalTimeout = Timeout.millis(120000)
+
+  def configProperties = {
+    val properties = new Properties()
+    properties.setProperty(KafkaConfig.LogDirsProp, "/tmp/foo")
+    properties.setProperty(KafkaConfig.ProcessRolesProp, "broker")
+    properties.setProperty(KafkaConfig.BrokerIdProp, "1")
+    properties.setProperty(KafkaConfig.InitialBrokerRegistrationTimeoutMs, "300000")
+    properties
+  }
+
+  class SimpleControllerNodeProvider extends ControllerNodeProvider {
+    val node = new AtomicReference[Node](null)
+    def controllerNode(): Option[Node] = Option(node.get())
+  }
+
+  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)
+    mockClient.enableBlockingUntilWakeup(10)
+    val metadataUpdater = new ManualMetadataUpdater()
+    val controllerNodeProvider = new SimpleControllerNodeProvider()
+    val channelManager = new BrokerToControllerChannelManager(mockClient,
+      metadataUpdater, controllerNodeProvider, time, 60000, config, "channelManager", None)
+    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))
+    }
+  }
+
+  @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)
+    Assert.assertEquals(BrokerState.NOT_RUNNING, manager.state())
+    manager.start(() => context.highestMetadataOffset.get(),
+      context.channelManager, context.clusterId, context.advertisedListeners,
+      Collections.emptyMap())
+    TestUtils.retry(60000) {
+      Assert.assertEquals(BrokerState.STARTING, manager.state())
+    }
+    manager.close()
+    Assert.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.channelManager, context.clusterId, context.advertisedListeners,
+      Collections.emptyMap())
+    TestUtils.retry(10000) {
+      context.mockClient.wakeup()
+      Assert.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)
+    }
+    newDuplicateRegistrationResponse()
+    Assert.assertEquals(1, context.mockClient.futureResponses().size)
+    manager.start(() => context.highestMetadataOffset.get(),
+      context.channelManager, context.clusterId, context.advertisedListeners,
+      Collections.emptyMap())
+    // We should send the first registration request and get a failure immediately
+    TestUtils.retry(60000) {
+      context.mockClient.wakeup()
+      Assert.assertEquals(0, context.mockClient.futureResponses().size)
+    }
+    // Verify that we resend the registration request.
+    newDuplicateRegistrationResponse()
+    TestUtils.retry(60000) {
+      context.time.sleep(100)
+      context.mockClient.wakeup()
+      manager.eventQueue.wakeup()
+      Assert.assertEquals(0, context.mockClient.futureResponses().size)
+    }
+    // Verify that we time out eventually.
+    context.time.sleep(300000)
+    TestUtils.retry(60000) {
+      context.mockClient.wakeup()
+      manager.eventQueue.wakeup()
+      Assert.assertEquals(BrokerState.SHUTTING_DOWN, manager.state())
+      Assert.assertTrue(manager.initialCatchUpFuture.isCompletedExceptionally())
+      Assert.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.channelManager, context.clusterId, context.advertisedListeners,
+      Collections.emptyMap())
+    TestUtils.retry(10000) {
+      context.mockClient.wakeup()
+      Assert.assertEquals(BrokerState.RECOVERY, manager.state())
+    }
+    context.mockClient.prepareResponseFrom(new BrokerHeartbeatResponse(
+      new BrokerHeartbeatResponseData().setIsFenced(false)), controllerNode)
+    context.time.sleep(20)
+    TestUtils.retry(10000) {
+      context.mockClient.wakeup()
+      Assert.assertEquals(BrokerState.RUNNING, manager.state())
+    }
+    manager.beginControlledShutdown()
+    TestUtils.retry(10000) {
+      context.mockClient.wakeup()
+      Assert.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.mockClient.wakeup()
+      Assert.assertEquals(BrokerState.SHUTTING_DOWN, manager.state())
+    }
+    manager.controlledShutdownFuture.get()
+    manager.close()
+  }
+}
\ No newline at end of file

From 191fb417c943f2ad1e51841e42ba6873b32bef64 Mon Sep 17 00:00:00 2001
From: "Colin P. Mccabe" <cmccabe@confluent.io>
Date: Sat, 6 Feb 2021 15:31:07 -0800
Subject: [PATCH 2/7] adapt PR

---
 .../apache/kafka/common/protocol/Errors.java  |  2 +-
 .../kafka/server/BrokerLifecycleManager.scala | 48 +++++----
 .../main/scala/kafka/server/KafkaConfig.scala |  9 +-
 .../server/BrokerLifecycleManagerTest.scala   | 98 ++++++++++++-------
 4 files changed, 96 insertions(+), 61 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java b/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java
index 00a3c65e1d596..03c1248055878 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java
@@ -354,7 +354,7 @@ public enum Errors {
         "Requested position is not greater than or equal to zero, and less than the size of the snapshot.",
         PositionOutOfRangeException::new),
     UNKNOWN_TOPIC_ID(100, "This server does not host this topic ID.", UnknownTopicIdException::new),
-    DUPLICATE_BROKER_REGISTRATION_EXCEPTION(101, "This broker ID is already in use.", DuplicateBrokerRegistrationException::new);
+    DUPLICATE_BROKER_REGISTRATION(101, "This broker ID is already in use.", DuplicateBrokerRegistrationException::new);
 
     private static final Logger log = LoggerFactory.getLogger(Errors.class);
 
diff --git a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
index efb812a6c898a..4db4c5507d838 100644
--- a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
+++ b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
@@ -20,7 +20,7 @@ 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, RequestCompletionHandler}
+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}
@@ -36,14 +36,14 @@ 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.brokerId}] ")
+  val logContext = new LogContext(s"[BrokerLifecycleManager id=${config.nodeId}] ")
 
   this.logIdent = logContext.logPrefix()
 
   /**
    * The broker id.
    */
-  private val brokerId = config.brokerId
+  private val nodeId = config.nodeId
 
   /**
    * The broker rack, or null if there is no configured rack.
@@ -239,7 +239,7 @@ class BrokerLifecycleManager(val config: KafkaConfig,
         new DeadlineFunction(time.nanoseconds() + initialTimeoutNs),
         new RegistrationTimeoutEvent())
       sendBrokerRegistration()
-      info(s"Incarnation ${incarnationId} of broker ${brokerId} in cluster ${clusterId} " +
+      info(s"Incarnation ${incarnationId} of broker ${nodeId} in cluster ${clusterId} " +
         "is now STARTING.")
     }
   }
@@ -253,7 +253,7 @@ class BrokerLifecycleManager(val config: KafkaConfig,
         setMaxSupportedVersion(range.max()))
     }
     val data = new BrokerRegistrationRequestData().
-        setBrokerId(brokerId).
+        setBrokerId(nodeId).
         setClusterId(_clusterId).
         setFeatures(features).
         setIncarnationId(incarnationId).
@@ -266,21 +266,21 @@ class BrokerLifecycleManager(val config: KafkaConfig,
       new BrokerRegistrationResponseHandler())
   }
 
-  class BrokerRegistrationResponseHandler extends RequestCompletionHandler {
+  class BrokerRegistrationResponseHandler extends ControllerRequestCompletionHandler {
     override def onComplete(response: ClientResponse): Unit = {
       if (response.authenticationException() != null) {
-        error(s"Unable to register broker ${brokerId} because of an authentication exception.",
+        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 ${brokerId} because of an API version problem.",
+        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 ${brokerId}.")
+        warn(s"Unable to register broker ${nodeId}.")
         scheduleNextCommunicationAfterFailure()
       } else if (!response.responseBody().isInstanceOf[BrokerRegistrationResponse]) {
-        error(s"Unable to register broker ${brokerId} because the controller returned an " +
+        error(s"Unable to register broker ${nodeId} because the controller returned an " +
           "invalid response type.")
         scheduleNextCommunicationAfterFailure()
       } else {
@@ -291,22 +291,27 @@ class BrokerLifecycleManager(val config: KafkaConfig,
           _brokerEpoch = message.data().brokerEpoch()
           registered = true
           initialRegistrationSucceeded = true
-          info(s"Successfully registered broker ${brokerId} with broker epoch ${_brokerEpoch}")
+          info(s"Successfully registered broker ${nodeId} with broker epoch ${_brokerEpoch}")
           scheduleNextCommunicationImmediately() // Immediately send a heartbeat
         } else {
-          info(s"Unable to register broker ${brokerId} because the controller returned " +
+          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(brokerId).
+      setBrokerId(nodeId).
       setCurrentMetadataOffset(metadataOffset).
       setWantFence(!readyToUnfence).
       setWantShutDown(_state == BrokerState.PENDING_CONTROLLED_SHUTDOWN)
@@ -317,21 +322,21 @@ class BrokerLifecycleManager(val config: KafkaConfig,
       new BrokerHeartbeatResponseHandler())
   }
 
-  class BrokerHeartbeatResponseHandler extends RequestCompletionHandler {
+  class BrokerHeartbeatResponseHandler extends ControllerRequestCompletionHandler {
     override def onComplete(response: ClientResponse): Unit = {
       if (response.authenticationException() != null) {
-        error(s"Unable to send broker heartbeat for ${brokerId} because of an " +
+        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 ${brokerId} because of an API " +
+        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 ${brokerId}. Retrying.")
+        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 ${brokerId} because the controller " +
+        error(s"Unable to send broker heartbeat for ${nodeId} because the controller " +
           "returned an invalid response type.")
         scheduleNextCommunicationAfterFailure()
       } else {
@@ -387,11 +392,16 @@ class BrokerLifecycleManager(val config: KafkaConfig,
               scheduleNextCommunicationAfterSuccess()
           }
         } else {
-          warn(s"Broker ${brokerId} sent a heartbeat request but received error ${errorCode}.")
+          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)
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 03c7818441bff..9e2898e557da7 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -665,6 +665,9 @@ object KafkaConfig {
   val ProcessRolesDoc = "The roles that this process plays: 'broker', 'controller', or 'broker,controller' if it is both. " +
     "This configuration is only for clusters upgraded for KIP-500, which replaces the dependence on Zookeeper with " +
     "a self-managed Raft quorum. Leave this config undefined or empty for Zookeeper clusters."
+  val InitialBrokerRegistrationTimeoutMsDoc = "When initially registering with the controller quorum, the number of milliseconds to wait before declaring failure and exiting the broker process."
+  val BrokerHeartbeatIntervalMsDoc = "The length of time in milliseconds between broker heartbeats. Used when running in KIP-500 mode."
+  val BrokerSessionTimeoutMsDoc = "The length of time in milliseconds that a broker lease lasts if no heartbeats are made. Used when running in KIP-500 mode."
   val NodeIdDoc = "The node ID associated with the roles this process is playing when `process.roles` is non-empty. " +
     "This is required configuration when the self-managed quorum is enabled."
   val MetadataLogDirDoc = "This configuration determines where we put the metadata log for clusters upgraded to " +
@@ -1070,9 +1073,9 @@ object KafkaConfig {
        */
       .defineInternal(ProcessRolesProp, LIST, Collections.emptyList(), ValidList.in("broker", "controller"), HIGH, ProcessRolesDoc)
       .defineInternal(NodeIdProp, INT, Defaults.EmptyNodeId, null, HIGH, NodeIdDoc)
-      .defineInternal(InitialBrokerRegistrationTimeoutMs, INT, Defaults.InitialBrokerRegistrationTimeoutMs, MEDIUM, InitialBrokerRegistrationTimeoutMsDoc)
-      .defineInternal(BrokerHeartbeatIntervalMsProp, INT, Defaults.BrokerHeartbeatIntervalMs, MEDIUM, BrokerHeartbeatIntervalMsDoc)
-      .defineInternal(BrokerSessionTimeoutMsProp, INT, Defaults.BrokerSessionTimeoutMs, MEDIUM, BrokerSessionTimeoutMsDoc)
+      .defineInternal(InitialBrokerRegistrationTimeoutMs, INT, Defaults.InitialBrokerRegistrationTimeoutMs, null, MEDIUM, InitialBrokerRegistrationTimeoutMsDoc)
+      .defineInternal(BrokerHeartbeatIntervalMsProp, INT, Defaults.BrokerHeartbeatIntervalMs, null, MEDIUM, BrokerHeartbeatIntervalMsDoc)
+      .defineInternal(BrokerSessionTimeoutMsProp, INT, Defaults.BrokerSessionTimeoutMs, null, MEDIUM, BrokerSessionTimeoutMsDoc)
       .defineInternal(MetadataLogDirProp, STRING, null, null, HIGH, MetadataLogDirDoc)
 
       /************* Authorizer Configuration ***********/
diff --git a/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala b/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala
index 89368588ddbb4..829c6402b5d7e 100644
--- a/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala
@@ -21,34 +21,43 @@ import java.util.{Collections, Properties}
 import java.util.concurrent.atomic.{AtomicLong, AtomicReference}
 
 import kafka.utils.{MockTime, TestUtils}
-import org.apache.kafka.clients.{ManualMetadataUpdater, Metadata, MockClient}
+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.rules.Timeout
-import org.junit.{Assert, Rule, Test}
+import org.junit.jupiter.api.{Assertions, Test, Timeout}
+
+import scala.jdk.CollectionConverters._
 
-class BrokerLifecycleManagerTest {
-  @Rule
-  def globalTimeout = Timeout.millis(120000)
 
+@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.BrokerIdProp, "1")
+    properties.setProperty(KafkaConfig.NodeIdProp, "1")
     properties.setProperty(KafkaConfig.InitialBrokerRegistrationTimeoutMs, "300000")
     properties
   }
 
   class SimpleControllerNodeProvider extends ControllerNodeProvider {
     val node = new AtomicReference[Node](null)
-    def controllerNode(): Option[Node] = Option(node.get())
+
+    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) {
@@ -57,11 +66,13 @@ class BrokerLifecycleManagerTest {
     val highestMetadataOffset = new AtomicLong(0)
     val metadata = new Metadata(1000, 1000, new LogContext(), new ClusterResourceListeners())
     val mockClient = new MockClient(time, metadata)
-    mockClient.enableBlockingUntilWakeup(10)
-    val metadataUpdater = new ManualMetadataUpdater()
     val controllerNodeProvider = new SimpleControllerNodeProvider()
-    val channelManager = new BrokerToControllerChannelManager(mockClient,
-      metadataUpdater, controllerNodeProvider, time, 60000, config, "channelManager", None)
+    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 =>
@@ -70,6 +81,11 @@ class BrokerLifecycleManagerTest {
         setPort(ep.port.shortValue()).
         setSecurityProtocol(ep.securityProtocol.id))
     }
+
+    def poll(): Unit = {
+      mockClient.wakeup()
+      mockChannelManager.poll()
+    }
   }
 
   @Test
@@ -83,15 +99,15 @@ class BrokerLifecycleManagerTest {
   def testCreateStartAndClose(): Unit = {
     val context = new BrokerLifecycleManagerTestContext(configProperties)
     val manager = new BrokerLifecycleManager(context.config, context.time, None)
-    Assert.assertEquals(BrokerState.NOT_RUNNING, manager.state())
+    Assertions.assertEquals(BrokerState.NOT_RUNNING, manager.state())
     manager.start(() => context.highestMetadataOffset.get(),
-      context.channelManager, context.clusterId, context.advertisedListeners,
+      context.mockChannelManager, context.clusterId, context.advertisedListeners,
       Collections.emptyMap())
     TestUtils.retry(60000) {
-      Assert.assertEquals(BrokerState.STARTING, manager.state())
+      Assertions.assertEquals(BrokerState.STARTING, manager.state())
     }
     manager.close()
-    Assert.assertEquals(BrokerState.SHUTTING_DOWN, manager.state())
+    Assertions.assertEquals(BrokerState.SHUTTING_DOWN, manager.state())
   }
 
   @Test
@@ -103,13 +119,14 @@ class BrokerLifecycleManagerTest {
     context.mockClient.prepareResponseFrom(new BrokerRegistrationResponse(
       new BrokerRegistrationResponseData().setBrokerEpoch(1000)), controllerNode)
     manager.start(() => context.highestMetadataOffset.get(),
-      context.channelManager, context.clusterId, context.advertisedListeners,
+      context.mockChannelManager, context.clusterId, context.advertisedListeners,
       Collections.emptyMap())
     TestUtils.retry(10000) {
-      context.mockClient.wakeup()
-      Assert.assertEquals(1000L, manager.brokerEpoch())
+      context.poll()
+      Assertions.assertEquals(1000L, manager.brokerEpoch())
     }
     manager.close()
+
   }
 
   @Test
@@ -122,33 +139,34 @@ class BrokerLifecycleManagerTest {
       context.mockClient.prepareResponseFrom(new BrokerRegistrationResponse(
         new BrokerRegistrationResponseData().
           setErrorCode(Errors.DUPLICATE_BROKER_REGISTRATION.code())), controllerNode)
+      context.mockChannelManager.poll()
     }
     newDuplicateRegistrationResponse()
-    Assert.assertEquals(1, context.mockClient.futureResponses().size)
+    Assertions.assertEquals(1, context.mockClient.futureResponses().size)
     manager.start(() => context.highestMetadataOffset.get(),
-      context.channelManager, context.clusterId, context.advertisedListeners,
+      context.mockChannelManager, context.clusterId, context.advertisedListeners,
       Collections.emptyMap())
     // We should send the first registration request and get a failure immediately
     TestUtils.retry(60000) {
-      context.mockClient.wakeup()
-      Assert.assertEquals(0, context.mockClient.futureResponses().size)
+      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.mockClient.wakeup()
+      context.poll()
       manager.eventQueue.wakeup()
-      Assert.assertEquals(0, context.mockClient.futureResponses().size)
+      Assertions.assertEquals(0, context.mockClient.futureResponses().size)
     }
     // Verify that we time out eventually.
     context.time.sleep(300000)
     TestUtils.retry(60000) {
-      context.mockClient.wakeup()
+      context.poll()
       manager.eventQueue.wakeup()
-      Assert.assertEquals(BrokerState.SHUTTING_DOWN, manager.state())
-      Assert.assertTrue(manager.initialCatchUpFuture.isCompletedExceptionally())
-      Assert.assertEquals(-1L, manager.brokerEpoch())
+      Assertions.assertEquals(BrokerState.SHUTTING_DOWN, manager.state())
+      Assertions.assertTrue(manager.initialCatchUpFuture.isCompletedExceptionally())
+      Assertions.assertEquals(-1L, manager.brokerEpoch())
     }
     manager.close()
   }
@@ -164,30 +182,34 @@ class BrokerLifecycleManagerTest {
     context.mockClient.prepareResponseFrom(new BrokerHeartbeatResponse(
       new BrokerHeartbeatResponseData().setIsCaughtUp(true)), controllerNode)
     manager.start(() => context.highestMetadataOffset.get(),
-      context.channelManager, context.clusterId, context.advertisedListeners,
+      context.mockChannelManager, context.clusterId, context.advertisedListeners,
       Collections.emptyMap())
     TestUtils.retry(10000) {
-      context.mockClient.wakeup()
-      Assert.assertEquals(BrokerState.RECOVERY, manager.state())
+      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.mockClient.wakeup()
-      Assert.assertEquals(BrokerState.RUNNING, manager.state())
+      context.poll()
+      manager.eventQueue.wakeup()
+      Assertions.assertEquals(BrokerState.RUNNING, manager.state())
     }
     manager.beginControlledShutdown()
     TestUtils.retry(10000) {
-      context.mockClient.wakeup()
-      Assert.assertEquals(BrokerState.PENDING_CONTROLLED_SHUTDOWN, manager.state())
+      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.mockClient.wakeup()
-      Assert.assertEquals(BrokerState.SHUTTING_DOWN, manager.state())
+      context.poll()
+      manager.eventQueue.wakeup()
+      Assertions.assertEquals(BrokerState.SHUTTING_DOWN, manager.state())
     }
     manager.controlledShutdownFuture.get()
     manager.close()

From 33b60ba1aaa671a569214359fac3b44894e0e078 Mon Sep 17 00:00:00 2001
From: "Colin P. Mccabe" <cmccabe@confluent.io>
Date: Wed, 10 Feb 2021 10:23:06 -0800
Subject: [PATCH 3/7] Improve JavaDoc

---
 .../kafka/server/BrokerLifecycleManager.scala | 68 ++++++++++++-------
 1 file changed, 44 insertions(+), 24 deletions(-)

diff --git a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
index 4db4c5507d838..36250c5bc2583 100644
--- a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
+++ b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
@@ -18,7 +18,7 @@ package kafka.server
 
 import java.util
 import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
-import java.util.concurrent.{CompletableFuture, TimeUnit}
+import java.util.concurrent.CompletableFuture
 import kafka.utils.Logging
 import org.apache.kafka.clients.ClientResponse
 import org.apache.kafka.common.Uuid
@@ -33,6 +33,24 @@ import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
 import scala.jdk.CollectionConverters._
 
 
+/**
+ * The broker lifecycle manager owns the broker state.
+ *
+ * Its inputs are messages passed in from other parts of the broker and from the
+ * controller: requests to start up, or shut down, for example. Its output are the broker
+ * state and various futures that can be used to wait for broker state transitions to
+ * occur.
+ *
+ * The lifecycle manager handles registering the broker with the controller, as described
+ * in KIP-631. After registration is complete, it handles sending periodic broker
+ * heartbeats and processing the responses.
+ *
+ * This code uses an event queue paradigm. Modifications get translated into events, which
+ * are placed on the queue to be processed sequentially. As described in the JavaDoc for
+ * each variable, most mutable state can be accessed only from that event queue thread.
+ * In some cases we expose a volatile variable which can be read from any thread, but only
+ * written from the event queue thread.
+ */
 class BrokerLifecycleManager(val config: KafkaConfig,
                              val time: Time,
                              val threadNamePrefix: Option[String]) extends Logging {
@@ -53,17 +71,17 @@ class BrokerLifecycleManager(val config: KafkaConfig,
   /**
    * How long to wait for registration to succeed before failing the startup process.
    */
-  private val initialTimeoutNs = NANOSECONDS.
-    convert(config.initialRegistrationTimeoutMs.longValue(), TimeUnit.MILLISECONDS)
+  private val initialTimeoutNs = MILLISECONDS.toNanos(config.initialRegistrationTimeoutMs.longValue())
 
   /**
    * The exponential backoff to use for resending communication.
    */
   private val resendExponentialBackoff =
-    new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.1)
+    new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.02)
 
   /**
-   * The number of tries we've tried to communicate.
+   * The number of tries we've tried to communicate.  This variable can only be read or
+   * written from the event queue thread.
    */
   private var failedAttempts = 0L
 
@@ -84,66 +102,68 @@ class BrokerLifecycleManager(val config: KafkaConfig,
   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.
+   * 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.
+   * 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.
+   * A thread-safe callback function which gives this manager the current highest metadata
+   * offset.  This variable can only be read or written from the event queue thread.
    */
   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.
+   * True only if we are ready to unfence the broker.  This variable can only be read or
+   * written from the event queue thread.
    */
   private var readyToUnfence = false
 
   /**
-   * True if we sent a heartbeat to the active controller requesting controlled
-   * shutdown.
+   * True if we sent a event queue to the active controller requesting controlled
+   * shutdown.  This variable can only be read or written from the event queue thread.
    */
   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.
+   * This variable can only be read or written 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.
+   * True if the initial registration succeeded.  This variable can only be read or
+   * written 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.
+   * The cluster ID, or null if this manager has not been started yet.  This variable can
+   * only be read or written from the event queue thread.
    */
   private var _clusterId: Uuid = null
 
   /**
-   * The listeners which this broker advertises.
+   * The listeners which this broker advertises.  This variable can only be read or
+   * written from the event queue thread.
    */
   private var _advertisedListeners: ListenerCollection = null
 
   /**
-   * The features supported by this broker.
+   * The features supported by this broker.  This variable can only be read or written
+   * from the event queue thread.
    */
   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.
+   * The channel manager, or null if this manager has not been started yet.  This variable
+   * can only be read or written from the event queue thread.
    */
   var _channelManager: BrokerToControllerChannelManager = null
 

From 42b18dce1b5609f0c50bb2791fc915b7bf0bcf7f Mon Sep 17 00:00:00 2001
From: "Colin P. Mccabe" <cmccabe@confluent.io>
Date: Wed, 10 Feb 2021 10:31:45 -0800
Subject: [PATCH 4/7] make event classes private

---
 .../kafka/server/BrokerLifecycleManager.scala    | 16 ++++++++--------
 1 file changed, 8 insertions(+), 8 deletions(-)

diff --git a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
index 36250c5bc2583..ceec40ba55df6 100644
--- a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
+++ b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
@@ -196,7 +196,7 @@ class BrokerLifecycleManager(val config: KafkaConfig,
 
   def state(): BrokerState = _state
 
-  class BeginControlledShutdownEvent extends EventQueue.Event {
+  private class BeginControlledShutdownEvent extends EventQueue.Event {
     override def run(): Unit = {
       _state match {
         case BrokerState.PENDING_CONTROLLED_SHUTDOWN =>
@@ -235,14 +235,14 @@ class BrokerLifecycleManager(val config: KafkaConfig,
     eventQueue.close()
   }
 
-  class SetReadyToUnfenceEvent() extends EventQueue.Event {
+  private class SetReadyToUnfenceEvent() extends EventQueue.Event {
     override def run(): Unit = {
       readyToUnfence = true
       scheduleNextCommunicationImmediately()
     }
   }
 
-  class StartupEvent(highestMetadataOffsetProvider: () => Long,
+  private class StartupEvent(highestMetadataOffsetProvider: () => Long,
                      channelManager: BrokerToControllerChannelManager,
                      clusterId: Uuid,
                      advertisedListeners: ListenerCollection,
@@ -286,7 +286,7 @@ class BrokerLifecycleManager(val config: KafkaConfig,
       new BrokerRegistrationResponseHandler())
   }
 
-  class BrokerRegistrationResponseHandler extends ControllerRequestCompletionHandler {
+  private 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.",
@@ -342,7 +342,7 @@ class BrokerLifecycleManager(val config: KafkaConfig,
       new BrokerHeartbeatResponseHandler())
   }
 
-  class BrokerHeartbeatResponseHandler extends ControllerRequestCompletionHandler {
+  private 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 " +
@@ -446,7 +446,7 @@ class BrokerLifecycleManager(val config: KafkaConfig,
       new CommunicationEvent())
   }
 
-  class RegistrationTimeoutEvent extends EventQueue.Event {
+  private class RegistrationTimeoutEvent extends EventQueue.Event {
     override def run(): Unit = {
       if (!initialRegistrationSucceeded) {
         error("Shutting down because we were unable to register with the controller quorum.")
@@ -455,7 +455,7 @@ class BrokerLifecycleManager(val config: KafkaConfig,
     }
   }
 
-  class CommunicationEvent extends EventQueue.Event {
+  private class CommunicationEvent extends EventQueue.Event {
     override def run(): Unit = {
       if (registered) {
         sendBrokerHeartbeat()
@@ -465,7 +465,7 @@ class BrokerLifecycleManager(val config: KafkaConfig,
     }
   }
 
-  class ShutdownEvent extends EventQueue.Event {
+  private class ShutdownEvent extends EventQueue.Event {
     override def run(): Unit = {
       info(s"Transitioning from ${_state} to ${BrokerState.SHUTTING_DOWN}.")
       _state = BrokerState.SHUTTING_DOWN

From eeef3048d574cd5661a0efade3693caf99e0c7f7 Mon Sep 17 00:00:00 2001
From: "Colin P. Mccabe" <cmccabe@confluent.io>
Date: Wed, 10 Feb 2021 10:40:27 -0800
Subject: [PATCH 5/7] KafkaConfig: add types to vals

---
 core/src/main/scala/kafka/server/KafkaConfig.scala | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 9e2898e557da7..0ecb48cd23cce 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -1509,9 +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)
+  val initialRegistrationTimeoutMs: Int = getInt(KafkaConfig.InitialBrokerRegistrationTimeoutMs)
+  val brokerHeartbeatIntervalMs: Int = getInt(KafkaConfig.BrokerHeartbeatIntervalMsProp)
+  val brokerSessionTimeoutMs: Int = getInt(KafkaConfig.BrokerSessionTimeoutMsProp)
 
   def requiresZookeeper: Boolean = processRoles.isEmpty
   def usesSelfManagedQuorum: Boolean = processRoles.nonEmpty

From 92ddccb9fccb0108523e246c470979357fe6ece4 Mon Sep 17 00:00:00 2001
From: "Colin P. Mccabe" <cmccabe@confluent.io>
Date: Wed, 10 Feb 2021 10:40:35 -0800
Subject: [PATCH 6/7] BrokerLifecycleManager: check for null channel manager in
 shutdown

---
 .../src/main/scala/kafka/server/BrokerLifecycleManager.scala | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
index ceec40ba55df6..9bf546ec4bea2 100644
--- a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
+++ b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
@@ -471,7 +471,10 @@ class BrokerLifecycleManager(val config: KafkaConfig,
       _state = BrokerState.SHUTTING_DOWN
       controlledShutdownFuture.complete(null)
       initialCatchUpFuture.cancel(false)
-      _channelManager.shutdown()
+      if (_channelManager != null) {
+        _channelManager.shutdown()
+        _channelManager = null
+      }
     }
   }
 }

From 0fc50f6e3b260837d51a5477e6d19b20fed8a86b Mon Sep 17 00:00:00 2001
From: "Colin P. Mccabe" <cmccabe@confluent.io>
Date: Wed, 10 Feb 2021 11:00:36 -0800
Subject: [PATCH 7/7] address review comments

---
 .../kafka/server/BrokerLifecycleManager.scala | 33 ++++++++++---------
 .../server/BrokerLifecycleManagerTest.scala   |  2 +-
 2 files changed, 18 insertions(+), 17 deletions(-)

diff --git a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
index 9bf546ec4bea2..64c92005cf7a4 100644
--- a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
+++ b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
@@ -66,12 +66,13 @@ class BrokerLifecycleManager(val config: KafkaConfig,
   /**
    * The broker rack, or null if there is no configured rack.
    */
-  private val rack = config.rack.orNull
+  private val rack = config.rack
 
   /**
    * How long to wait for registration to succeed before failing the startup process.
    */
-  private val initialTimeoutNs = MILLISECONDS.toNanos(config.initialRegistrationTimeoutMs.longValue())
+  private val initialTimeoutNs =
+    MILLISECONDS.toNanos(config.initialRegistrationTimeoutMs.longValue())
 
   /**
    * The exponential backoff to use for resending communication.
@@ -80,8 +81,8 @@ class BrokerLifecycleManager(val config: KafkaConfig,
     new ExponentialBackoff(100, 2, config.brokerSessionTimeoutMs.toLong, 0.02)
 
   /**
-   * The number of tries we've tried to communicate.  This variable can only be read or
-   * written from the event queue thread.
+   * The number of times we've tried and failed to communicate.  This variable can only be
+   * read or written from the event queue thread.
    */
   private var failedAttempts = 0L
 
@@ -117,7 +118,7 @@ class BrokerLifecycleManager(val config: KafkaConfig,
    * A thread-safe callback function which gives this manager the current highest metadata
    * offset.  This variable can only be read or written from the event queue thread.
    */
-  private var _highestMetadataOffsetProvider: () => Long = null
+  private var _highestMetadataOffsetProvider: () => Long = _
 
   /**
    * True only if we are ready to unfence the broker.  This variable can only be read or
@@ -132,7 +133,7 @@ class BrokerLifecycleManager(val config: KafkaConfig,
   private var gotControlledShutdownResponse = false
 
   /**
-   * Whether or not we this broker is registered with the controller quorum.
+   * Whether or not this broker is registered with the controller quorum.
    * This variable can only be read or written from the event queue thread.
    */
   private var registered = false
@@ -147,25 +148,25 @@ class BrokerLifecycleManager(val config: KafkaConfig,
    * The cluster ID, or null if this manager has not been started yet.  This variable can
    * only be read or written from the event queue thread.
    */
-  private var _clusterId: Uuid = null
+  private var _clusterId: Uuid = _
 
   /**
    * The listeners which this broker advertises.  This variable can only be read or
    * written from the event queue thread.
    */
-  private var _advertisedListeners: ListenerCollection = null
+  private var _advertisedListeners: ListenerCollection = _
 
   /**
    * The features supported by this broker.  This variable can only be read or written
    * from the event queue thread.
    */
-  private var _supportedFeatures: util.Map[String, VersionRange] = null
+  private var _supportedFeatures: util.Map[String, VersionRange] = _
 
   /**
    * The channel manager, or null if this manager has not been started yet.  This variable
    * can only be read or written from the event queue thread.
    */
-  var _channelManager: BrokerToControllerChannelManager = null
+  var _channelManager: BrokerToControllerChannelManager = _
 
   /**
    * The event queue.
@@ -200,10 +201,10 @@ class BrokerLifecycleManager(val config: KafkaConfig,
     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.")
+          info("Attempted to enter pending controlled shutdown state, but we are " +
+          "already in that state.")
         case BrokerState.RUNNING =>
-          info(s"Beginning controlled shutdown.")
+          info("Beginning controlled shutdown.")
           _state = BrokerState.PENDING_CONTROLLED_SHUTDOWN
         case _ =>
           info(s"Skipping controlled shutdown because we are in state ${_state}.")
@@ -253,8 +254,8 @@ class BrokerLifecycleManager(val config: KafkaConfig,
       _channelManager.start()
       _state = BrokerState.STARTING
       _clusterId = clusterId
-      _advertisedListeners = advertisedListeners
-      _supportedFeatures = supportedFeatures
+      _advertisedListeners = advertisedListeners.duplicate()
+      _supportedFeatures = new util.HashMap[String, VersionRange](supportedFeatures)
       eventQueue.scheduleDeferred("initialRegistrationTimeout",
         new DeadlineFunction(time.nanoseconds() + initialTimeoutNs),
         new RegistrationTimeoutEvent())
@@ -278,7 +279,7 @@ class BrokerLifecycleManager(val config: KafkaConfig,
         setFeatures(features).
         setIncarnationId(incarnationId).
         setListeners(_advertisedListeners).
-        setRack(rack)
+        setRack(rack.orNull)
     if (isTraceEnabled) {
       trace(s"Sending broker registration ${data}")
     }
diff --git a/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala b/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala
index 829c6402b5d7e..a823ce63bed87 100644
--- a/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala
@@ -214,4 +214,4 @@ class BrokerLifecycleManagerTest {
     manager.controlledShutdownFuture.get()
     manager.close()
   }
-}
\ No newline at end of file
+}