KAFKA-19443: test-common ClusterInstance should make it easy to have more listeners

test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java

@@ -121,9 +121,17 @@ default Set<Integer> brokerIds() {
 
     /**
      * The broker connect string which can be used by clients for bootstrapping
+     * using the broker listenerName
      */
     String bootstrapServers();
 
+    /**
+     * The broker connect string which can be used by clients for bootstrapping
+     * using an alternate listener
+     * @throws org.apache.kafka.common.KafkaException if listenerName is not defined
+     */
+    String bootstrapServers(ListenerName listenerName);
+
     /**
      * The broker connect string which can be used by clients for bootstrapping to the controller quorum.
      */

test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java

@@ -550,12 +550,15 @@ public Properties clientProperties() {
     }
 
     public String bootstrapServers() {
+        return bootstrapServers(nodes.brokerListenerName());
+    }
+
+    public String bootstrapServers(ListenerName listenerName) {
         StringBuilder bld = new StringBuilder();
         String prefix = "";
         for (Entry<Integer, BrokerServer> entry : brokers.entrySet()) {
             int brokerId = entry.getKey();
             BrokerServer broker = entry.getValue();
-            ListenerName listenerName = nodes.brokerListenerName();
             // The KafkaConfig#listeners method normalizes the listener name.
             // The result from TestKitNodes#brokerListenerName method should be normalized as well,
             // so that it matches the listener name in the KafkaConfig.

test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java

@@ -145,6 +145,11 @@ public String bootstrapServers() {
             return clusterTestKit.bootstrapServers();
         }
 
+        @Override
+        public String bootstrapServers(ListenerName listenerName) {
+            return clusterTestKit.bootstrapServers(listenerName);
+        }
+
         @Override
         public String bootstrapControllers() {
             return clusterTestKit.bootstrapControllers();