Repository: apache/kafka
File: core/src/main/scala/kafka/server/BrokerServer.scala (Scala)

Code before:
      // Enable inbound TCP connections. Each endpoint will be started only once its matching
      // authorizer future is completed.
      val endpointReadyFutures = {
        val builder = new EndpointReadyFutures.Builder()
        builder.build(authorizerPlugin.toJava,
          new KafkaAuthorizerServerInfo(
            new ClusterResource(clusterId),
            config.nodeId,
            listenerInfo.listeners().values(),
            listenerInfo.firstListener(),
            config.earlyStartListeners.map(_.value()).asJava))
      }
      val authorizerFutures = endpointReadyFutures.futures().asScala.toMap
      val enableRequestProcessingFuture = socketServer.enableRequestProcessing(authorizerFutures)

      // Block here until all the authorizer futures are complete.
      FutureUtils.waitWithLogging(logger.underlying, logIdent,
        "all of the authorizer futures to be completed",
        CompletableFuture.allOf(authorizerFutures.values.toSeq: _*), startupDeadline, time)

      // Wait for all the SocketServer ports to be open, and the Acceptors to be started.
      FutureUtils.waitWithLogging(logger.underlying, logIdent,
        "all of the SocketServer Acceptors to be started",
        enableRequestProcessingFuture, startupDeadline, time)

      brokerReadyCallbacks.foreach { callback =>
        try {
          callback.onBrokerReady()
        } catch {
          case e: Exception =>
            error(s"Error executing broker ready callback: ${callback.getClass.getSimpleName}", e)
        }
      }

      maybeChangeStatus(STARTING, STARTED)
    } catch {
      case e: Throwable =>
        maybeChangeStatus(STARTING, STARTED)
        fatal("Fatal error during broker startup. Prepare to shutdown", e)
        shutdown()
        throw if (e.isInstanceOf[ExecutionException]) e.getCause else e
    }
  }

  def registerBrokerReadyCallback(callback: BrokerReadyCallback): Unit = {
    brokerReadyCallbacks += callback
  }

  private def createGroupCoordinator(): GroupCoordinator = {
    // Create group coordinator, but don't start it until we've started replica manager.
    // Hardcode Time.SYSTEM for now as some Streams tests fail otherwise, it would be good

Review comment (suggestion):
Is the `brokerReadyCallbacks` collection necessary? Would you mind reviewing the code to see if it can be simplified? for example:

```scala
      remoteLogManagerOpt.foreach(rlm =>
        rlm.remoteLogMetadataManager() match {
          case callback: BrokerReadyCallback =>
            try callback.onBrokerReady()
            catch {
              case e: Exception =>
                error(s"Error executing broker ready callback: ${callback.getClass.getSimpleName}", e)
            }
          case _ => // Skip registration
        }
      )
```

Code after:
            listenerInfo.listeners().values(),
            listenerInfo.firstListener(),
            config.earlyStartListeners.map(_.value()).asJava))
      }
      val authorizerFutures = endpointReadyFutures.futures().asScala.toMap
      val enableRequestProcessingFuture = socketServer.enableRequestProcessing(authorizerFutures)

      // Block here until all the authorizer futures are complete.
      FutureUtils.waitWithLogging(logger.underlying, logIdent,
        "all of the authorizer futures to be completed",
        CompletableFuture.allOf(authorizerFutures.values.toSeq: _*), startupDeadline, time)

      // Wait for all the SocketServer ports to be open, and the Acceptors to be started.
      FutureUtils.waitWithLogging(logger.underlying, logIdent,
        "all of the SocketServer Acceptors to be started",
        enableRequestProcessingFuture, startupDeadline, time)

      remoteLogManagerOpt.foreach(rlm =>
        rlm.remoteLogMetadataManager() match {
          case callback: BrokerReadyCallback =>
            try {
              callback.onBrokerReady()
            } catch {
              case e: Exception =>
                error(s"Error executing broker ready callback: ${callback.getClass.getSimpleName}", e)
            }
          case _ => // Skip
        }
      )

      maybeChangeStatus(STARTING, STARTED)
    } catch {
      case e: Throwable =>
        maybeChangeStatus(STARTING, STARTED)
        fatal("Fatal error during broker startup. Prepare to shutdown", e)
        shutdown()
        throw if (e.isInstanceOf[ExecutionException]) e.getCause else e
    }
  }

  private def createGroupCoordinator(): GroupCoordinator = {
    // Create group coordinator, but don't start it until we've started replica manager.
    // Hardcode Time.SYSTEM for now as some Streams tests fail otherwise, it would be good
    // to fix the underlying issue.
    val time = Time.SYSTEM
    val serde = new GroupCoordinatorRecordSerde
    val timer = new SystemTimerReaper(
      "group-coordinator-reaper",
      new SystemTimer("group-coordinator")
    )
    val loader = new CoordinatorLoaderImpl[CoordinatorRecord](