KAFKA-21074 follow-up: reject a multi-broker bootstrap in KafkaProtocolFaultProxy (#23515)

`KafkaProtocolFaultProx` supports a single broker.

Follow-up for this comment
#23438 (comment)

Author: bbejeck

clients/src/test/java/org/apache/kafka/test/faultproxy/KafkaProtocolFaultProxyTest.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.test.faultproxy;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class KafkaProtocolFaultProxyTest {
+
+    @Test
+    public void shouldRejectMultiBrokerBootstrap() {
+        // The proxy rewrites all routing to itself and forwards to a single upstream broker, so a
+        // multi-broker bootstrap must fail fast rather than silently proxy only the first broker.
+        final IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
+                () -> KafkaProtocolFaultProxy.inFrontOf("localhost:9092,localhost:9093,localhost:9094"));
+        assertTrue(e.getMessage().contains("single broker"), e.getMessage());
+        assertTrue(e.getMessage().contains("3 servers"), e.getMessage());
+    }
+
+    @Test
+    public void shouldAcceptSingleBrokerBootstrap() throws Exception {
+        // A single host:port is accepted; bootstrapServers() exposes the proxy's own listening address.
+        try (KafkaProtocolFaultProxy proxy = KafkaProtocolFaultProxy.inFrontOf("localhost:9092")) {
+            final String bootstrap = proxy.bootstrapServers();
+            assertTrue(bootstrap.startsWith("localhost:"), bootstrap);
+            assertEquals(1, bootstrap.split(",").length, bootstrap);
+        }
+    }
+}

clients/src/testFixtures/java/org/apache/kafka/test/faultproxy/KafkaProtocolFaultProxy.java

@@ -82,6 +82,13 @@
  * advertised address points back at itself, so a single-broker embedded cluster needs no special config
  * (its own ephemeral port is discovered from {@code bootstrapServers()}).
  *
+ * <p><b>Single broker only.</b> Because every advertised broker and coordinator address is rewritten to this
+ * one proxy — which forwards to a single upstream broker — the proxy supports exactly one broker. A
+ * multi-broker bootstrap would silently funnel every partition and coordinator to the first broker, producing
+ * {@code NOT_LEADER_OR_FOLLOWER}/{@code NOT_COORDINATOR} retry loops, so {@link #inFrontOf(String)} rejects a
+ * bootstrap that lists more than one server. Front a single-broker {@code EmbeddedKafkaCluster}, or pass one
+ * broker's address (e.g. {@code cluster.bootstrapServers().split(",")[0]}).
+ *
  * <p>Determinism: {@code once()}/{@code onCall(n)}/{@code times(n)} are deterministic and safe for
  * assertions; {@code withProbability(p)} is chaos-mode only. The proxy never closes sockets unless a
  * {@code disconnectOn(...)} rule fires, so it is not itself a source of flakiness.
@@ -140,13 +147,27 @@ public final class KafkaProtocolFaultProxy implements AutoCloseable {
     private volatile int proxyPort;
 
     private KafkaProtocolFaultProxy(final String targetBootstrap) {
-        final String hostPort = targetBootstrap.split(",")[0].trim();
+        final String[] servers = targetBootstrap.split(",");
+        // Routing rewrites all addresses to this proxy (one upstream broker), so a multi-broker bootstrap
+        // would funnel every partition and coordinator to the first broker
+        // causing NOT_LEADER / NOT_COORDINATOR loops.
+        if (servers.length > 1) {
+            throw new IllegalArgumentException(
+                    "KafkaProtocolFaultProxy proxies a single broker, but the bootstrap '" + targetBootstrap
+                    + "' lists " + servers.length + " servers");
+        }
+        final String hostPort = servers[0].trim();
         final int idx = hostPort.lastIndexOf(':');
         this.targetHost = hostPort.substring(0, idx);
         this.targetPort = Integer.parseInt(hostPort.substring(idx + 1));
     }
 
-    /** Create and start a proxy in front of the given broker bootstrap address. */
+    /**
+     * Create and start a proxy in front of the given broker bootstrap address.
+     *
+     * @param targetBootstrap a single broker's {@code host:port} — the proxy supports one broker only
+     * @throws IllegalArgumentException if {@code targetBootstrap} lists more than one server
+     */
     public static KafkaProtocolFaultProxy inFrontOf(final String targetBootstrap) {
         final KafkaProtocolFaultProxy proxy = new KafkaProtocolFaultProxy(targetBootstrap);
         try {