Fix NPE in networkListeners from BrokerServer
Reviewers: Chia-Ping Tsai
@@ -21,6 +21,7 @@ import java.util
import java.util.concurrent.{CompletableFuture, TimeUnit, TimeoutException}
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.locks.ReentrantLock
+import java.net.InetAddress
import kafka.cluster.Broker.ServerInfo
import kafka.coordinator.group.GroupCoordinator
@@ -39,7 +40,7 @@ import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.security.auth.SecurityProtocol
import org.apache.kafka.common.security.scram.internals.ScramMechanism
import org.apache.kafka.common.security.token.delegation.internals.DelegationTokenCache
-import org.apache.kafka.common.utils.{AppInfoParser, LogContext, Time}
+import org.apache.kafka.common.utils.{AppInfoParser, LogContext, Time, Utils}
import org.apache.kafka.common.{ClusterResource, Endpoint, KafkaException}
import org.apache.kafka.metadata.{BrokerState, VersionRange}
import org.apache.kafka.metalog.MetaLogManager
@@ -272,7 +273,7 @@ class BrokerServer(
val networkListeners = new ListenerCollection()
config.advertisedListeners.foreach { ep =>
networkListeners.add(new Listener().
- setHost(ep.host).
+ setHost(if (Utils.isBlank(ep.host)) InetAddress.getLocalHost.getCanonicalHostName else ep.host).
setName(ep.listenerName.value()).
setPort(socketServer.boundPort(ep.listenerName)).
setSecurityProtocol(ep.securityProtocol.id))
