KAFKA-16487 Support to define server properties by ClusterTestDefaults (

#15687)

Sometimes we want to define server properties for all test cases, and using BeforeEach to modify the ClusterConfig is the only way. The side effect is that the IDE does not like the style since IDE can't recognize custom ParameterResolver of ClusterConfig.

The alternative is that we can take ClusterInstance from constructor first, and then we modify the inner ClusterConfig in BeforeEach phase. However, that may confuse users about the life cycle of "configs".

In short, I prefer to define the server property by ClusterTestDefaults. It already includes some server-side default property, and we can enhance that to deal with more existent test case.

Reviewers: Chia-Ping Tsai <[email protected]>

core/src/test/java/kafka/test/ClusterTestExtensionsTest.java

@@ -32,7 +32,9 @@
 import org.junit.jupiter.api.extension.ExtendWith;
 
 
-@ClusterTestDefaults(clusterType = Type.ZK)   // Set defaults for a few params in @ClusterTest(s)
+@ClusterTestDefaults(clusterType = Type.ZK, serverProperties = {
+    @ClusterConfigProperty(key = "default.key", value = "default.value"),
+})   // Set defaults for a few params in @ClusterTest(s)
 @ExtendWith(ClusterTestExtensions.class)
 public class ClusterTestExtensionsTest {
 
@@ -89,20 +91,24 @@ public void testClusterTemplate() {
         }),
         @ClusterTest(name = "cluster-tests-2", clusterType = Type.KRAFT, serverProperties = {
             @ClusterConfigProperty(key = "foo", value = "baz"),
-            @ClusterConfigProperty(key = "spam", value = "eggz")
+            @ClusterConfigProperty(key = "spam", value = "eggz"),
+            @ClusterConfigProperty(key = "default.key", value = "overwrite.value")
         }),
         @ClusterTest(name = "cluster-tests-3", clusterType = Type.CO_KRAFT, serverProperties = {
             @ClusterConfigProperty(key = "foo", value = "baz"),
-            @ClusterConfigProperty(key = "spam", value = "eggz")
+            @ClusterConfigProperty(key = "spam", value = "eggz"),
+            @ClusterConfigProperty(key = "default.key", value = "overwrite.value")
         })
     })
     public void testClusterTests() {
         if (clusterInstance.clusterType().equals(ClusterInstance.ClusterType.ZK)) {
-            Assertions.assertEquals(clusterInstance.config().serverProperties().getProperty("foo"), "bar");
-            Assertions.assertEquals(clusterInstance.config().serverProperties().getProperty("spam"), "eggs");
+            Assertions.assertEquals("bar", clusterInstance.config().serverProperties().getProperty("foo"));
+            Assertions.assertEquals("eggs", clusterInstance.config().serverProperties().getProperty("spam"));
+            Assertions.assertEquals("default.value", clusterInstance.config().serverProperties().getProperty("default.key"));
         } else if (clusterInstance.clusterType().equals(ClusterInstance.ClusterType.RAFT)) {
-            Assertions.assertEquals(clusterInstance.config().serverProperties().getProperty("foo"), "baz");
-            Assertions.assertEquals(clusterInstance.config().serverProperties().getProperty("spam"), "eggz");
+            Assertions.assertEquals("baz", clusterInstance.config().serverProperties().getProperty("foo"));
+            Assertions.assertEquals("eggz", clusterInstance.config().serverProperties().getProperty("spam"));
+            Assertions.assertEquals("overwrite.value", clusterInstance.config().serverProperties().getProperty("default.key"));
         } else {
             Assertions.fail("Unknown cluster type " + clusterInstance.clusterType());
         }

core/src/test/java/kafka/test/annotation/ClusterTestDefaults.java

@@ -39,4 +39,6 @@
     int brokers() default 1;
     int controllers() default 1;
     boolean autoStart() default true;
+    // Set default server properties for all @ClusterTest(s)
+    ClusterConfigProperty[] serverProperties() default {};
 }

core/src/test/java/kafka/test/junit/ClusterTestExtensions.java

@@ -34,7 +34,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
-import java.util.Properties;
 import java.util.function.Consumer;
 import java.util.stream.Stream;
 
@@ -189,13 +188,13 @@ private void processClusterTest(ExtensionContext context, ClusterTest annot, Clu
             builder.listenerName(annot.listener());
         }
 
-        Properties properties = new Properties();
+        ClusterConfig config = builder.build();
+        for (ClusterConfigProperty property : defaults.serverProperties()) {
+            config.serverProperties().put(property.key(), property.value());
+        }
         for (ClusterConfigProperty property : annot.serverProperties()) {
-            properties.put(property.key(), property.value());
+            config.serverProperties().put(property.key(), property.value());
         }
-
-        ClusterConfig config = builder.build();
-        config.serverProperties().putAll(properties);
         type.invocationContexts(context.getRequiredTestMethod().getName(), config, testInvocations);
     }