public String bootstrapServers() {
        StringBuilder bld = new StringBuilder();
        String prefix = "";
        for (Entry<Integer, BrokerServer> entry : brokers.entrySet()) {
            int brokerId = entry.getKey();
            BrokerServer broker = entry.getValue();
            ListenerName listenerName = nodes.externalListenerName();
            int port = broker.boundPort(listenerName);
            if (port <= 0) {
                throw new RuntimeException("Broker " + brokerId + " does not yet " +
                    "have a bound port for " + listenerName + ".  Did you start " +
                    "the cluster yet?");
            }
            bld.append(prefix).append("localhost:").append(port);
            prefix = ",";
        }
        return bld.toString();
    }
Wait for a controller to mark all the brokers as ready (registered and unfenced).
And also wait for the metadata cache up-to-date in each broker server.
core/src/test/java/kafka/testkit/KafkaClusterTestKit.java