KAFKA-18758: NullPointerException in shutdown following InvalidConfig…

…urationException (#18833)

* KAFKA-18758:  NullPointerException in shutdown following InvalidConfigurationException

Add checks for null in shutdown as BrokerLifecycleManager is not instantiaited if LogManager constructor throws an Exception

core/src/main/scala/kafka/server/BrokerServer.scala

@@ -748,18 +748,22 @@ class BrokerServer(
         if (replicaManager != null)
           replicaManager.beginControlledShutdown()
 
-        lifecycleManager.beginControlledShutdown()
-        try {
-          val controlledShutdownTimeoutMs = deadline - time.milliseconds()
-          lifecycleManager.controlledShutdownFuture.get(controlledShutdownTimeoutMs, TimeUnit.MILLISECONDS)
-        } catch {
-          case _: TimeoutException =>
-            error("Timed out waiting for the controller to approve controlled shutdown")
-          case e: Throwable =>
-            error("Got unexpected exception waiting for controlled shutdown future", e)
+        if (lifecycleManager != null) {
+          lifecycleManager.beginControlledShutdown()
+          try {
+            val controlledShutdownTimeoutMs = deadline - time.milliseconds()
+            lifecycleManager.controlledShutdownFuture.get(controlledShutdownTimeoutMs, TimeUnit.MILLISECONDS)
+          } catch {
+            case _: TimeoutException =>
+              error("Timed out waiting for the controller to approve controlled shutdown")
+            case e: Throwable =>
+              error("Got unexpected exception waiting for controlled shutdown future", e)
+          }
         }
       }
-      lifecycleManager.beginShutdown()
+      if (lifecycleManager != null)
+        lifecycleManager.beginShutdown()
+
       // Stop socket server to stop accepting any more connections and requests.
       // Socket server will be shutdown towards the end of the sequence.
       if (socketServer != null) {
@@ -811,8 +815,10 @@ class BrokerServer(
       if (clientToControllerChannelManager != null)
         CoreUtils.swallow(clientToControllerChannelManager.shutdown(), this)
 
-      if (logManager != null)
-        CoreUtils.swallow(logManager.shutdown(lifecycleManager.brokerEpoch), this)
+      if (logManager != null) {
+        val brokerEpoch = if (lifecycleManager != null) lifecycleManager.brokerEpoch else -1
+        CoreUtils.swallow(logManager.shutdown(brokerEpoch), this)
+      }
 
       // Close remote log manager to give a chance to any of its underlying clients
       // (especially in RemoteStorageManager and RemoteLogMetadataManager) to close gracefully.
@@ -832,7 +838,9 @@ class BrokerServer(
 
       isShuttingDown.set(false)
 
-      CoreUtils.swallow(lifecycleManager.close(), this)
+      if (lifecycleManager != null)
+        CoreUtils.swallow(lifecycleManager.close(), this)
+
       CoreUtils.swallow(config.dynamicConfig.clear(), this)
       Utils.closeQuietly(clientMetricsManager, "client metrics manager")
       sharedServer.stopForBroker()