Repository: apache/kafka
File: server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java (Java)

Code before:
    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);

            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 (!duplicatesWithoutIpHosts.isEmpty()) {
                    throw new IllegalArgumentException(errorMessage);
                }
            } else {
                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);
            }

Review comment (security):
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);
        }
    }
```

Code after:
            return mapValue;
        }
    }

    public static Map<String, String> getMap(String propName, String propValue) {
        try {
            return Csv.parseCsvMap(propValue);
        } catch (Exception e) {
            throw new IllegalArgumentException(
                    String.format("Error parsing configuration property '%s': %s", propName, e.getMessage()));
        }
    }

    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);