diff --git a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java
index 5c345a35ed146..11d3e84687d69 100644
--- a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java
+++ b/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.
      */
diff --git a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java
index 45b7cada93665..4bb15d8a6897a 100644
--- a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java
+++ b/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.
diff --git a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java
index 9e46706537b3b..ee5c19c15e9c3 100644
--- a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java
+++ b/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();