You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "TaiJuWu (via GitHub)" <gi...@apache.org> on 2025/09/22 00:49:10 UTC
[PR] MINOR: migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig [kafka]
TaiJuWu opened a new pull request, #20567:
URL: https://github.com/apache/kafka/pull/20567
Migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20567:
URL: https://github.com/apache/kafka/pull/20567#discussion_r2704388628
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -162,6 +169,82 @@ public static Map<String, String> getMap(String propName, String propValue) {
}
}
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap) {
+ return listenerListToEndPoints(listeners, securityProtocolMap, true);
+ }
+
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap, boolean requireDistinctPorts) {
+ try {
+ List<Endpoint> endPoints = SocketServerConfigs.listenerListToEndPoints(listeners, securityProtocolMap);
+ validate(endPoints, listeners, requireDistinctPorts);
+ return endPoints;
+ } catch (Exception e) {
+ throw new IllegalArgumentException(String.format("Error creating broker listeners from '%s': %s", listeners, e.getMessage()), e);
+ }
+ }
+
+ private static void validate(List<Endpoint> endPoints, List<String> listeners, boolean requireDistinctPorts) {
+ long distinctListenerNames = endPoints.stream().map(Endpoint::listener).distinct().count();
+ if (distinctListenerNames != endPoints.size()) {
+ throw new IllegalArgumentException("Each listener must have a different name, listeners: " + listeners);
+ }
+
+ if (!requireDistinctPorts) return;
+
+ endPoints.stream()
+ .filter(ep -> ep.port() != 0) // filter port 0 for unit tests
+ .collect(Collectors.groupingBy(Endpoint::port))
+ .entrySet().stream()
+ .filter(entry -> entry.getValue().size() > 1)
+ .forEach(entry -> {
+ // Iterate through every grouping of duplicates by port to see if they are valid
+ int port = entry.getKey();
+ List<Endpoint> eps = entry.getValue();
+ // Exception case, let's allow duplicate ports if one host is on IPv4 and the other one is on IPv6
+ Map<Boolean, List<Endpoint>> partitionedByValidIp = eps.stream()
+ .collect(Collectors.partitioningBy(ep -> ep.host() != null && INET_ADDRESS_VALIDATOR.isValid(ep.host())));
+
+ List<Endpoint> duplicatesWithIpHosts = partitionedByValidIp.get(true);
+ List<Endpoint> duplicatesWithoutIpHosts = partitionedByValidIp.get(false);
+
+ checkDuplicateListenerPorts(duplicatesWithoutIpHosts, listeners);
+
+ if (duplicatesWithIpHosts.isEmpty()) {
+ // No-op
+ } else if (duplicatesWithIpHosts.size() == 2) {
+ String errorMessage = "If you have two listeners on the same port then one needs to be IPv4 and the other IPv6, listeners: " + listeners + ", port: " + port;
+ Endpoint ep1 = duplicatesWithIpHosts.get(0);
+ Endpoint ep2 = duplicatesWithIpHosts.get(1);
+ if (!validateOneIsIpv4AndOtherIpv6(ep1.host(), ep2.host())) {
+ throw new IllegalArgumentException(errorMessage);
+ }
+
+ // If we reach this point it means that even though duplicatesWithIpHosts in isolation can be valid, if
+ // there happens to be ANOTHER listener on this port without an IP host (such as a null host) then its
+ // not valid.
+ if (!duplicatesWithoutIpHosts.isEmpty()) {
+ throw new IllegalArgumentException(errorMessage);
+ }
+ } else {
+ // Having more than 2 duplicate endpoints doesn't make sense since we only have 2 IP stacks (one is IPv4
+ // and the other is IPv6)
+ throw new IllegalArgumentException("Each listener must have a different port unless exactly one listener has an IPv4 address and the other IPv6 address, listeners: " + listeners + ", port: " + port);
+ }
+ });
+ }
+
+ private static boolean validateOneIsIpv4AndOtherIpv6(String first, String second) {
+ return (INET_ADDRESS_VALIDATOR.isValidInet4Address(first) && INET_ADDRESS_VALIDATOR.isValidInet6Address(second)) ||
+ (INET_ADDRESS_VALIDATOR.isValidInet6Address(first) && INET_ADDRESS_VALIDATOR.isValidInet4Address(second));
+ }
+
+ private static void checkDuplicateListenerPorts(List<Endpoint> endpoints, List<String> listeners) {
+ Set<String> distinctHosts = endpoints.stream().map(ep -> ep.host() == null ? "" : ep.host()).collect(Collectors.toSet());
Review Comment:
This change should have caused some test to fail, but it didn't. This highlights that our test coverage needs to be enhanced. I have opened KAFKA-20080 to address this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #20567:
URL: https://github.com/apache/kafka/pull/20567#discussion_r2392661028
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -161,6 +169,76 @@ public static Map<String, String> getMap(String propName, String propValue) {
}
}
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap) {
+ return listenerListToEndPoints(listeners, securityProtocolMap, true);
+ }
+
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap, boolean requireDistinctPorts) {
+ List<Endpoint> endPoints;
+ try {
+ endPoints = SocketServerConfigs.listenerListToEndPoints(listeners, securityProtocolMap);
+ } catch (Exception e) {
+ throw new IllegalArgumentException(String.format("Error creating broker listeners from '%s': %s", listeners, e.getMessage()), e);
+ }
+ validate(endPoints, listeners, requireDistinctPorts);
+ return endPoints;
+ }
+
+ private static void validate(List<Endpoint> endPoints, List<String> listeners, boolean requireDistinctPorts) {
+ long distinctListenerNames = endPoints.stream().map(Endpoint::listener).distinct().count();
+ if (distinctListenerNames != endPoints.size()) {
+ throw new IllegalArgumentException("Each listener must have a different name, listeners: " + listeners);
+ }
+
+ if (!requireDistinctPorts) return;
+
+ Map<Integer, List<Endpoint>> duplicatePorts = endPoints.stream()
+ .filter(ep -> ep.port() != 0) // filter port 0 for unit tests
+ .collect(Collectors.groupingBy(Endpoint::port))
+ .entrySet().stream()
+ .filter(entry -> entry.getValue().size() > 1)
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
Review Comment:
fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Move listener validation logic from KafkaConfig to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #20567:
URL: https://github.com/apache/kafka/pull/20567
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Move listener validation logic from KafkaConfig to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20567:
URL: https://github.com/apache/kafka/pull/20567#discussion_r2742642083
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -163,6 +170,82 @@ public static Map<String, String> getMap(String propName, String propValue) {
}
}
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap) {
+ return listenerListToEndPoints(listeners, securityProtocolMap, true);
+ }
+
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap, boolean requireDistinctPorts) {
+ try {
+ List<Endpoint> endPoints = SocketServerConfigs.listenerListToEndPoints(listeners, securityProtocolMap);
+ validate(endPoints, listeners, requireDistinctPorts);
+ return endPoints;
+ } catch (Exception e) {
+ throw new IllegalArgumentException(String.format("Error creating broker listeners from '%s': %s", listeners, e.getMessage()), e);
+ }
+ }
+
+ private static void validate(List<Endpoint> endPoints, List<String> listeners, boolean requireDistinctPorts) {
+ long distinctListenerNames = endPoints.stream().map(Endpoint::listener).distinct().count();
+ if (distinctListenerNames != endPoints.size()) {
+ throw new IllegalArgumentException("Each listener must have a different name, listeners: " + listeners);
+ }
+
+ if (!requireDistinctPorts) return;
+
+ endPoints.stream()
+ .filter(ep -> ep.port() != 0) // filter port 0 for unit tests
+ .collect(Collectors.groupingBy(Endpoint::port))
+ .entrySet().stream()
+ .filter(entry -> entry.getValue().size() > 1)
+ .forEach(entry -> {
+ // Iterate through every grouping of duplicates by port to see if they are valid
+ int port = entry.getKey();
+ List<Endpoint> eps = entry.getValue();
+ // Exception case, let's allow duplicate ports if one host is on IPv4 and the other one is on IPv6
+ Map<Boolean, List<Endpoint>> partitionedByValidIp = eps.stream()
+ .collect(Collectors.partitioningBy(ep -> ep.host() != null && INET_ADDRESS_VALIDATOR.isValid(ep.host())));
+
+ List<Endpoint> duplicatesWithIpHosts = partitionedByValidIp.get(true);
+ List<Endpoint> duplicatesWithoutIpHosts = partitionedByValidIp.get(false);
+
+ checkDuplicateListenerPorts(duplicatesWithoutIpHosts, listeners);
+
+ if (duplicatesWithIpHosts.isEmpty()) {
Review Comment:
the empty body does bother me. Would you mind trying following style?
```java
if (duplicatesWithIpHosts.isEmpty()) return;
if (duplicatesWithIpHosts.size() == 2) {
String errorMessage = "If you have two listeners on the same port then one needs to be IPv4 and the other IPv6, listeners: " + listeners + ", port: " + port;
Endpoint ep1 = duplicatesWithIpHosts.get(0);
Endpoint ep2 = duplicatesWithIpHosts.get(1);
if (!validateOneIsIpv4AndOtherIpv6(ep1.host(), ep2.host())) {
throw new IllegalArgumentException(errorMessage);
}
// If we reach this point it means that even though duplicatesWithIpHosts in isolation can be valid, if
// there happens to be ANOTHER listener on this port without an IP host (such as a null host) then its
// not valid.
if (!duplicatesWithoutIpHosts.isEmpty()) {
throw new IllegalArgumentException(errorMessage);
}
return;
}
// Having more than 2 duplicate endpoints doesn't make sense since we only have 2 IP stacks (one is IPv4
// and the other is IPv6)
throw new IllegalArgumentException("Each listener must have a different port unless exactly one listener has an IPv4 address and the other IPv6 address, listeners: " + listeners + ", port: " + port);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #20567:
URL: https://github.com/apache/kafka/pull/20567#discussion_r2378895835
##########
core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala:
##########
@@ -623,9 +623,9 @@ class KafkaConfigTest {
// configuration with no listeners
val conf = KafkaConfig.fromProps(props)
- assertEquals(listenerListToEndPoints(util.List.of("PLAINTEXT://:9092")), conf.listeners)
+ assertEquals(listenerListToEndPoints(util.List.of("PLAINTEXT://:9092")).asScala, conf.listeners)
assertNull(conf.listeners.find(_.securityProtocol == SecurityProtocol.PLAINTEXT).get.host)
- assertEquals(conf.effectiveAdvertisedBrokerListeners, listenerListToEndPoints(util.List.of("PLAINTEXT://:9092")))
+ assertEquals(conf.effectiveAdvertisedBrokerListeners, listenerListToEndPoints(util.List.of("PLAINTEXT://:9092")).asScala)
Review Comment:
It should be in reverse order.
```suggestion
assertEquals(listenerListToEndPoints(util.List.of("PLAINTEXT://:9092")).asScala, conf.effectiveAdvertisedBrokerListeners))
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #20567:
URL: https://github.com/apache/kafka/pull/20567#issuecomment-3749543254
@TaiJuWu please fix the conflicts :_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu closed pull request #20567: MINOR: migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig
URL: https://github.com/apache/kafka/pull/20567
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20567:
URL: https://github.com/apache/kafka/pull/20567#discussion_r2377888370
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -161,6 +169,82 @@ public static Map<String, String> getMap(String propName, String propValue) {
}
}
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap) {
+ return listenerListToEndPoints(listeners, securityProtocolMap, true);
+ }
+
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap, boolean requireDistinctPorts) {
+ List<Endpoint> endPoints;
+ try {
+ endPoints = SocketServerConfigs.listenerListToEndPoints(listeners, securityProtocolMap);
+ } catch (Exception e) {
+ throw new IllegalArgumentException(String.format("Error creating broker listeners from '%s': %s", listeners, e.getMessage()), e);
+ }
+ validate(endPoints, listeners, requireDistinctPorts);
+ return endPoints;
+ }
+
+ private static void validate(List<Endpoint> endPoints, List<String> listeners, boolean requireDistinctPorts) {
+ long distinctListenerNames = endPoints.stream().map(Endpoint::listener).distinct().count();
+ if (distinctListenerNames != endPoints.size()) {
+ throw new IllegalArgumentException("Each listener must have a different name, listeners: " + listeners);
+ }
+
+ if (!requireDistinctPorts) return;
+
+ Map<Integer, List<Endpoint>> duplicatePorts = endPoints.stream()
+ .filter(ep -> ep.port() != 0) // filter port 0 for unit tests
+ .collect(Collectors.groupingBy(Endpoint::port))
+ .entrySet().stream()
+ .filter(entry -> entry.getValue().size() > 1)
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+ duplicatePorts.forEach((port, eps) -> {
+ Map<Boolean, List<Endpoint>> partitionedByValidIp = eps.stream()
+ .collect(Collectors.partitioningBy(ep -> ep.host() != null && INET_ADDRESS_VALIDATOR.isValid(ep.host())));
+
+ List<Endpoint> duplicatesWithIpHosts = partitionedByValidIp.get(true);
+ List<Endpoint> duplicatesWithoutIpHosts = partitionedByValidIp.get(false);
+
+ if (requireDistinctPorts) {
Review Comment:
`requireDistinctPorts` is always `false`, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #20567:
URL: https://github.com/apache/kafka/pull/20567#discussion_r2380719626
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -161,6 +169,82 @@ public static Map<String, String> getMap(String propName, String propValue) {
}
}
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap) {
+ return listenerListToEndPoints(listeners, securityProtocolMap, true);
+ }
+
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap, boolean requireDistinctPorts) {
+ List<Endpoint> endPoints;
+ try {
+ endPoints = SocketServerConfigs.listenerListToEndPoints(listeners, securityProtocolMap);
+ } catch (Exception e) {
+ throw new IllegalArgumentException(String.format("Error creating broker listeners from '%s': %s", listeners, e.getMessage()), e);
+ }
+ validate(endPoints, listeners, requireDistinctPorts);
+ return endPoints;
+ }
+
+ private static void validate(List<Endpoint> endPoints, List<String> listeners, boolean requireDistinctPorts) {
+ long distinctListenerNames = endPoints.stream().map(Endpoint::listener).distinct().count();
+ if (distinctListenerNames != endPoints.size()) {
+ throw new IllegalArgumentException("Each listener must have a different name, listeners: " + listeners);
+ }
+
+ if (!requireDistinctPorts) return;
+
+ Map<Integer, List<Endpoint>> duplicatePorts = endPoints.stream()
+ .filter(ep -> ep.port() != 0) // filter port 0 for unit tests
+ .collect(Collectors.groupingBy(Endpoint::port))
+ .entrySet().stream()
+ .filter(entry -> entry.getValue().size() > 1)
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+ duplicatePorts.forEach((port, eps) -> {
+ Map<Boolean, List<Endpoint>> partitionedByValidIp = eps.stream()
+ .collect(Collectors.partitioningBy(ep -> ep.host() != null && INET_ADDRESS_VALIDATOR.isValid(ep.host())));
+
+ List<Endpoint> duplicatesWithIpHosts = partitionedByValidIp.get(true);
+ List<Endpoint> duplicatesWithoutIpHosts = partitionedByValidIp.get(false);
+
+ if (requireDistinctPorts) {
Review Comment:
fixed, thanks.
##########
core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala:
##########
@@ -623,9 +623,9 @@ class KafkaConfigTest {
// configuration with no listeners
val conf = KafkaConfig.fromProps(props)
- assertEquals(listenerListToEndPoints(util.List.of("PLAINTEXT://:9092")), conf.listeners)
+ assertEquals(listenerListToEndPoints(util.List.of("PLAINTEXT://:9092")).asScala, conf.listeners)
assertNull(conf.listeners.find(_.securityProtocol == SecurityProtocol.PLAINTEXT).get.host)
- assertEquals(conf.effectiveAdvertisedBrokerListeners, listenerListToEndPoints(util.List.of("PLAINTEXT://:9092")))
+ assertEquals(conf.effectiveAdvertisedBrokerListeners, listenerListToEndPoints(util.List.of("PLAINTEXT://:9092")).asScala)
Review Comment:
fixed, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #20567:
URL: https://github.com/apache/kafka/pull/20567#issuecomment-3476709030
@TaiJuWu could you please fix the conflicts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20567:
URL: https://github.com/apache/kafka/pull/20567#discussion_r2426790292
##########
core/src/main/scala/kafka/utils/CoreUtils.scala:
##########
@@ -121,76 +111,4 @@ object CoreUtils {
def inWriteLock[T](lock: ReadWriteLock)(fun: => T): T = inLock[T](lock.writeLock)(fun)
- def listenerListToEndPoints(listeners: java.util.List[String], securityProtocolMap: java.util.Map[ListenerName, SecurityProtocol]): Seq[Endpoint] = {
- listenerListToEndPoints(listeners, securityProtocolMap, requireDistinctPorts = true)
- }
-
- private def checkDuplicateListenerPorts(endpoints: Seq[Endpoint], listeners: java.util.List[String]): Unit = {
- val distinctPorts = endpoints.map(_.port).distinct
- require(distinctPorts.size == endpoints.map(_.port).size, s"Each listener must have a different port, listeners: $listeners")
- }
-
- def listenerListToEndPoints(listeners: java.util.List[String], securityProtocolMap: java.util.Map[ListenerName, SecurityProtocol], requireDistinctPorts: Boolean): Seq[Endpoint] = {
- def validateOneIsIpv4AndOtherIpv6(first: String, second: String): Boolean =
- (inetAddressValidator.isValidInet4Address(first) && inetAddressValidator.isValidInet6Address(second)) ||
- (inetAddressValidator.isValidInet6Address(first) && inetAddressValidator.isValidInet4Address(second))
-
- def validate(endPoints: Seq[Endpoint]): Unit = {
- val distinctListenerNames = endPoints.map(_.listener).distinct
- require(distinctListenerNames.size == endPoints.size, s"Each listener must have a different name, listeners: $listeners")
-
- val (duplicatePorts, _) = endPoints.filter {
- // filter port 0 for unit tests
- ep => ep.port != 0
- }.groupBy(_.port).partition {
- case (_, endpoints) => endpoints.size > 1
- }
-
- // Exception case, let's allow duplicate ports if one host is on IPv4 and the other one is on IPv6
- val duplicatePortsPartitionedByValidIps = duplicatePorts.map {
- case (port, eps) =>
- (port, eps.partition(ep =>
- ep.host != null && inetAddressValidator.isValid(ep.host)
- ))
- }
-
- // Iterate through every grouping of duplicates by port to see if they are valid
- duplicatePortsPartitionedByValidIps.foreach {
- case (port, (duplicatesWithIpHosts, duplicatesWithoutIpHosts)) =>
- if (requireDistinctPorts)
- checkDuplicateListenerPorts(duplicatesWithoutIpHosts, listeners)
-
- duplicatesWithIpHosts match {
- case eps if eps.isEmpty =>
- case Seq(ep1, ep2) =>
- if (requireDistinctPorts) {
- val errorMessage = "If you have two listeners on " +
- s"the same port then one needs to be IPv4 and the other IPv6, listeners: $listeners, port: $port"
- require(validateOneIsIpv4AndOtherIpv6(ep1.host, ep2.host), errorMessage)
-
- // If we reach this point it means that even though duplicatesWithIpHosts in isolation can be valid, if
Review Comment:
Could you please keep this comment
##########
core/src/main/scala/kafka/utils/CoreUtils.scala:
##########
@@ -121,76 +111,4 @@ object CoreUtils {
def inWriteLock[T](lock: ReadWriteLock)(fun: => T): T = inLock[T](lock.writeLock)(fun)
- def listenerListToEndPoints(listeners: java.util.List[String], securityProtocolMap: java.util.Map[ListenerName, SecurityProtocol]): Seq[Endpoint] = {
- listenerListToEndPoints(listeners, securityProtocolMap, requireDistinctPorts = true)
- }
-
- private def checkDuplicateListenerPorts(endpoints: Seq[Endpoint], listeners: java.util.List[String]): Unit = {
- val distinctPorts = endpoints.map(_.port).distinct
- require(distinctPorts.size == endpoints.map(_.port).size, s"Each listener must have a different port, listeners: $listeners")
- }
-
- def listenerListToEndPoints(listeners: java.util.List[String], securityProtocolMap: java.util.Map[ListenerName, SecurityProtocol], requireDistinctPorts: Boolean): Seq[Endpoint] = {
- def validateOneIsIpv4AndOtherIpv6(first: String, second: String): Boolean =
- (inetAddressValidator.isValidInet4Address(first) && inetAddressValidator.isValidInet6Address(second)) ||
- (inetAddressValidator.isValidInet6Address(first) && inetAddressValidator.isValidInet4Address(second))
-
- def validate(endPoints: Seq[Endpoint]): Unit = {
- val distinctListenerNames = endPoints.map(_.listener).distinct
- require(distinctListenerNames.size == endPoints.size, s"Each listener must have a different name, listeners: $listeners")
-
- val (duplicatePorts, _) = endPoints.filter {
- // filter port 0 for unit tests
- ep => ep.port != 0
- }.groupBy(_.port).partition {
- case (_, endpoints) => endpoints.size > 1
- }
-
- // Exception case, let's allow duplicate ports if one host is on IPv4 and the other one is on IPv6
- val duplicatePortsPartitionedByValidIps = duplicatePorts.map {
- case (port, eps) =>
- (port, eps.partition(ep =>
- ep.host != null && inetAddressValidator.isValid(ep.host)
- ))
- }
-
- // Iterate through every grouping of duplicates by port to see if they are valid
- duplicatePortsPartitionedByValidIps.foreach {
- case (port, (duplicatesWithIpHosts, duplicatesWithoutIpHosts)) =>
- if (requireDistinctPorts)
- checkDuplicateListenerPorts(duplicatesWithoutIpHosts, listeners)
-
- duplicatesWithIpHosts match {
- case eps if eps.isEmpty =>
- case Seq(ep1, ep2) =>
- if (requireDistinctPorts) {
- val errorMessage = "If you have two listeners on " +
- s"the same port then one needs to be IPv4 and the other IPv6, listeners: $listeners, port: $port"
- require(validateOneIsIpv4AndOtherIpv6(ep1.host, ep2.host), errorMessage)
-
- // If we reach this point it means that even though duplicatesWithIpHosts in isolation can be valid, if
- // there happens to be ANOTHER listener on this port without an IP host (such as a null host) then its
- // not valid.
- if (duplicatesWithoutIpHosts.nonEmpty)
- throw new IllegalArgumentException(errorMessage)
- }
- case _ =>
- // Having more than 2 duplicate endpoints doesn't make sense since we only have 2 IP stacks (one is IPv4
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20567:
URL: https://github.com/apache/kafka/pull/20567#discussion_r2386192751
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -161,6 +169,76 @@ public static Map<String, String> getMap(String propName, String propValue) {
}
}
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap) {
+ return listenerListToEndPoints(listeners, securityProtocolMap, true);
+ }
+
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap, boolean requireDistinctPorts) {
+ List<Endpoint> endPoints;
+ try {
+ endPoints = SocketServerConfigs.listenerListToEndPoints(listeners, securityProtocolMap);
+ } catch (Exception e) {
+ throw new IllegalArgumentException(String.format("Error creating broker listeners from '%s': %s", listeners, e.getMessage()), e);
+ }
+ validate(endPoints, listeners, requireDistinctPorts);
+ return endPoints;
+ }
+
+ private static void validate(List<Endpoint> endPoints, List<String> listeners, boolean requireDistinctPorts) {
+ long distinctListenerNames = endPoints.stream().map(Endpoint::listener).distinct().count();
+ if (distinctListenerNames != endPoints.size()) {
+ throw new IllegalArgumentException("Each listener must have a different name, listeners: " + listeners);
+ }
+
+ if (!requireDistinctPorts) return;
+
+ Map<Integer, List<Endpoint>> duplicatePorts = endPoints.stream()
+ .filter(ep -> ep.port() != 0) // filter port 0 for unit tests
+ .collect(Collectors.groupingBy(Endpoint::port))
+ .entrySet().stream()
+ .filter(entry -> entry.getValue().size() > 1)
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
Review Comment:
It seems we can call `forEach` in this phase, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #20567:
URL: https://github.com/apache/kafka/pull/20567#discussion_r2386513598
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -161,6 +169,76 @@ public static Map<String, String> getMap(String propName, String propValue) {
}
}
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap) {
+ return listenerListToEndPoints(listeners, securityProtocolMap, true);
+ }
+
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap, boolean requireDistinctPorts) {
+ List<Endpoint> endPoints;
+ try {
+ endPoints = SocketServerConfigs.listenerListToEndPoints(listeners, securityProtocolMap);
+ } catch (Exception e) {
+ throw new IllegalArgumentException(String.format("Error creating broker listeners from '%s': %s", listeners, e.getMessage()), e);
+ }
+ validate(endPoints, listeners, requireDistinctPorts);
+ return endPoints;
+ }
Review Comment:
Could we do the following?
```suggestion
public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap, boolean requireDistinctPorts) {
try {
List<Endpoint> endPoints = SocketServerConfigs.listenerListToEndPoints(listeners, securityProtocolMap);
validate(endPoints, listeners, requireDistinctPorts);
return endPoints;
} catch (Exception e) {
throw new IllegalArgumentException(String.format("Error creating broker listeners from '%s': %s", listeners, e.getMessage()), e);
}
}
```
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -52,6 +56,10 @@
* For more details check KAFKA-15853
*/
public abstract class AbstractKafkaConfig extends AbstractConfig {
+
+ private static final InetAddressValidator INET_ADDRESS_VALIDATOR = InetAddressValidator.getInstance();
+
+
Review Comment:
Redundant empty line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20567:
URL: https://github.com/apache/kafka/pull/20567#discussion_r2375305246
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -161,6 +169,80 @@ public static Map<String, String> getMap(String propName, String propValue) {
}
}
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap) {
+ return listenerListToEndPoints(listeners, securityProtocolMap, true);
+ }
+
+ public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap, boolean requireDistinctPorts) {
+ List<Endpoint> endPoints;
+ try {
+ endPoints = SocketServerConfigs.listenerListToEndPoints(listeners, securityProtocolMap);
+ } catch (Exception e) {
+ throw new IllegalArgumentException(String.format("Error creating broker listeners from '%s': %s", listeners, e.getMessage()), e);
+ }
+ validate(endPoints, listeners, requireDistinctPorts);
+ return endPoints;
+ }
+
+ private static void validate(List<Endpoint> endPoints, List<String> listeners, boolean requireDistinctPorts) {
+ long distinctListenerNames = endPoints.stream().map(Endpoint::listener).distinct().count();
+ if (distinctListenerNames != endPoints.size()) {
+ throw new IllegalArgumentException("Each listener must have a different name, listeners: " + listeners);
+ }
+
Review Comment:
It can return early if `requireDistinctPorts` is `false`, right? for example:
```java
private static void validate(List<Endpoint> endPoints, List<String> listeners, boolean requireDistinctPorts) {
long distinct = endPoints.stream().map(Endpoint::listener).distinct().count();
if (distinct != endPoints.size()) {
throw new IllegalArgumentException("Each listener must have a different name, listeners: " + listeners);
}
if (!requireDistinctPorts) return;
...
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: migrate listenerListToEndPoints from CoreUtils to AbstractKafkaConfig [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on PR #20567:
URL: https://github.com/apache/kafka/pull/20567#issuecomment-3399936392
@chia7712 Thanks for review, all comments are addressed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org