You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by js...@apache.org on 2025/02/25 02:51:34 UTC
(kafka) branch trunk updated: KAFKA-18281: Kafka is improperly validating non-advertised listeners for routable controller addresses (#18387)
This is an automated email from the ASF dual-hosted git repository.
jsancio pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 10873e42101 KAFKA-18281: Kafka is improperly validating non-advertised listeners for routable controller addresses (#18387)
10873e42101 is described below
commit 10873e42101671e3e917f7d20bdd99870a700564
Author: PoAn Yang <pa...@apache.org>
AuthorDate: Tue Feb 25 10:51:28 2025 +0800
KAFKA-18281: Kafka is improperly validating non-advertised listeners for routable controller addresses (#18387)
When a cluster is configured with a dynamic controller quorum, KRaft replica's endpoint are computed using the advertised.listeners property and not the quorum.controller.voters property. This change in the configuration makes it difficult to keeping all previous node configurations compatible with the new endpoint discovery functionality.
The least intrusive solution is to rely on Kafka's reverse hostname lookup when the hostname is not specified. The effective advertised controller listener now remove '0.0.0.0' hostname if the endpoint came from the listener configuration and not the advertised.listener configuration.
Reviewers: José Armando García Sancio <js...@apache.org>, Alyssa Huang <ah...@confluent.io>
---
core/src/main/scala/kafka/server/KafkaConfig.scala | 33 +++++++---
.../scala/unit/kafka/server/KafkaConfigTest.scala | 76 ++++++++++++++++++++++
2 files changed, 100 insertions(+), 9 deletions(-)
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 17b8cefb1ee..db5b84b58ae 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -519,29 +519,44 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
}
def effectiveAdvertisedControllerListeners: Seq[EndPoint] = {
- val controllerAdvertisedListeners = advertisedListeners.filter(l => controllerListenerNames.contains(l.listenerName.value()))
+ val advertisedListenersProp = getString(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG)
+ val controllerAdvertisedListeners = if (advertisedListenersProp != null) {
+ CoreUtils.listenerListToEndPoints(advertisedListenersProp, effectiveListenerSecurityProtocolMap, requireDistinctPorts=false)
+ .filter(l => controllerListenerNames.contains(l.listenerName.value()))
+ } else {
+ Seq.empty
+ }
val controllerListenersValue = controllerListeners
controllerListenerNames.flatMap { name =>
controllerAdvertisedListeners
.find(endpoint => endpoint.listenerName.equals(ListenerName.normalised(name)))
- .orElse(controllerListenersValue.find(endpoint => endpoint.listenerName.equals(ListenerName.normalised(name))))
+ .orElse(
+ // If users don't define advertised.listeners, the advertised controller listeners inherit from listeners configuration
+ // which match listener names in controller.listener.names.
+ // Removing "0.0.0.0" host to avoid validation errors. This is to be compatible with the old behavior before 3.9.
+ // The null or "" host does a reverse lookup in ListenerInfo#withWildcardHostnamesResolved.
+ controllerListenersValue
+ .find(endpoint => endpoint.listenerName.equals(ListenerName.normalised(name)))
+ .map(endpoint => if (endpoint.host == "0.0.0.0") {
+ new EndPoint(null, endpoint.port, endpoint.listenerName, endpoint.securityProtocol)
+ } else {
+ endpoint
+ })
+ )
}
}
def effectiveAdvertisedBrokerListeners: Seq[EndPoint] = {
- // Only expose broker listeners
- advertisedListeners.filterNot(l => controllerListenerNames.contains(l.listenerName.value()))
- }
-
- // Use advertised listeners if defined, fallback to listeners otherwise
- private def advertisedListeners: Seq[EndPoint] = {
+ // Use advertised listeners if defined, fallback to listeners otherwise
val advertisedListenersProp = getString(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG)
- if (advertisedListenersProp != null) {
+ val advertisedListeners = if (advertisedListenersProp != null) {
CoreUtils.listenerListToEndPoints(advertisedListenersProp, effectiveListenerSecurityProtocolMap, requireDistinctPorts=false)
} else {
listeners
}
+ // Only expose broker listeners
+ advertisedListeners.filterNot(l => controllerListenerNames.contains(l.listenerName.value()))
}
private def getInterBrokerListenerNameAndSecurityProtocol: (ListenerName, SecurityProtocol) = {
diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
index 5c95f409846..70111b5fde8 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -1230,6 +1230,82 @@ class KafkaConfigTest {
KafkaConfig.fromProps(props)
}
+ @Test
+ def testImplicitAllBindingListenersCanBeAdvertisedForBroker(): Unit = {
+ val props = new Properties()
+ props.setProperty(KRaftConfigs.PROCESS_ROLES_CONFIG, "broker")
+ val listeners = "PLAINTEXT://:9092"
+ props.setProperty(SocketServerConfigs.LISTENERS_CONFIG, listeners)
+ props.setProperty(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, listeners) // explicitly setting it in broker
+ props.setProperty(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "CONTROLLER")
+ props.setProperty(KRaftConfigs.NODE_ID_CONFIG, "2")
+ props.setProperty(QuorumConfig.QUORUM_VOTERS_CONFIG, "1@localhost:9093")
+
+ // Valid
+ KafkaConfig.fromProps(props)
+
+ // Also valid if we allow advertised listeners to derive from listeners
+ props.remove(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG)
+ KafkaConfig.fromProps(props)
+ }
+
+ @Test
+ def testExplicitAllBindingListenersCannotBeUsedForBroker(): Unit = {
+ val props = new Properties()
+ props.setProperty(KRaftConfigs.PROCESS_ROLES_CONFIG, "broker")
+ val listeners = "PLAINTEXT://0.0.0.0:9092"
+ props.setProperty(SocketServerConfigs.LISTENERS_CONFIG, listeners)
+ props.setProperty(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, listeners) // explicitly setting it in KRaft
+ props.setProperty(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "CONTROLLER")
+ props.setProperty(KRaftConfigs.NODE_ID_CONFIG, "2")
+ props.setProperty(QuorumConfig.QUORUM_VOTERS_CONFIG, "1@localhost:9093")
+
+ val expectedExceptionContainsText = "advertised.listeners cannot use the nonroutable meta-address 0.0.0.0. Use a routable IP address."
+ assertBadConfigContainingMessage(props, expectedExceptionContainsText)
+
+ // invalid if we allow advertised listeners to derive from listeners
+ props.remove(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG)
+ assertBadConfigContainingMessage(props, expectedExceptionContainsText)
+ }
+
+ @Test
+ def testImplicitAllBindingControllerListenersCanBeAdvertisedForKRaftController(): Unit = {
+ val props = new Properties()
+ props.setProperty(KRaftConfigs.PROCESS_ROLES_CONFIG, "controller")
+ val listeners = "CONTROLLER://:9093"
+ props.setProperty(SocketServerConfigs.LISTENERS_CONFIG, listeners)
+ props.setProperty(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, listeners) // explicitly setting it in KRaft
+ props.setProperty(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "CONTROLLER")
+ props.setProperty(KRaftConfigs.NODE_ID_CONFIG, "2")
+ props.setProperty(QuorumConfig.QUORUM_VOTERS_CONFIG, "2@localhost:9093")
+
+ // Valid
+ KafkaConfig.fromProps(props)
+
+ // Also valid if we allow advertised listeners to derive from listeners/controller.listener.names
+ props.remove(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG)
+ KafkaConfig.fromProps(props)
+ }
+
+ @Test
+ def testExplicitAllBindingControllerListenersCanBeAdvertisedForKRaftController(): Unit = {
+ val props = new Properties()
+ props.setProperty(KRaftConfigs.PROCESS_ROLES_CONFIG, "controller")
+ val listeners = "CONTROLLER://0.0.0.0:9093"
+ props.setProperty(SocketServerConfigs.LISTENERS_CONFIG, listeners)
+ props.setProperty(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, listeners) // explicitly setting it in KRaft
+ props.setProperty(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "CONTROLLER")
+ props.setProperty(KRaftConfigs.NODE_ID_CONFIG, "2")
+ props.setProperty(QuorumConfig.QUORUM_VOTERS_CONFIG, "2@localhost:9093")
+
+ val expectedExceptionContainsText = "advertised.listeners cannot use the nonroutable meta-address 0.0.0.0. Use a routable IP address."
+ assertBadConfigContainingMessage(props, expectedExceptionContainsText)
+
+ // Valid if we allow advertised listeners to derive from listeners/controller.listener.names
+ props.remove(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG)
+ KafkaConfig.fromProps(props)
+ }
+
@Test
def testControllerListenersCanBeAdvertisedForKRaftCombined(): Unit = {
val props = new Properties()