KAFKA-15104: Fix flaky tests in MetadataQuorumCommandTest

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

@@ -588,23 +588,28 @@ private void formatNode(
     }
 
     public void startup() throws ExecutionException, InterruptedException {
-        List<Future<?>> futures = new ArrayList<>();
+        List<Future<?>> controllerFutures = new ArrayList<>();
+        List<Future<?>> brokerFutures = new ArrayList<>();
         try {
-            // Note the startup order here is chosen to be consistent with
-            // `KafkaRaftServer`. See comments in that class for an explanation.
+            // Startup order matches `KafkaRaftServer`: controllers before brokers.
+            // Submit all controllers concurrently (avoids deadlocking a multi-controller
+            // quorum), then await them before starting brokers so a broker can't fail
+            // catching up to a not-yet-ready controller and leak threads (KAFKA-15104).
             for (ControllerServer controller : controllers.values()) {
-                futures.add(executorService.submit(controller::startup));
+                controllerFutures.add(executorService.submit(controller::startup));
+            }
+            for (Future<?> future : controllerFutures) {
+                future.get();
             }
             for (BrokerServer broker : brokers.values()) {
-                futures.add(executorService.submit(broker::startup));
+                brokerFutures.add(executorService.submit(broker::startup));
             }
-            for (Future<?> future: futures) {
+            for (Future<?> future : brokerFutures) {
                 future.get();
             }
         } catch (Exception e) {
-            for (Future<?> future: futures) {
-                future.cancel(true);
-            }
+            brokerFutures.forEach(future -> future.cancel(true));
+            controllerFutures.forEach(future -> future.cancel(true));
             throw e;
         }
     }

tools/src/test/java/org/apache/kafka/tools/MetadataQuorumCommandTest.java

@@ -20,7 +20,6 @@
 import org.apache.kafka.common.test.ClusterInstance;
 import org.apache.kafka.common.test.api.ClusterTest;
 import org.apache.kafka.common.test.api.ClusterTests;
-import org.apache.kafka.common.test.api.Flaky;
 import org.apache.kafka.common.test.api.Type;
 import org.apache.kafka.test.TestUtils;
 
@@ -43,7 +42,6 @@ class MetadataQuorumCommandTest {
      * 2. More brokers than controllers
      * 3. Fewer brokers than controllers
      */
-    @Flaky("KAFKA-15104")
     @ClusterTests({
         @ClusterTest(brokers = 2, controllers = 2),
         @ClusterTest(brokers = 2, controllers = 1),