public ListenerInfo withEphemeralPortsCorrected(Function<String, Integer> getBoundPortCallback) {
        LinkedHashMap<String, Endpoint> newListeners = new LinkedHashMap<>();
        for (Map.Entry<String, Endpoint> entry : listeners.entrySet()) {
            if (entry.getValue().port() == 0) {
                Endpoint prevEndpoint = entry.getValue();
                int newPort = getBoundPortCallback.apply(entry.getKey());
                checkPortIsSerializable(newPort);
                log.info("{}: resolved ephemeral port to {}", entry.getValue().listenerName().get(),
                        newPort);
                newListeners.put(entry.getKey(), new Endpoint(prevEndpoint.listenerName().get(),
                        prevEndpoint.securityProtocol(),
                        prevEndpoint.host(),
                        newPort));
            } else {
                newListeners.put(entry.getKey(), entry.getValue());
            }
        }
        return new ListenerInfo(newListeners);
    }
Create a new ListenerInfo object where ephemeral ports are populated with their true runtime
values.

In other words, if a port was set to 0, indicating that a random port should be assigned by the
operating system, this function will replace it with the value the operating system actually
chose.

@param getBoundPortCallback  The callback used to correct ephemeral endpoints.

@return A new ListenerInfo object.
metadata/src/main/java/org/apache/kafka/metadata/ListenerInfo.java