MINOR: Cleanup test-common-runtime module (#19305)

This patch does some cleanups for the `test-common-runtime`.

- `.collect(Collectors.toList())` -> `.toList()`
- lambda block -> lambda expression
- remove unnecessary semicolon
- switch block -> switch expression
- add `final` keyword

Reviewers: PoAn Yang <[email protected]>, Ken Huang
<[email protected]>, TaiJuWu <[email protected]>, Chia-Ping Tsai
<[email protected]>

Author: frankvicky

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

@@ -307,7 +307,7 @@ default void waitForTopic(String topic, int partitions) throws InterruptedExcept
         if (partitions == 0) {
             List<TopicPartition> topicPartitions = IntStream.range(0, 1)
                 .mapToObj(partition -> new TopicPartition(topic, partition))
-                .collect(Collectors.toList());
+                .toList();
 
             // Ensure that the topic-partition has been deleted from all brokers' replica managers
             TestUtils.waitForCondition(() -> brokers.stream().allMatch(broker ->
@@ -358,10 +358,10 @@ default List<Authorizer> authorizers() {
         List<Authorizer> authorizers = new ArrayList<>();
         authorizers.addAll(brokers().values().stream()
                 .filter(server -> server.authorizer().isDefined())
-                .map(server -> server.authorizer().get()).collect(Collectors.toList()));
+                .map(server -> server.authorizer().get()).toList());
         authorizers.addAll(controllers().values().stream()
                 .filter(server -> server.authorizer().isDefined())
-                .map(server -> server.authorizer().get()).collect(Collectors.toList()));
+                .map(server -> server.authorizer().get()).toList());
         return authorizers;
     }
 

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

@@ -49,7 +49,7 @@ public String toString() {
     public static File writeJaasContextsToFile(Set<JaasSection> jaasSections) throws IOException {
         File jaasFile = TestUtils.tempFile();
         try (FileOutputStream fileStream = new FileOutputStream(jaasFile);
-             OutputStreamWriter writer = new OutputStreamWriter(fileStream, StandardCharsets.UTF_8);) {
+             OutputStreamWriter writer = new OutputStreamWriter(fileStream, StandardCharsets.UTF_8)) {
             writer.write(String.join("", jaasSections.stream().map(Object::toString).toArray(String[]::new)));
         }
         return jaasFile;

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

@@ -391,16 +391,12 @@ public void format() throws Exception {
         List<Future<?>> futures = new ArrayList<>();
         try {
             for (ControllerServer controller : controllers.values()) {
-                futures.add(executorService.submit(() -> {
-                    formatNode(controller.sharedServer().metaPropsEnsemble(), true);
-                }));
+                futures.add(executorService.submit(() -> formatNode(controller.sharedServer().metaPropsEnsemble(), true)));
             }
             for (Entry<Integer, BrokerServer> entry : brokers.entrySet()) {
                 BrokerServer broker = entry.getValue();
-                futures.add(executorService.submit(() -> {
-                    formatNode(broker.sharedServer().metaPropsEnsemble(),
-                        !nodes.isCombined(nodes().brokerNodes().get(entry.getKey()).id()));
-                }));
+                futures.add(executorService.submit(() -> formatNode(broker.sharedServer().metaPropsEnsemble(),
+                    !nodes.isCombined(nodes().brokerNodes().get(entry.getKey()).id()))));
             }
             for (Future<?> future: futures) {
                 future.get();
@@ -502,7 +498,7 @@ public void waitForReadyBrokers() throws ExecutionException, InterruptedExceptio
     }
 
     public class ClientPropertiesBuilder {
-        private Properties properties;
+        private final Properties properties;
         private boolean usingBootstrapControllers = false;
 
         public ClientPropertiesBuilder() {

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

@@ -175,10 +175,10 @@ public TestKitNodes build() {
             int controllerId = combined ? TestKitDefaults.BROKER_ID_OFFSET : TestKitDefaults.BROKER_ID_OFFSET + TestKitDefaults.CONTROLLER_ID_OFFSET;
             List<Integer> controllerNodeIds = IntStream.range(controllerId, controllerId + numControllerNodes)
                 .boxed()
-                .collect(Collectors.toList());
+                .toList();
             List<Integer> brokerNodeIds = IntStream.range(TestKitDefaults.BROKER_ID_OFFSET, TestKitDefaults.BROKER_ID_OFFSET + numBrokerNodes)
                 .boxed()
-                .collect(Collectors.toList());
+                .toList();
 
             String unknownIds = perServerProperties.keySet().stream()
                     .filter(id -> !controllerNodeIds.contains(id))
@@ -316,7 +316,7 @@ private static TestKitNode buildBrokerNode(int id,
                 }
                 return new File(baseDirectory, logDir).getAbsolutePath();
             })
-            .collect(Collectors.toList());
+            .toList();
         MetaPropertiesEnsemble.Copier copier = new MetaPropertiesEnsemble.Copier(MetaPropertiesEnsemble.EMPTY);
 
         copier.setMetaLogDir(Optional.of(logDataDirectories.get(0)));

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

@@ -194,14 +194,10 @@ private TestTemplateInvocationContext invocationContextForClusterType(
         String baseDisplayName,
         ClusterConfig config
     ) {
-        switch (type) {
-            case KRAFT:
-                return new RaftClusterInvocationContext(baseDisplayName, config, false);
-            case CO_KRAFT:
-                return new RaftClusterInvocationContext(baseDisplayName, config, true);
-            default:
-                throw new IllegalArgumentException("Unsupported @Type value " + type);
-        }
+        return switch (type) {
+            case KRAFT -> new RaftClusterInvocationContext(baseDisplayName, config, false);
+            case CO_KRAFT -> new RaftClusterInvocationContext(baseDisplayName, config, true);
+        };
     }
 
     List<TestTemplateInvocationContext> processClusterTemplate(ExtensionContext context, ClusterTemplate annot) {

test-common/test-common-runtime/src/test/java/org/apache/kafka/common/test/junit/ClusterTestExtensionsTest.java

@@ -186,9 +186,8 @@ public void testClusterTests() throws ExecutionException, InterruptedException {
     public void testClusterTestWithDisksPerBroker() throws ExecutionException, InterruptedException {
         try (Admin admin = clusterInstance.admin()) {
             DescribeLogDirsResult result = admin.describeLogDirs(clusterInstance.brokerIds());
-            result.allDescriptions().get().forEach((brokerId, logDirDescriptionMap) -> {
-                assertEquals(clusterInstance.config().numDisksPerBroker(), logDirDescriptionMap.size());
-            });
+            result.allDescriptions().get().forEach((brokerId, logDirDescriptionMap) ->
+                assertEquals(clusterInstance.config().numDisksPerBroker(), logDirDescriptionMap.size()));
         }
     }