KAFKA-18617 Allow use of ClusterInstance inside BeforeEach

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

@@ -247,8 +247,12 @@ default Set<GroupProtocol> supportedGroupProtocols() {
 
     void start();
 
+    boolean started();
+
     void stop();
 
+    boolean stopped();
+
     void shutdownBroker(int brokerId);
 
     void startBroker(int brokerId);

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

@@ -35,7 +35,7 @@
 import org.apache.kafka.server.fault.FaultHandlerException;
 
 import org.junit.jupiter.api.extension.AfterTestExecutionCallback;
-import org.junit.jupiter.api.extension.BeforeTestExecutionCallback;
+import org.junit.jupiter.api.extension.BeforeEachCallback;
 import org.junit.jupiter.api.extension.Extension;
 import org.junit.jupiter.api.extension.TestTemplateInvocationContext;
 
@@ -87,7 +87,7 @@ public String getDisplayName(int invocationIndex) {
     public List<Extension> getAdditionalExtensions() {
         RaftClusterInstance clusterInstance = new RaftClusterInstance(clusterConfig, isCombined);
         return Arrays.asList(
-                (BeforeTestExecutionCallback) context -> {
+                (BeforeEachCallback) context -> {
                     clusterInstance.format();
                     if (clusterConfig.isAutoStart()) {
                         clusterInstance.start();
@@ -186,13 +186,23 @@ public void start() {
             }
         }
 
+        @Override
+        public boolean started() {
+            return started.get();
+        }
+
         @Override
         public void stop() {
             if (stopped.compareAndSet(false, true)) {
                 Utils.closeQuietly(clusterTestKit, "cluster");
             }
         }
 
+        @Override
+        public boolean stopped() {
+            return stopped.get();
+        }
+
         @Override
         public void shutdownBroker(int brokerId) {
             findBrokerOrThrow(brokerId).shutdown();

test-common/test-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestBeforeEachTest.java

@@ -0,0 +1,38 @@
+package org.apache.kafka.common.test.api;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ExtendWith(ClusterTestExtensions.class)
+public class ClusterTestBeforeEachTest {
+    private final ClusterInstance clusterInstance;
+
+    ClusterTestBeforeEachTest(ClusterInstance clusterInstance) {     // Constructor injections
+        this.clusterInstance = clusterInstance;
+    }
+
+    @BeforeEach
+    void before() {
+        assertNotNull(clusterInstance);
+        if (!clusterInstance.started()) {
+            clusterInstance.start();
+        }
+        assertDoesNotThrow(clusterInstance::waitForReadyBrokers);
+    }
+
+    @ClusterTest
+    public void testDefault() {
+        assertTrue(true);
+        assertNotNull(clusterInstance);
+    }
+
+    @ClusterTest(autoStart = AutoStart.NO)
+    public void testNoAutoStart() {
+        assertTrue(true);
+        assertNotNull(clusterInstance);
+    }
+}