MINOR: Cleanup Tools Module (2/n) (#20096)

Now that Kafka support Java 17, this PR makes some changes in tools
module. The changes in this PR are limited to only some files. A future
PR(s) shall follow. The changes mostly include:

- Collections.emptyList(), Collections.singletonList() and
Arrays.asList() are replaced with List.of()
- Collections.emptyMap() and Collections.singletonMap() are replaced
with Map.of()
- Collections.singleton() is replaced with Set.of()

Some minor changes to use the enhanced switch.

Sub modules targeted: tools/src/test

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

Author: sjhajharia

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

@@ -51,12 +51,10 @@
 import java.io.File;
 import java.io.FileWriter;
 import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
-import java.util.stream.Collectors;
+import java.util.Set;
 
 import static java.time.Duration.ofMillis;
 import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitForEmptyConsumerGroup;
@@ -176,7 +174,7 @@ void cleanupTest() throws Exception {
     }
 
     private void add10InputElements() {
-        final List<KeyValue<Long, String>> records = Arrays.asList(KeyValue.pair(0L, "aaa"),
+        final List<KeyValue<Long, String>> records = List.of(KeyValue.pair(0L, "aaa"),
                                                                    KeyValue.pair(1L, "bbb"),
                                                                    KeyValue.pair(0L, "ccc"),
                                                                    KeyValue.pair(1L, "ddd"),
@@ -189,7 +187,7 @@ private void add10InputElements() {
 
         for (final KeyValue<Long, String> record : records) {
             mockTime.sleep(10);
-            IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(record), producerConfig, mockTime.milliseconds());
+            IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Set.of(record), producerConfig, mockTime.milliseconds());
         }
     }
 
@@ -211,7 +209,7 @@ public void testResetWhenInternalTopicsAreSpecified(final TestInfo testInfo) thr
 
         final List<String> internalTopics = cluster.getAllTopicsInCluster().stream()
                 .filter(StreamsResetter::matchesInternalTopicFormat)
-                .collect(Collectors.toList());
+                .toList();
         cleanGlobal(false,
                 "--internal-topics",
                 String.join(",", internalTopics.subList(1, internalTopics.size())),
@@ -288,7 +286,7 @@ private void testReprocessingFromScratchAfterResetWithIntermediateUserTopic(fina
         if (!useRepartitioned) {
             IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
                 INTERMEDIATE_USER_TOPIC,
-                Collections.singleton(badMessage),
+                Set.of(badMessage),
                 producerConfig,
                 mockTime.milliseconds());
         }
@@ -375,7 +373,7 @@ protected boolean tryCleanGlobal(final boolean withIntermediateTopics,
                                    final String resetScenarioArg,
                                    final String appID) throws Exception {
         final List<String> parameterList = new ArrayList<>(
-            Arrays.asList("--application-id", appID,
+            List.of("--application-id", appID,
                     "--bootstrap-server", cluster.bootstrapServers(),
                     "--input-topics", INPUT_TOPIC
             ));

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

@@ -31,8 +31,8 @@
 import org.apache.kafka.server.config.ServerConfigs;
 
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Properties;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -56,7 +56,7 @@ public void testBrokerApiVersionsCommandOutput(ClusterInstance clusterInstance)
 
         NodeApiVersions nodeApiVersions = new NodeApiVersions(
                 ApiVersionsResponse.filterApis(listenerType, true, true),
-                Collections.emptyList());
+                List.of());
         Iterator<ApiKeys> apiKeysIter = ApiKeys.clientApis().iterator();
         while (apiKeysIter.hasNext()) {
             ApiKeys apiKey = apiKeysIter.next();

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

@@ -33,7 +33,6 @@
 import org.mockito.ArgumentCaptor;
 
 import java.util.Collection;
-import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
@@ -209,7 +208,7 @@ public void testDelete() {
         ClientMetricsCommand.ClientMetricsService service = new ClientMetricsCommand.ClientMetricsService(adminClient);
 
         ConfigResource cr = new ConfigResource(ConfigResource.Type.CLIENT_METRICS, clientMetricsName);
-        Config cfg = new Config(Collections.singleton(new ConfigEntry("metrics", "org.apache.kafka.producer.")));
+        Config cfg = new Config(Set.of(new ConfigEntry("metrics", "org.apache.kafka.producer.")));
         DescribeConfigsResult describeResult = AdminClientTestUtils.describeConfigsResult(cr, cfg);
         when(adminClient.describeConfigs(any())).thenReturn(describeResult);
         AlterConfigsResult alterResult = AdminClientTestUtils.alterConfigsResult(cr);
@@ -237,7 +236,7 @@ public void testDescribe() {
             ConfigResource.Type.CLIENT_METRICS, Set.of(clientMetricsName)
         ));
         when(adminClient.listConfigResources(any(), any())).thenReturn(listConfigResourcesResult);
-        Config cfg = new Config(Collections.singleton(new ConfigEntry("metrics", "org.apache.kafka.producer.")));
+        Config cfg = new Config(Set.of(new ConfigEntry("metrics", "org.apache.kafka.producer.")));
         DescribeConfigsResult describeResult = AdminClientTestUtils.describeConfigsResult(cr, cfg);
         when(adminClient.describeConfigs(any())).thenReturn(describeResult);
 
@@ -284,7 +283,7 @@ public void testDescribeAll() {
         ListConfigResourcesResult result = AdminClientTestUtils.listConfigResourcesResult(clientMetricsName);
         when(adminClient.listConfigResources(any(), any())).thenReturn(result);
         ConfigResource cr = new ConfigResource(ConfigResource.Type.CLIENT_METRICS, clientMetricsName);
-        Config cfg = new Config(Collections.singleton(new ConfigEntry("metrics", "org.apache.kafka.producer.")));
+        Config cfg = new Config(Set.of(new ConfigEntry("metrics", "org.apache.kafka.producer.")));
         DescribeConfigsResult describeResult = AdminClientTestUtils.describeConfigsResult(cr, cfg);
         when(adminClient.describeConfigs(any())).thenReturn(describeResult);
 
@@ -326,7 +325,7 @@ public void testListFailsWithUnsupportedVersionException() {
         ListConfigResourcesResult result = AdminClientTestUtils.listConfigResourcesResult(Errors.UNSUPPORTED_VERSION.exception());
         when(adminClient.listConfigResources(any(), any())).thenReturn(result);
 
-        assertThrows(ExecutionException.class, () -> service.listClientMetrics());
+        assertThrows(ExecutionException.class, service::listClientMetrics);
     }
 
     private void assertInitializeInvalidOptionsExitCode(int expected, String[] options) {

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

@@ -29,7 +29,6 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -71,10 +70,10 @@ public void testListEndpointsWithBootstrapServer(ClusterInstance clusterInstance
 
     @ClusterTest(brokers = 2, types = {Type.KRAFT, Type.CO_KRAFT})
     public void testListEndpointsArgumentWithBootstrapServer(ClusterInstance clusterInstance) {
-        List<Integer> brokerIds = clusterInstance.brokerIds().stream().collect(Collectors.toList());
+        List<Integer> brokerIds = clusterInstance.brokerIds().stream().toList();
         clusterInstance.shutdownBroker(brokerIds.get(0));
 
-        List<String> ports = Arrays.stream(clusterInstance.bootstrapServers().split(",")).map(b ->  b.split(":")[1]).collect(Collectors.toList());
+        List<String> ports = Arrays.stream(clusterInstance.bootstrapServers().split(",")).map(b ->  b.split(":")[1]).toList();
         String format = "%-10s %-9s %-10s %-10s %-10s %-15s%n%-10s %-9s %-10s %-10s %-10s %-15s%n%-10s %-9s %-10s %-10s %-10s %-6s";
         String expected = String.format(format,
                 "ID", "HOST", "PORT", "RACK", "STATE", "ENDPOINT_TYPE",
@@ -110,7 +109,7 @@ public void testListEndpointsWithBootstrapController(ClusterInstance clusterInst
         int id = clusterInstance.controllerIds().iterator().next();
         String format = "%-10s %-9s %-10s %-10s %-15s%n%-10s %-9s %-10s %-10s %-10s";
         String expected = String.format(format, "ID", "HOST", "PORT", "RACK", "ENDPOINT_TYPE", id, "localhost", port, "null", "controller");
-        assertTrue(output.equals(expected));
+        assertEquals(expected, output);
     }
 
     @ClusterTest(brokers = 3, types = {Type.KRAFT, Type.CO_KRAFT})