MINOR: Cleanup Trogdor Module

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;
 
@@ -276,7 +275,7 @@ public static void main(String[] args) throws Exception {
                         dateString(status.serverStartMs(), localOffset));
                     List<List<String>> lines = new ArrayList<>();
                     List<String> header = new ArrayList<>(
-                        Arrays.asList("WORKER_ID", "TASK_ID", "STATE", "TASK_TYPE"));
+                        List.of("WORKER_ID", "TASK_ID", "STATE", "TASK_TYPE"));
                     lines.add(header);
                     for (Map.Entry<Long, WorkerState> entry : status.workers().entrySet()) {
                         List<String> cols = new ArrayList<>();

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 = Set.of();
         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;
@@ -472,7 +471,7 @@ static String prettyPrintTasksResponse(TasksResponse response, ZoneOffset zoneOf
         }
         List<List<String>> lines = new ArrayList<>();
         List<String> header = new ArrayList<>(
-            Arrays.asList("ID", "TYPE", "STATE", "INFO"));
+            List.of("ID", "TYPE", "STATE", "INFO"));
         lines.add(header);
         for (Map.Entry<String, TaskState> entry : response.tasks().entrySet()) {
             String taskId = entry.getKey();

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,14 +23,13 @@
 /**
  * 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;
     }
 
+    @Override
     @JsonProperty
     public String taskId() {
         return taskId;

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

@@ -21,7 +21,6 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HashSet;
 import java.util.Optional;
 import java.util.Set;
@@ -69,8 +68,7 @@ public TasksRequest(@JsonProperty("taskIds") Collection<String> taskIds,
             @JsonProperty("firstEndMs") long firstEndMs,
             @JsonProperty("lastEndMs") long lastEndMs,
             @JsonProperty("state") Optional<TaskStateType> state) {
-        this.taskIds = Collections.unmodifiableSet((taskIds == null) ?
-            new HashSet<>() : new HashSet<>(taskIds));
+        this.taskIds = Set.copyOf((taskIds == null) ? new HashSet<>() : new HashSet<>(taskIds));
         this.firstStartMs = Math.max(0, firstStartMs);
         this.lastStartMs = Math.max(0, lastStartMs);
         this.firstEndMs = Math.max(0, firstEndMs);

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

@@ -20,7 +20,6 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
-import java.util.Collections;
 import java.util.Map;
 import java.util.TreeMap;
 
@@ -32,7 +31,7 @@ public class TasksResponse extends Message {
 
     @JsonCreator
     public TasksResponse(@JsonProperty("tasks") TreeMap<String, TaskState> tasks) {
-        this.tasks = Collections.unmodifiableMap((tasks == null) ? new TreeMap<>() : tasks);
+        this.tasks = Map.copyOf((tasks == null) ? new TreeMap<>() : tasks);
     }
 
     @JsonProperty

trogdor/src/main/java/org/apache/kafka/trogdor/task/TaskSpec.java

@@ -22,7 +22,6 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 
-import java.util.Collections;
 import java.util.Map;
 import java.util.Objects;
 
@@ -112,6 +111,6 @@ public String toString() {
     }
 
     protected static Map<String, String> configOrEmptyMap(Map<String, String> config) {
-        return (config == null) ? Collections.emptyMap() : config;
+        return (config == null) ? Map.of() : config;
     }
 }

trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConfigurableProducerSpec.java

@@ -24,9 +24,9 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
-import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 
 /**
  * This is the spec to pass in to be able to run the `ConfigurableProducerWorker` workload.  This allows for customized
@@ -206,7 +206,7 @@ public int activePartition() {
 
     @Override
     public TaskController newController(String id) {
-        return topology -> Collections.singleton(producerNode);
+        return topology -> Set.of(producerNode);
     }
 
     @Override

trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConnectionStressSpec.java

@@ -24,7 +24,6 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.TreeSet;
@@ -56,7 +55,7 @@ public ConnectionStressSpec(@JsonProperty("startMs") long startMs,
             @JsonProperty("numThreads") int numThreads,
             @JsonProperty("action") ConnectionStressAction action) {
         super(startMs, durationMs);
-        this.clientNodes = (clientNodes == null) ? Collections.emptyList() :
+        this.clientNodes = (clientNodes == null) ? List.of() :
             List.copyOf(clientNodes);
         this.bootstrapServers = (bootstrapServers == null) ? "" : bootstrapServers;
         this.commonClientConf = configOrEmptyMap(commonClientConf);

trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConnectionStressWorker.java

@@ -129,13 +129,10 @@ private static class ConnectStressThrottle extends Throttle {
 
     interface Stressor extends AutoCloseable {
         static Stressor fromSpec(ConnectionStressSpec spec) {
-            switch (spec.action()) {
-                case CONNECT:
-                    return new ConnectStressor(spec);
-                case FETCH_METADATA:
-                    return new FetchMetadataStressor(spec);
-            }
-            throw new RuntimeException("invalid spec.action " + spec.action());
+            return switch (spec.action()) {
+                case CONNECT -> new ConnectStressor(spec);
+                case FETCH_METADATA -> new FetchMetadataStressor(spec);
+            };
         }
 
         boolean tryConnect();

trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConsumeBenchSpec.java

@@ -29,7 +29,6 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -189,7 +188,7 @@ public List<String> activeTopics() {
 
     @Override
     public TaskController newController(String id) {
-        return topology -> Collections.singleton(consumerNode);
+        return topology -> Set.of(consumerNode);
     }
 
     @Override

trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConsumeBenchWorker.java

@@ -58,7 +58,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.ReentrantLock;
-import java.util.stream.Collectors;
 
 
 public class ConsumeBenchWorker implements TaskWorker {
@@ -132,7 +131,7 @@ private List<ConsumeMessages> consumeTasks() {
                 }
             } else {
                 List<TopicPartition> partitions = populatePartitionsByTopic(consumer.consumer(), partitionsByTopic)
-                    .values().stream().flatMap(List::stream).collect(Collectors.toList());
+                    .values().stream().flatMap(List::stream).toList();
                 tasks.add(new ConsumeMessages(consumer, spec.recordProcessor(), partitions));
 
                 for (int i = 0; i < consumerCount - 1; i++) {
@@ -182,7 +181,7 @@ private Map<String, List<TopicPartition>> populatePartitionsByTopic(KafkaConsume
                 if (partitions.isEmpty()) {
                     List<TopicPartition> fetchedPartitions = consumer.partitionsFor(topicName).stream()
                         .map(partitionInfo -> new TopicPartition(partitionInfo.topic(), partitionInfo.partition()))
-                        .collect(Collectors.toList());
+                        .toList();
                     partitions.addAll(fetchedPartitions);
                 }
 
@@ -550,7 +549,7 @@ List<String> assignedPartitions() {
             this.consumerLock.lock();
             try {
                 return consumer.assignment().stream()
-                    .map(TopicPartition::toString).collect(Collectors.toList());
+                    .map(TopicPartition::toString).toList();
             } finally {
                 this.consumerLock.unlock();
             }

trogdor/src/main/java/org/apache/kafka/trogdor/workload/ExternalCommandSpec.java

@@ -26,10 +26,9 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.NullNode;
 
-import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 
 /**
  * ExternalCommandSpec describes a task that executes Trogdor tasks with the command.
@@ -78,7 +77,7 @@ public ExternalCommandSpec(
             @JsonProperty("shutdownGracePeriodMs") Optional<Integer> shutdownGracePeriodMs) {
         super(startMs, durationMs);
         this.commandNode = (commandNode == null) ? "" : commandNode;
-        this.command = (command == null) ? Collections.unmodifiableList(new ArrayList<>()) : command;
+        this.command = (command == null) ? List.of() : command;
         this.workload = (workload == null) ? NullNode.instance : workload;
         this.shutdownGracePeriodMs = shutdownGracePeriodMs;
     }
@@ -105,7 +104,7 @@ public Optional<Integer> shutdownGracePeriodMs() {
 
     @Override
     public TaskController newController(String id) {
-        return topology -> Collections.singleton(commandNode);
+        return topology -> Set.of(commandNode);
     }
 
     @Override