MINOR: Cleanup Trogdor Module (#20214)

Now that Kafka support Java 17, this PR makes some changes in `trogdor`
module. 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 cleanups around use of enhanced switch blocks and conversion
of classes to record classes.

Reviewers: Ken Huang <[email protected]>, Vincent Jiang
 <[email protected]>, Chia-Ping Tsai <[email protected]>

Author: sjhajharia

trogdor/src/main/java/org/apache/kafka/trogdor/agent/AgentClient.java

@@ -45,7 +45,6 @@
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 
@@ -275,9 +274,7 @@ public static void main(String[] args) throws Exception {
                     System.out.printf("\tStart time: %s%n",
                         dateString(status.serverStartMs(), localOffset));
                     List<List<String>> lines = new ArrayList<>();
-                    List<String> header = new ArrayList<>(
-                        Arrays.asList("WORKER_ID", "TASK_ID", "STATE", "TASK_TYPE"));
-                    lines.add(header);
+                    lines.add(List.of("WORKER_ID", "TASK_ID", "STATE", "TASK_TYPE"));
                     for (Map.Entry<Long, WorkerState> entry : status.workers().entrySet()) {
                         List<String> cols = new ArrayList<>();
                         cols.add(Long.toString(entry.getKey()));

trogdor/src/main/java/org/apache/kafka/trogdor/basic/BasicNode.java

@@ -21,7 +21,6 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -46,7 +45,7 @@ public BasicNode(String name, String hostname, Map<String, String> config,
     public BasicNode(String name, JsonNode root) {
         this.name = name;
         String hostname = "localhost";
-        Set<String> tags = Collections.emptySet();
+        Set<String> tags = new HashSet<>();
         Map<String, String> config = new HashMap<>();
         for (Map.Entry<String, JsonNode> entry : root.properties()) {
             String key = entry.getKey();

trogdor/src/main/java/org/apache/kafka/trogdor/common/StringFormatter.java

@@ -102,9 +102,7 @@ public static String prettyPrintGrid(List<List<String>> lines) {
                 String val = cols.get(x);
                 int minWidth = widths.get(x);
                 bld.append(val);
-                for (int i = 0; i < minWidth - val.length(); i++) {
-                    bld.append(" ");
-                }
+                bld.append(" ".repeat(Math.max(0, minWidth - val.length())));
             }
             bld.append(String.format("%n"));
         }

trogdor/src/main/java/org/apache/kafka/trogdor/coordinator/CoordinatorClient.java

@@ -55,7 +55,6 @@
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -471,9 +470,7 @@ static String prettyPrintTasksResponse(TasksResponse response, ZoneOffset zoneOf
             return "No matching tasks found.";
         }
         List<List<String>> lines = new ArrayList<>();
-        List<String> header = new ArrayList<>(
-            Arrays.asList("ID", "TYPE", "STATE", "INFO"));
-        lines.add(header);
+        lines.add(List.of("ID", "TYPE", "STATE", "INFO"));
         for (Map.Entry<String, TaskState> entry : response.tasks().entrySet()) {
             String taskId = entry.getKey();
             TaskState taskState = entry.getValue();

trogdor/src/main/java/org/apache/kafka/trogdor/coordinator/TaskManager.java

@@ -256,17 +256,12 @@ void maybeSetError(String newError) {
         }
 
         TaskState taskState() {
-            switch (state) {
-                case PENDING:
-                    return new TaskPending(spec);
-                case RUNNING:
-                    return new TaskRunning(spec, startedMs, getCombinedStatus());
-                case STOPPING:
-                    return new TaskStopping(spec, startedMs, getCombinedStatus());
-                case DONE:
-                    return new TaskDone(spec, startedMs, doneMs, error, cancelled, getCombinedStatus());
-            }
-            throw new RuntimeException("unreachable");
+            return switch (state) {
+                case PENDING -> new TaskPending(spec);
+                case RUNNING -> new TaskRunning(spec, startedMs, getCombinedStatus());
+                case STOPPING -> new TaskStopping(spec, startedMs, getCombinedStatus());
+                case DONE -> new TaskDone(spec, startedMs, doneMs, error, cancelled, getCombinedStatus());
+            };
         }
 
         private JsonNode getCombinedStatus() {

trogdor/src/main/java/org/apache/kafka/trogdor/fault/DegradedNetworkFaultSpec.java

@@ -24,7 +24,6 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
-import java.util.Collections;
 import java.util.Map;
 
 public class DegradedNetworkFaultSpec extends TaskSpec {
@@ -75,7 +74,7 @@ public DegradedNetworkFaultSpec(@JsonProperty("startMs") long startMs,
                                     @JsonProperty("durationMs") long durationMs,
                                     @JsonProperty("nodeSpecs") Map<String, NodeDegradeSpec> nodeSpecs) {
         super(startMs, durationMs);
-        this.nodeSpecs = nodeSpecs == null ? Collections.emptyMap() : Collections.unmodifiableMap(nodeSpecs);
+        this.nodeSpecs = nodeSpecs == null ? Map.of() : Map.copyOf(nodeSpecs);
     }
 
     @Override

trogdor/src/main/java/org/apache/kafka/trogdor/fault/Kibosh.java

@@ -29,7 +29,6 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.TreeMap;
@@ -123,7 +122,7 @@ public static class KiboshControlFile {
         private final List<KiboshFaultSpec> faults;
 
         public static final KiboshControlFile EMPTY =
-            new KiboshControlFile(Collections.emptyList());
+            new KiboshControlFile(List.of());
 
         public static KiboshControlFile read(Path controlPath) throws IOException {
             byte[] controlFileBytes = Files.readAllBytes(controlPath);

trogdor/src/main/java/org/apache/kafka/trogdor/fault/NetworkPartitionFaultWorker.java

@@ -72,9 +72,7 @@ private void runIptablesCommands(Platform platform, String iptablesAction) throw
         TreeSet<String> toBlock = new TreeSet<>();
         for (Set<String> partitionSet : partitionSets) {
             if (!partitionSet.contains(curNode.name())) {
-                for (String nodeName : partitionSet) {
-                    toBlock.add(nodeName);
-                }
+                toBlock.addAll(partitionSet);
             }
         }
         for (String nodeName : toBlock) {

trogdor/src/main/java/org/apache/kafka/trogdor/rest/ErrorResponse.java

@@ -22,46 +22,29 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
-import java.util.Objects;
-
 /**
  * An error response.
  */
-public class ErrorResponse {
-    private final int code;
-    private final String message;
-
+public record ErrorResponse(int code, String message) {
     @JsonCreator
     public ErrorResponse(@JsonProperty("code") int code,
                          @JsonProperty("message") String message) {
         this.code = code;
         this.message = message;
     }
 
+    @Override
     @JsonProperty
     public int code() {
         return code;
     }
 
+    @Override
     @JsonProperty
     public String message() {
         return message;
     }
 
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) return true;
-        if (o == null || getClass() != o.getClass()) return false;
-        ErrorResponse that = (ErrorResponse) o;
-        return Objects.equals(code, that.code) &&
-            Objects.equals(message, that.message);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(code, message);
-    }
-
     @Override
     public String toString() {
         return JsonUtil.toJsonString(this);

trogdor/src/main/java/org/apache/kafka/trogdor/rest/TaskRequest.java

@@ -23,16 +23,9 @@
 /**
  * The request to /coordinator/tasks/{taskId}
  */
-public class TaskRequest {
-    private final String taskId;
-
+public record TaskRequest(String taskId) {
     @JsonCreator
     public TaskRequest(@JsonProperty("taskId") String taskId) {
         this.taskId = taskId == null ? "" : taskId;
     }
-
-    @JsonProperty
-    public String taskId() {
-        return taskId;
-    }
 }