diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/agent/AgentClient.java b/trogdor/src/main/java/org/apache/kafka/trogdor/agent/AgentClient.java index fb3a7d8162ffd..736ff493b2417 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/agent/AgentClient.java +++ b/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,8 +274,7 @@ public static void main(String[] args) throws Exception { System.out.printf("\tStart time: %s%n", dateString(status.serverStartMs(), localOffset)); List> lines = new ArrayList<>(); - List header = new ArrayList<>( - Arrays.asList("WORKER_ID", "TASK_ID", "STATE", "TASK_TYPE")); + List header = List.of("WORKER_ID", "TASK_ID", "STATE", "TASK_TYPE"); lines.add(header); for (Map.Entry entry : status.workers().entrySet()) { List cols = new ArrayList<>(); diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/basic/BasicNode.java b/trogdor/src/main/java/org/apache/kafka/trogdor/basic/BasicNode.java index 729031f0b841e..6750be37b0257 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/basic/BasicNode.java +++ b/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 config, public BasicNode(String name, JsonNode root) { this.name = name; String hostname = "localhost"; - Set tags = Collections.emptySet(); + Set tags = Set.of(); Map config = new HashMap<>(); for (Map.Entry entry : root.properties()) { String key = entry.getKey(); diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/common/StringFormatter.java b/trogdor/src/main/java/org/apache/kafka/trogdor/common/StringFormatter.java index 35ca59550522f..072fcf8fa633d 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/common/StringFormatter.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/common/StringFormatter.java @@ -102,9 +102,7 @@ public static String prettyPrintGrid(List> 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")); } diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/coordinator/CoordinatorClient.java b/trogdor/src/main/java/org/apache/kafka/trogdor/coordinator/CoordinatorClient.java index d5e14e959d340..c9ed9c90d54d4 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/coordinator/CoordinatorClient.java +++ b/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,8 +470,7 @@ static String prettyPrintTasksResponse(TasksResponse response, ZoneOffset zoneOf return "No matching tasks found."; } List> lines = new ArrayList<>(); - List header = new ArrayList<>( - Arrays.asList("ID", "TYPE", "STATE", "INFO")); + List header = List.of("ID", "TYPE", "STATE", "INFO"); lines.add(header); for (Map.Entry entry : response.tasks().entrySet()) { String taskId = entry.getKey(); diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/coordinator/TaskManager.java b/trogdor/src/main/java/org/apache/kafka/trogdor/coordinator/TaskManager.java index 03e78796bc8b8..6b10de71d4857 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/coordinator/TaskManager.java +++ b/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() { diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/fault/DegradedNetworkFaultSpec.java b/trogdor/src/main/java/org/apache/kafka/trogdor/fault/DegradedNetworkFaultSpec.java index 59a2eb00eb0e6..d3153b3c7f284 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/fault/DegradedNetworkFaultSpec.java +++ b/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 nodeSpecs) { super(startMs, durationMs); - this.nodeSpecs = nodeSpecs == null ? Collections.emptyMap() : Collections.unmodifiableMap(nodeSpecs); + this.nodeSpecs = nodeSpecs == null ? Map.of() : Map.copyOf(nodeSpecs); } @Override diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/fault/Kibosh.java b/trogdor/src/main/java/org/apache/kafka/trogdor/fault/Kibosh.java index eb3de14722833..862cb0f5b4198 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/fault/Kibosh.java +++ b/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 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); diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/fault/NetworkPartitionFaultWorker.java b/trogdor/src/main/java/org/apache/kafka/trogdor/fault/NetworkPartitionFaultWorker.java index 9cb8b5724a22d..331abd0038d3c 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/fault/NetworkPartitionFaultWorker.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/fault/NetworkPartitionFaultWorker.java @@ -72,9 +72,7 @@ private void runIptablesCommands(Platform platform, String iptablesAction) throw TreeSet toBlock = new TreeSet<>(); for (Set partitionSet : partitionSets) { if (!partitionSet.contains(curNode.name())) { - for (String nodeName : partitionSet) { - toBlock.add(nodeName); - } + toBlock.addAll(partitionSet); } } for (String nodeName : toBlock) { diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/rest/ErrorResponse.java b/trogdor/src/main/java/org/apache/kafka/trogdor/rest/ErrorResponse.java index fa8df147b8423..1936ea84787a2 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/rest/ErrorResponse.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/rest/ErrorResponse.java @@ -22,15 +22,10 @@ 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) { @@ -38,30 +33,18 @@ public ErrorResponse(@JsonProperty("code") int 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); diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/rest/TaskRequest.java b/trogdor/src/main/java/org/apache/kafka/trogdor/rest/TaskRequest.java index e42738f558785..f435176a378ac 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/rest/TaskRequest.java +++ b/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; diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/rest/TasksRequest.java b/trogdor/src/main/java/org/apache/kafka/trogdor/rest/TasksRequest.java index 7ed3d7c8d65f6..fa903580478d6 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/rest/TasksRequest.java +++ b/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 taskIds, @JsonProperty("firstEndMs") long firstEndMs, @JsonProperty("lastEndMs") long lastEndMs, @JsonProperty("state") Optional 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); diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/rest/TasksResponse.java b/trogdor/src/main/java/org/apache/kafka/trogdor/rest/TasksResponse.java index 5f8655fd563d0..cd69b9fb5137f 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/rest/TasksResponse.java +++ b/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 tasks) { - this.tasks = Collections.unmodifiableMap((tasks == null) ? new TreeMap<>() : tasks); + this.tasks = Map.copyOf((tasks == null) ? new TreeMap<>() : tasks); } @JsonProperty diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/task/TaskSpec.java b/trogdor/src/main/java/org/apache/kafka/trogdor/task/TaskSpec.java index f7a538195b888..115dcdbe8390f 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/task/TaskSpec.java +++ b/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 configOrEmptyMap(Map config) { - return (config == null) ? Collections.emptyMap() : config; + return (config == null) ? Map.of() : config; } } diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConfigurableProducerSpec.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConfigurableProducerSpec.java index 8f6556842f529..aa4340ef2d326 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConfigurableProducerSpec.java +++ b/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 diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConnectionStressSpec.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConnectionStressSpec.java index dc0b09da96a2a..ba6568fa41464 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConnectionStressSpec.java +++ b/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); diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConnectionStressWorker.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConnectionStressWorker.java index af861d38195c7..9452c03ec9937 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConnectionStressWorker.java +++ b/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(); diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConsumeBenchSpec.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConsumeBenchSpec.java index 24869e3e966e2..12dd1bb80a0ea 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConsumeBenchSpec.java +++ b/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 activeTopics() { @Override public TaskController newController(String id) { - return topology -> Collections.singleton(consumerNode); + return topology -> Set.of(consumerNode); } @Override diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConsumeBenchWorker.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConsumeBenchWorker.java index 9c5e053293fd3..0a51db48404ea 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConsumeBenchWorker.java +++ b/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 consumeTasks() { } } else { List 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> populatePartitionsByTopic(KafkaConsume if (partitions.isEmpty()) { List 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 assignedPartitions() { this.consumerLock.lock(); try { return consumer.assignment().stream() - .map(TopicPartition::toString).collect(Collectors.toList()); + .map(TopicPartition::toString).toList(); } finally { this.consumerLock.unlock(); } diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ExternalCommandSpec.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ExternalCommandSpec.java index 838beb62cff27..dae242c22d027 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ExternalCommandSpec.java +++ b/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 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 shutdownGracePeriodMs() { @Override public TaskController newController(String id) { - return topology -> Collections.singleton(commandNode); + return topology -> Set.of(commandNode); } @Override diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/Histogram.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/Histogram.java index a8c24a446c165..33615eb5f1510 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/Histogram.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/Histogram.java @@ -18,7 +18,6 @@ package org.apache.kafka.trogdor.workload; import java.util.ArrayList; -import java.util.Collections; import java.util.List; /** @@ -175,7 +174,7 @@ public Summary summarize(float[] percentiles) { private List summarizePercentiles(int[] countsCopy, float[] percentiles, long numSamples) { if (percentiles.length == 0) { - return Collections.emptyList(); + return List.of(); } List summaries = new ArrayList<>(percentiles.length); int i = 0, j = 0; diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/PartitionsSpec.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/PartitionsSpec.java index 9cb00efcbc655..aa73af60e2bb6 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/PartitionsSpec.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/PartitionsSpec.java @@ -24,7 +24,6 @@ import com.fasterxml.jackson.annotation.JsonProperty; import java.util.ArrayList; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -59,12 +58,12 @@ public PartitionsSpec(@JsonProperty("numPartitions") int numPartitions, assignments.add(brokerId == null ? Integer.valueOf(0) : brokerId); } } - partMap.put(partition, Collections.unmodifiableList(assignments)); + partMap.put(partition, List.copyOf(assignments)); } } - this.partitionAssignments = Collections.unmodifiableMap(partMap); + this.partitionAssignments = Map.copyOf(partMap); if (configs == null) { - this.configs = Collections.emptyMap(); + this.configs = Map.of(); } else { this.configs = Map.copyOf(configs); } @@ -84,7 +83,9 @@ public List partitionNumbers() { } return partitionNumbers; } else { - return new ArrayList<>(partitionAssignments.keySet()); + ArrayList partitionNumbers = new ArrayList<>(partitionAssignments.keySet()); + partitionNumbers.sort(Integer::compareTo); + return partitionNumbers; } } diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java index 621333d74f90f..f33855c94e4ec 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.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; /** * The specification for a benchmark that produces messages to a set of topics. @@ -185,7 +185,7 @@ public boolean skipFlush() { @Override public TaskController newController(String id) { - return topology -> Collections.singleton(producerNode); + return topology -> Set.of(producerNode); } @Override diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RandomComponent.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RandomComponent.java index 5f5da5735dea0..5fff8a3b28a69 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RandomComponent.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RandomComponent.java @@ -22,13 +22,9 @@ /** * Contains a percent value represented as an integer between 1 and 100 and a PayloadGenerator to specify - * how often that PayloadGenerator should be used. + * how often that PayloadGenerator should be used. */ -public class RandomComponent { - private final int percent; - private final PayloadGenerator component; - - +public record RandomComponent(int percent, PayloadGenerator component) { @JsonCreator public RandomComponent(@JsonProperty("percent") int percent, @JsonProperty("component") PayloadGenerator component) { @@ -36,11 +32,13 @@ public RandomComponent(@JsonProperty("percent") int percent, this.component = component; } + @Override @JsonProperty public int percent() { return percent; } + @Override @JsonProperty public PayloadGenerator component() { return component; diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RoundTripWorkloadSpec.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RoundTripWorkloadSpec.java index 67fec60990ba1..fd8f522899592 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RoundTripWorkloadSpec.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RoundTripWorkloadSpec.java @@ -24,8 +24,8 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; -import java.util.Collections; import java.util.Map; +import java.util.Set; /** * The specification for a workload that sends messages to a broker and then @@ -123,7 +123,7 @@ public Map consumerConf() { @Override public TaskController newController(String id) { - return topology -> Collections.singleton(clientNode); + return topology -> Set.of(clientNode); } @Override diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpec.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpec.java index 503d2b9b8275f..559121358d7ad 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpec.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpec.java @@ -27,7 +27,6 @@ import java.util.ArrayList; import java.util.Collection; -import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -176,7 +175,7 @@ public List activeTopics() { @Override public TaskController newController(String id) { - return topology -> Collections.singleton(consumerNode); + return topology -> Set.of(consumerNode); } @Override diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/SustainedConnectionSpec.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/SustainedConnectionSpec.java index ccd94784f357d..144ce3a2d450f 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/SustainedConnectionSpec.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/SustainedConnectionSpec.java @@ -24,8 +24,8 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; -import java.util.Collections; import java.util.Map; +import java.util.Set; /** * The specification for a benchmark that creates sustained connections. @@ -187,7 +187,7 @@ public int refreshRateMs() { } public TaskController newController(String id) { - return topology -> Collections.singleton(clientNode); + return topology -> Set.of(clientNode); } @Override diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/SustainedConnectionWorker.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/SustainedConnectionWorker.java index 559a188585035..3e8c3c7fd14cd 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/SustainedConnectionWorker.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/SustainedConnectionWorker.java @@ -343,15 +343,15 @@ public void refresh() { this.consumer = new KafkaConsumer<>(this.props, new ByteArrayDeserializer(), new ByteArrayDeserializer()); List partitions = this.consumer.partitionsFor(this.topicName).stream() .map(partitionInfo -> new TopicPartition(partitionInfo.topic(), partitionInfo.partition())) - .collect(Collectors.toList()); + .toList(); // Select a random partition and assign it. this.activePartition = partitions.get(this.rand.nextInt(partitions.size())); - this.consumer.assign(Collections.singletonList(this.activePartition)); + this.consumer.assign(List.of(this.activePartition)); } // The behavior when passing in an empty list is to seek to the end of all subscribed partitions. - this.consumer.seekToEnd(Collections.emptyList()); + this.consumer.seekToEnd(List.of()); // Poll to keep the connection alive, ignoring any records returned. this.consumer.poll(Duration.ofMillis(50)); diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/TopicsSpec.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/TopicsSpec.java index 600fd46f21fcc..e597902865a7a 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/TopicsSpec.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/TopicsSpec.java @@ -24,7 +24,6 @@ import com.fasterxml.jackson.annotation.JsonAnySetter; import com.fasterxml.jackson.annotation.JsonCreator; -import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -68,9 +67,8 @@ public void set(String name, PartitionsSpec value) { } public TopicsSpec immutableCopy() { - HashMap mapCopy = new HashMap<>(); - mapCopy.putAll(map); - return new TopicsSpec(Collections.unmodifiableMap(mapCopy)); + HashMap mapCopy = new HashMap<>(map); + return new TopicsSpec(Map.copyOf(mapCopy)); } /** diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/agent/AgentTest.java b/trogdor/src/test/java/org/apache/kafka/trogdor/agent/AgentTest.java index 070de473056e7..e21059823f352 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/agent/AgentTest.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/agent/AgentTest.java @@ -61,11 +61,12 @@ import java.nio.charset.StandardCharsets; import java.nio.file.Path; import java.nio.file.Paths; -import java.util.Collections; import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; import java.util.TreeMap; -import static java.util.Arrays.asList; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -80,7 +81,7 @@ private static BasicPlatform createBasicPlatform(Scheduler scheduler) { HashMap config = new HashMap<>(); config.put(Platform.Config.TROGDOR_AGENT_PORT, Integer.toString(Agent.DEFAULT_PORT)); nodes.put("node01", new BasicNode("node01", "localhost", - config, Collections.emptySet())); + config, Set.of())); BasicTopology topology = new BasicTopology(nodes); return new BasicPlatform("node01", topology, scheduler, new BasicPlatform.ShellCommandRunner()); @@ -153,7 +154,7 @@ public void sleep(long ms) { maxTries(10).target("localhost", agent.port()).build(); AgentStatusResponse status = client.status(); - assertEquals(Collections.emptyMap(), status.workers()); + assertEquals(Map.of(), status.workers()); new ExpectedTasks().waitFor(client); final NoOpTaskSpec fooSpec = new NoOpTaskSpec(10, 10); @@ -191,7 +192,7 @@ public void testAgentCreateWorkers() throws Exception { AgentClient client = new AgentClient.Builder(). maxTries(10).target("localhost", agent.port()).build(); AgentStatusResponse status = client.status(); - assertEquals(Collections.emptyMap(), status.workers()); + assertEquals(Map.of(), status.workers()); new ExpectedTasks().waitFor(client); final NoOpTaskSpec fooSpec = new NoOpTaskSpec(1000, 600000); @@ -304,7 +305,7 @@ public void testWorkerCompletions() throws Exception { new ExpectedTasks().waitFor(client); SampleTaskSpec fooSpec = new SampleTaskSpec(0, 900000, - Collections.singletonMap("node01", 1L), ""); + Map.of("node01", 1L), ""); client.createWorker(new CreateWorkerRequest(0, "foo", fooSpec)); new ExpectedTasks(). addTask(new ExpectedTaskBuilder("foo"). @@ -313,7 +314,7 @@ public void testWorkerCompletions() throws Exception { waitFor(client); SampleTaskSpec barSpec = new SampleTaskSpec(0, 900000, - Collections.singletonMap("node01", 2L), "baz"); + Map.of("node01", 2L), "baz"); client.createWorker(new CreateWorkerRequest(1, "bar", barSpec)); time.sleep(1); @@ -373,17 +374,17 @@ public void testKiboshFaults() throws Exception { try (MockKibosh mockKibosh = new MockKibosh()) { assertEquals(KiboshControlFile.EMPTY, mockKibosh.read()); FilesUnreadableFaultSpec fooSpec = new FilesUnreadableFaultSpec(0, 900000, - Collections.singleton("myAgent"), mockKibosh.tempDir.getPath(), "/foo", 123); + Set.of("myAgent"), mockKibosh.tempDir.getPath(), "/foo", 123); client.createWorker(new CreateWorkerRequest(0, "foo", fooSpec)); new ExpectedTasks(). addTask(new ExpectedTaskBuilder("foo"). workerState(new WorkerRunning("foo", fooSpec, 0, new TextNode("Added fault foo"))). build()). waitFor(client); - assertEquals(new KiboshControlFile(Collections.singletonList( + assertEquals(new KiboshControlFile(List.of( new KiboshFilesUnreadableFaultSpec("/foo", 123))), mockKibosh.read()); FilesUnreadableFaultSpec barSpec = new FilesUnreadableFaultSpec(0, 900000, - Collections.singleton("myAgent"), mockKibosh.tempDir.getPath(), "/bar", 456); + Set.of("myAgent"), mockKibosh.tempDir.getPath(), "/bar", 456); client.createWorker(new CreateWorkerRequest(1, "bar", barSpec)); new ExpectedTasks(). addTask(new ExpectedTaskBuilder("foo"). @@ -391,7 +392,7 @@ public void testKiboshFaults() throws Exception { addTask(new ExpectedTaskBuilder("bar"). workerState(new WorkerRunning("bar", barSpec, 0, new TextNode("Added fault bar"))).build()). waitFor(client); - assertEquals(new KiboshControlFile(asList( + assertEquals(new KiboshControlFile(List.of( new KiboshFilesUnreadableFaultSpec("/foo", 123), new KiboshFilesUnreadableFaultSpec("/bar", 456)) ), mockKibosh.read()); @@ -403,7 +404,7 @@ public void testKiboshFaults() throws Exception { addTask(new ExpectedTaskBuilder("bar"). workerState(new WorkerRunning("bar", barSpec, 0, new TextNode("Added fault bar"))).build()). waitFor(client); - assertEquals(new KiboshControlFile(Collections.singletonList( + assertEquals(new KiboshControlFile(List.of( new KiboshFilesUnreadableFaultSpec("/bar", 456))), mockKibosh.read()); } } @@ -476,7 +477,7 @@ public void testAgentExecWithTimeout() throws Exception { public void testAgentExecWithNormalExit() throws Exception { Agent agent = createAgent(Scheduler.SYSTEM); SampleTaskSpec spec = new SampleTaskSpec(0, 120000, - Collections.singletonMap("node01", 1L), ""); + Map.of("node01", 1L), ""); TaskSpec rebasedSpec = agent.rebaseTaskSpecTime(spec); testExec(agent, String.format("Waiting for completion of task:%s%n", diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/common/JsonSerializationTest.java b/trogdor/src/test/java/org/apache/kafka/trogdor/common/JsonSerializationTest.java index 93d5d9f39847a..4eefe2e0263d3 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/common/JsonSerializationTest.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/common/JsonSerializationTest.java @@ -35,7 +35,6 @@ import java.lang.reflect.Field; import java.lang.reflect.Modifier; -import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -63,8 +62,8 @@ public void testDeserializationDoesNotProduceNulls() throws Exception { verify(new TopicsSpec()); verify(new PartitionsSpec(0, (short) 0, null, null)); Map> partitionAssignments = new HashMap<>(); - partitionAssignments.put(0, Arrays.asList(1, 2, 3)); - partitionAssignments.put(1, Arrays.asList(1, 2, 3)); + partitionAssignments.put(0, List.of(1, 2, 3)); + partitionAssignments.put(1, List.of(1, 2, 3)); verify(new PartitionsSpec(0, (short) 0, partitionAssignments, null)); verify(new PartitionsSpec(0, (short) 0, null, null)); } diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/common/JsonUtilTest.java b/trogdor/src/test/java/org/apache/kafka/trogdor/common/JsonUtilTest.java index 6b7b67a894bec..8f42d043311c4 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/common/JsonUtilTest.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/common/JsonUtilTest.java @@ -45,10 +45,7 @@ public void testOpenBraceComesFirst() { assertFalse(JsonUtil.openBraceComesFirst(" blah{}")); } - static final class Foo { - @JsonProperty - final int bar; - + record Foo(@JsonProperty int bar) { @JsonCreator Foo(@JsonProperty("bar") int bar) { this.bar = bar; diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/common/MiniTrogdorCluster.java b/trogdor/src/test/java/org/apache/kafka/trogdor/common/MiniTrogdorCluster.java index 64e8c5def1ce3..946dcc7a41337 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/common/MiniTrogdorCluster.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/common/MiniTrogdorCluster.java @@ -33,9 +33,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.Set; import java.util.TreeMap; import java.util.TreeSet; import java.util.concurrent.Callable; @@ -161,7 +161,7 @@ public MiniTrogdorCluster build() throws Exception { Integer.toString(node.coordinatorPort)); } node.node = new BasicNode(entry.getKey(), node.hostname, config, - Collections.emptySet()); + Set.of()); } TreeMap topologyNodes = new TreeMap<>(); for (Map.Entry entry : nodes.entrySet()) { diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/common/StringExpanderTest.java b/trogdor/src/test/java/org/apache/kafka/trogdor/common/StringExpanderTest.java index 4ece85846ad2d..4c52a67096018 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/common/StringExpanderTest.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/common/StringExpanderTest.java @@ -20,7 +20,6 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; -import java.util.Collections; import java.util.Set; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -30,9 +29,9 @@ public class StringExpanderTest { @Test public void testNoExpansionNeeded() { - assertEquals(Collections.singleton("foo"), StringExpander.expand("foo")); - assertEquals(Collections.singleton("bar"), StringExpander.expand("bar")); - assertEquals(Collections.singleton(""), StringExpander.expand("")); + assertEquals(Set.of("foo"), StringExpander.expand("foo")); + assertEquals(Set.of("bar"), StringExpander.expand("bar")); + assertEquals(Set.of(""), StringExpander.expand("")); } @Test diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/common/StringFormatterTest.java b/trogdor/src/test/java/org/apache/kafka/trogdor/common/StringFormatterTest.java index 2c78f61aac0a9..b2a800d032f99 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/common/StringFormatterTest.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/common/StringFormatterTest.java @@ -21,7 +21,7 @@ import org.junit.jupiter.api.Timeout; import java.time.ZoneOffset; -import java.util.Arrays; +import java.util.List; import static org.apache.kafka.trogdor.common.StringFormatter.dateString; import static org.apache.kafka.trogdor.common.StringFormatter.durationString; @@ -53,8 +53,8 @@ public void testPrettyPrintGrid() { "lion 1 12345 %n" + "manatee 50 1 %n"), StringFormatter.prettyPrintGrid( - Arrays.asList(Arrays.asList("ANIMAL", "NUMBER", "INDEX"), - Arrays.asList("lion", "1", "12345"), - Arrays.asList("manatee", "50", "1")))); + List.of(List.of("ANIMAL", "NUMBER", "INDEX"), + List.of("lion", "1", "12345"), + List.of("manatee", "50", "1")))); } } diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/common/WorkerUtilsTest.java b/trogdor/src/test/java/org/apache/kafka/trogdor/common/WorkerUtilsTest.java index 3ada5d75470c4..58affc48ca5cf 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/common/WorkerUtilsTest.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/common/WorkerUtilsTest.java @@ -33,9 +33,7 @@ import org.slf4j.LoggerFactory; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collection; -import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -53,8 +51,8 @@ public class WorkerUtilsTest { private final Node broker1 = new Node(0, "testHost-1", 1234); private final Node broker2 = new Node(1, "testHost-2", 1234); private final Node broker3 = new Node(1, "testHost-3", 1234); - private final List cluster = Arrays.asList(broker1, broker2, broker3); - private final List singleReplica = Collections.singletonList(broker1); + private final List cluster = List.of(broker1, broker2, broker3); + private final List singleReplica = List.of(broker1); private static final String TEST_TOPIC = "test-topic-1"; private static final short TEST_REPLICATION_FACTOR = 1; @@ -72,17 +70,17 @@ public void setUp() { @Test public void testCreateOneTopic() throws Throwable { - Map newTopics = Collections.singletonMap(TEST_TOPIC, NEW_TEST_TOPIC); + Map newTopics = Map.of(TEST_TOPIC, NEW_TEST_TOPIC); WorkerUtils.createTopics(log, adminClient, newTopics, true); - assertEquals(Collections.singleton(TEST_TOPIC), adminClient.listTopics().names().get()); + assertEquals(Set.of(TEST_TOPIC), adminClient.listTopics().names().get()); assertEquals( new TopicDescription( TEST_TOPIC, false, - Collections.singletonList( - new TopicPartitionInfo(0, broker1, singleReplica, Collections.emptyList(), Collections.emptyList(), Collections.emptyList()))), + List.of( + new TopicPartitionInfo(0, broker1, singleReplica, List.of(), List.of(), List.of()))), adminClient.describeTopics( - Collections.singleton(TEST_TOPIC)).topicNameValues().get(TEST_TOPIC).get() + Set.of(TEST_TOPIC)).topicNameValues().get(TEST_TOPIC).get() ); } @@ -91,21 +89,21 @@ public void testCreateRetriesOnTimeout() throws Throwable { adminClient.timeoutNextRequest(1); WorkerUtils.createTopics( - log, adminClient, Collections.singletonMap(TEST_TOPIC, NEW_TEST_TOPIC), true); + log, adminClient, Map.of(TEST_TOPIC, NEW_TEST_TOPIC), true); assertEquals( new TopicDescription( TEST_TOPIC, false, - Collections.singletonList( - new TopicPartitionInfo(0, broker1, singleReplica, Collections.emptyList(), Collections.emptyList(), Collections.emptyList()))), + List.of( + new TopicPartitionInfo(0, broker1, singleReplica, List.of(), List.of(), List.of()))), adminClient.describeTopics( - Collections.singleton(TEST_TOPIC)).topicNameValues().get(TEST_TOPIC).get() + Set.of(TEST_TOPIC)).topicNameValues().get(TEST_TOPIC).get() ); } @Test public void testCreateZeroTopicsDoesNothing() throws Throwable { - WorkerUtils.createTopics(log, adminClient, Collections.emptyMap(), true); + WorkerUtils.createTopics(log, adminClient, Map.of(), true); assertEquals(0, adminClient.listTopics().names().get().size()); } @@ -114,7 +112,7 @@ public void testCreateTopicsFailsIfAtLeastOneTopicExists() { adminClient.addTopic( false, TEST_TOPIC, - Collections.singletonList(new TopicPartitionInfo(0, broker1, singleReplica, Collections.emptyList())), + List.of(new TopicPartitionInfo(0, broker1, singleReplica, List.of())), null); Map newTopics = new HashMap<>(); @@ -130,8 +128,8 @@ public void testCreateTopicsFailsIfAtLeastOneTopicExists() { @Test public void testExistingTopicsMustHaveRequestedNumberOfPartitions() { List tpInfo = new ArrayList<>(); - tpInfo.add(new TopicPartitionInfo(0, broker1, singleReplica, Collections.emptyList())); - tpInfo.add(new TopicPartitionInfo(1, broker2, singleReplica, Collections.emptyList())); + tpInfo.add(new TopicPartitionInfo(0, broker1, singleReplica, List.of())); + tpInfo.add(new TopicPartitionInfo(1, broker2, singleReplica, List.of())); adminClient.addTopic( false, TEST_TOPIC, @@ -139,16 +137,16 @@ public void testExistingTopicsMustHaveRequestedNumberOfPartitions() { null); assertThrows(RuntimeException.class, () -> WorkerUtils.createTopics( - log, adminClient, Collections.singletonMap(TEST_TOPIC, NEW_TEST_TOPIC), false)); + log, adminClient, Map.of(TEST_TOPIC, NEW_TEST_TOPIC), false)); } @Test public void testExistingTopicsNotCreated() throws Throwable { final String existingTopic = "existing-topic"; List tpInfo = new ArrayList<>(); - tpInfo.add(new TopicPartitionInfo(0, broker1, singleReplica, Collections.emptyList())); - tpInfo.add(new TopicPartitionInfo(1, broker2, singleReplica, Collections.emptyList())); - tpInfo.add(new TopicPartitionInfo(2, broker3, singleReplica, Collections.emptyList())); + tpInfo.add(new TopicPartitionInfo(0, broker1, singleReplica, List.of())); + tpInfo.add(new TopicPartitionInfo(1, broker2, singleReplica, List.of())); + tpInfo.add(new TopicPartitionInfo(2, broker3, singleReplica, List.of())); adminClient.addTopic( false, existingTopic, @@ -157,11 +155,11 @@ public void testExistingTopicsNotCreated() throws Throwable { WorkerUtils.createTopics( log, adminClient, - Collections.singletonMap( + Map.of( existingTopic, new NewTopic(existingTopic, tpInfo.size(), TEST_REPLICATION_FACTOR)), false); - assertEquals(Collections.singleton(existingTopic), adminClient.listTopics().names().get()); + assertEquals(Set.of(existingTopic), adminClient.listTopics().names().get()); } @Test @@ -170,15 +168,15 @@ public void testCreatesNotExistingTopics() throws Throwable { assertEquals(0, adminClient.listTopics().names().get().size()); WorkerUtils.createTopics( - log, adminClient, Collections.singletonMap(TEST_TOPIC, NEW_TEST_TOPIC), false); + log, adminClient, Map.of(TEST_TOPIC, NEW_TEST_TOPIC), false); - assertEquals(Collections.singleton(TEST_TOPIC), adminClient.listTopics().names().get()); + assertEquals(Set.of(TEST_TOPIC), adminClient.listTopics().names().get()); assertEquals( new TopicDescription( TEST_TOPIC, false, - Collections.singletonList( - new TopicPartitionInfo(0, broker1, singleReplica, Collections.emptyList(), Collections.emptyList(), Collections.emptyList()))), - adminClient.describeTopics(Collections.singleton(TEST_TOPIC)).topicNameValues().get(TEST_TOPIC).get() + List.of( + new TopicPartitionInfo(0, broker1, singleReplica, List.of(), List.of(), List.of()))), + adminClient.describeTopics(Set.of(TEST_TOPIC)).topicNameValues().get(TEST_TOPIC).get() ); } @@ -186,8 +184,8 @@ public void testCreatesNotExistingTopics() throws Throwable { public void testCreatesOneTopicVerifiesOneTopic() throws Throwable { final String existingTopic = "existing-topic"; List tpInfo = new ArrayList<>(); - tpInfo.add(new TopicPartitionInfo(0, broker1, singleReplica, Collections.emptyList())); - tpInfo.add(new TopicPartitionInfo(1, broker2, singleReplica, Collections.emptyList())); + tpInfo.add(new TopicPartitionInfo(0, broker1, singleReplica, List.of())); + tpInfo.add(new TopicPartitionInfo(1, broker2, singleReplica, List.of())); adminClient.addTopic( false, existingTopic, @@ -207,7 +205,7 @@ public void testCreatesOneTopicVerifiesOneTopic() throws Throwable { @Test public void testCreateNonExistingTopicsWithZeroTopicsDoesNothing() throws Throwable { WorkerUtils.createTopics( - log, adminClient, Collections.emptyMap(), false); + log, adminClient, Map.of(), false); assertEquals(0, adminClient.listTopics().names().get().size()); } @@ -224,8 +222,8 @@ public void testAddConfigsToPropertiesAddsAllConfigs() { WorkerUtils.addConfigsToProperties( props, - Collections.singletonMap(ProducerConfig.CLIENT_ID_CONFIG, "test-client"), - Collections.singletonMap(ProducerConfig.LINGER_MS_CONFIG, "1000")); + Map.of(ProducerConfig.CLIENT_ID_CONFIG, "test-client"), + Map.of(ProducerConfig.LINGER_MS_CONFIG, "1000")); assertEquals(resultProps, props); } @@ -242,8 +240,8 @@ public void testCommonConfigOverwritesDefaultProps() { WorkerUtils.addConfigsToProperties( props, - Collections.singletonMap(ProducerConfig.ACKS_CONFIG, "1"), - Collections.singletonMap(ProducerConfig.LINGER_MS_CONFIG, "1000")); + Map.of(ProducerConfig.ACKS_CONFIG, "1"), + Map.of(ProducerConfig.LINGER_MS_CONFIG, "1000")); assertEquals(resultProps, props); } @@ -259,8 +257,8 @@ public void testClientConfigOverwritesBothDefaultAndCommonConfigs() { WorkerUtils.addConfigsToProperties( props, - Collections.singletonMap(ProducerConfig.ACKS_CONFIG, "1"), - Collections.singletonMap(ProducerConfig.ACKS_CONFIG, "0")); + Map.of(ProducerConfig.ACKS_CONFIG, "1"), + Map.of(ProducerConfig.ACKS_CONFIG, "0")); assertEquals(resultProps, props); } @@ -308,7 +306,7 @@ private void makeExistingTopicWithOneReplica(String topicName, int numPartitions for (int i = 0; i < numPartitions; ++i) { Node broker = cluster.get(brokerIndex); tpInfo.add(new TopicPartitionInfo( - i, broker, singleReplica, Collections.emptyList())); + i, broker, singleReplica, List.of())); brokerIndex = (brokerIndex + 1) % cluster.size(); } adminClient.addTopic( @@ -320,14 +318,14 @@ private void makeExistingTopicWithOneReplica(String topicName, int numPartitions @Test public void testVerifyTopics() throws Throwable { - Map newTopics = Collections.singletonMap(TEST_TOPIC, NEW_TEST_TOPIC); + Map newTopics = Map.of(TEST_TOPIC, NEW_TEST_TOPIC); WorkerUtils.createTopics(log, adminClient, newTopics, true); adminClient.setFetchesRemainingUntilVisible(TEST_TOPIC, 2); - WorkerUtils.verifyTopics(log, adminClient, Collections.singleton(TEST_TOPIC), - Collections.singletonMap(TEST_TOPIC, NEW_TEST_TOPIC), 3, 1); + WorkerUtils.verifyTopics(log, adminClient, Set.of(TEST_TOPIC), + Map.of(TEST_TOPIC, NEW_TEST_TOPIC), 3, 1); adminClient.setFetchesRemainingUntilVisible(TEST_TOPIC, 100); assertThrows(UnknownTopicOrPartitionException.class, () -> - WorkerUtils.verifyTopics(log, adminClient, Collections.singleton(TEST_TOPIC), - Collections.singletonMap(TEST_TOPIC, NEW_TEST_TOPIC), 2, 1)); + WorkerUtils.verifyTopics(log, adminClient, Set.of(TEST_TOPIC), + Map.of(TEST_TOPIC, NEW_TEST_TOPIC), 2, 1)); } } diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/coordinator/CoordinatorTest.java b/trogdor/src/test/java/org/apache/kafka/trogdor/coordinator/CoordinatorTest.java index 313a5db743741..ee5f143cd368a 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/coordinator/CoordinatorTest.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/coordinator/CoordinatorTest.java @@ -57,7 +57,6 @@ import org.slf4j.LoggerFactory; import java.util.ArrayList; -import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Optional; @@ -370,7 +369,7 @@ public String toString() { private static List> createPartitionLists(String[][] array) { List> list = new ArrayList<>(); for (String[] a : array) { - list.add(Arrays.asList(a)); + list.add(List.of(a)); } return list; } @@ -486,7 +485,7 @@ public void testTasksRequest() throws Exception { assertEquals(0, coordinatorClient.tasks( new TasksRequest(null, 10, 0, 10, 0, Optional.empty())).tasks().size()); TasksResponse resp1 = coordinatorClient.tasks( - new TasksRequest(Arrays.asList("foo", "baz"), 0, 0, 0, 0, Optional.empty())); + new TasksRequest(List.of("foo", "baz"), 0, 0, 0, 0, Optional.empty())); assertTrue(resp1.tasks().containsKey("foo")); assertFalse(resp1.tasks().containsKey("bar")); assertEquals(1, resp1.tasks().size()); diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/task/SampleTaskSpec.java b/trogdor/src/test/java/org/apache/kafka/trogdor/task/SampleTaskSpec.java index 595f5f13d4213..7bd610f896eb8 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/task/SampleTaskSpec.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/task/SampleTaskSpec.java @@ -20,7 +20,6 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; -import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -35,7 +34,7 @@ public SampleTaskSpec(@JsonProperty("startMs") long startMs, @JsonProperty("error") String error) { super(startMs, durationMs); this.nodeToExitMs = nodeToExitMs == null ? new HashMap<>() : - Collections.unmodifiableMap(nodeToExitMs); + Map.copyOf(nodeToExitMs); this.error = error == null ? "" : error; } diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ConsumeBenchSpecTest.java b/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ConsumeBenchSpecTest.java index e07991d3bf021..7ef1c43e87f80 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ConsumeBenchSpecTest.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ConsumeBenchSpecTest.java @@ -22,13 +22,10 @@ import org.junit.jupiter.api.Test; import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Optional; -import java.util.stream.Collectors; import java.util.stream.IntStream; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -38,7 +35,7 @@ public class ConsumeBenchSpecTest { @Test public void testMaterializeTopicsWithNoPartitions() { - Map> materializedTopics = consumeBenchSpec(Arrays.asList("topic[1-3]", "secondTopic")).materializeTopics(); + Map> materializedTopics = consumeBenchSpec(List.of("topic[1-3]", "secondTopic")).materializeTopics(); Map> expected = new HashMap<>(); expected.put("topic1", new ArrayList<>()); expected.put("topic2", new ArrayList<>()); @@ -50,28 +47,28 @@ public void testMaterializeTopicsWithNoPartitions() { @Test public void testMaterializeTopicsWithSomePartitions() { - Map> materializedTopics = consumeBenchSpec(Arrays.asList("topic[1-3]:[1-5]", "secondTopic", "thirdTopic:1")).materializeTopics(); + Map> materializedTopics = consumeBenchSpec(List.of("topic[1-3]:[1-5]", "secondTopic", "thirdTopic:1")).materializeTopics(); Map> expected = new HashMap<>(); - expected.put("topic1", IntStream.range(1, 6).asLongStream().mapToObj(i -> new TopicPartition("topic1", (int) i)).collect(Collectors.toList())); - expected.put("topic2", IntStream.range(1, 6).asLongStream().mapToObj(i -> new TopicPartition("topic2", (int) i)).collect(Collectors.toList())); - expected.put("topic3", IntStream.range(1, 6).asLongStream().mapToObj(i -> new TopicPartition("topic3", (int) i)).collect(Collectors.toList())); + expected.put("topic1", IntStream.range(1, 6).asLongStream().mapToObj(i -> new TopicPartition("topic1", (int) i)).toList()); + expected.put("topic2", IntStream.range(1, 6).asLongStream().mapToObj(i -> new TopicPartition("topic2", (int) i)).toList()); + expected.put("topic3", IntStream.range(1, 6).asLongStream().mapToObj(i -> new TopicPartition("topic3", (int) i)).toList()); expected.put("secondTopic", new ArrayList<>()); - expected.put("thirdTopic", Collections.singletonList(new TopicPartition("thirdTopic", 1))); + expected.put("thirdTopic", List.of(new TopicPartition("thirdTopic", 1))); assertEquals(expected, materializedTopics); } @Test public void testInvalidTopicNameRaisesExceptionInMaterialize() { - for (String invalidName : Arrays.asList("In:valid", "invalid:", ":invalid", "in:valid:1", "invalid:2:2", "invalid::1", "invalid[1-3]:")) { - assertThrows(IllegalArgumentException.class, () -> consumeBenchSpec(Collections.singletonList(invalidName)).materializeTopics()); + for (String invalidName : List.of("In:valid", "invalid:", ":invalid", "in:valid:1", "invalid:2:2", "invalid::1", "invalid[1-3]:")) { + assertThrows(IllegalArgumentException.class, () -> consumeBenchSpec(List.of(invalidName)).materializeTopics()); } } private ConsumeBenchSpec consumeBenchSpec(List activeTopics) { return new ConsumeBenchSpec(0, 0, "node", "localhost", 123, 1234, "cg-1", - Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap(), 1, + Map.of(), Map.of(), Map.of(), 1, Optional.empty(), activeTopics); } } diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ExternalCommandWorkerTest.java b/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ExternalCommandWorkerTest.java index 852dd3c3b473b..0960f8f895515 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ExternalCommandWorkerTest.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ExternalCommandWorkerTest.java @@ -35,7 +35,7 @@ import java.io.OutputStream; import java.nio.charset.StandardCharsets; import java.nio.file.Files; -import java.util.Arrays; +import java.util.List; import java.util.Optional; import java.util.concurrent.CompletableFuture; @@ -62,7 +62,7 @@ ExternalCommandWorker build() { ExternalCommandSpec spec = new ExternalCommandSpec(0, 30000, "node0", - Arrays.asList(command), + List.of(command), workload, Optional.of(shutdownGracePeriodMs)); return new ExternalCommandWorker(id, spec); diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/workload/PayloadGeneratorTest.java b/trogdor/src/test/java/org/apache/kafka/trogdor/workload/PayloadGeneratorTest.java index 999cc51ef7656..21d1e418070f0 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/workload/PayloadGeneratorTest.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/workload/PayloadGeneratorTest.java @@ -22,7 +22,6 @@ import java.nio.ByteBuffer; import java.nio.ByteOrder; -import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -146,8 +145,8 @@ public void testRandomComponentPayloadGenerator() { new ConstantPayloadGenerator(4, new byte[0]); RandomComponent constantConfig = new RandomComponent(25, constantGenerator); - List components1 = new ArrayList<>(Arrays.asList(nullConfig, uniformConfig)); - List components2 = new ArrayList<>(Arrays.asList(sequentialConfig, constantConfig)); + List components1 = List.of(nullConfig, uniformConfig); + List components2 = List.of(sequentialConfig, constantConfig); byte[] expected = new byte[4]; PayloadIterator iter = new PayloadIterator( @@ -180,8 +179,8 @@ public void testRandomComponentPayloadGenerator() { RandomComponent sequentialConfig2 = new RandomComponent(25, sequentialGenerator); RandomComponent nullConfig2 = new RandomComponent(25, nullGenerator); - List components3 = new ArrayList<>(Arrays.asList(sequentialConfig2, uniformConfig2, nullConfig)); - List components4 = new ArrayList<>(Arrays.asList(uniformConfig2, sequentialConfig2, constantConfig, nullConfig2)); + List components3 = List.of(sequentialConfig2, uniformConfig2, nullConfig); + List components4 = List.of(uniformConfig2, sequentialConfig2, constantConfig, nullConfig2); testReproducible(new RandomComponentPayloadGenerator(4, components1)); testReproducible(new RandomComponentPayloadGenerator(123, components2)); @@ -200,12 +199,12 @@ public void testRandomComponentPayloadGeneratorErrors() { new ConstantPayloadGenerator(4, new byte[0]); RandomComponent constantConfig = new RandomComponent(-25, constantGenerator); - List components1 = new ArrayList<>(Arrays.asList(nullConfig, uniformConfig)); - List components2 = new ArrayList<>(Arrays.asList( - nullConfig, constantConfig, uniformConfig, nullConfig, uniformConfig, uniformConfig)); + List components1 = List.of(nullConfig, uniformConfig); + List components2 = List.of( + nullConfig, constantConfig, uniformConfig, nullConfig, uniformConfig, uniformConfig); assertThrows(IllegalArgumentException.class, () -> - new PayloadIterator(new RandomComponentPayloadGenerator(1, new ArrayList<>()))); + new PayloadIterator(new RandomComponentPayloadGenerator(1, List.of()))); assertThrows(IllegalArgumentException.class, () -> new PayloadIterator(new RandomComponentPayloadGenerator(13, components2))); assertThrows(IllegalArgumentException.class, () -> diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpecTest.java b/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpecTest.java index 0dbd57651eb76..a4680557248ec 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpecTest.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpecTest.java @@ -18,10 +18,9 @@ import org.junit.jupiter.api.Test; -import java.util.Arrays; -import java.util.Collections; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.Set; @@ -32,7 +31,7 @@ class ShareConsumeBenchSpecTest { @Test public void testExpandTopicNames() { - ShareConsumeBenchSpec shareConsumeBenchSpec = shareConsumeBenchSpec(Arrays.asList("foo[1-3]", "bar")); + ShareConsumeBenchSpec shareConsumeBenchSpec = shareConsumeBenchSpec(List.of("foo[1-3]", "bar")); Set expectedNames = new HashSet<>(); expectedNames.add("foo1"); @@ -45,15 +44,15 @@ public void testExpandTopicNames() { @Test public void testInvalidNameRaisesException() { - for (String invalidName : Arrays.asList("In:valid", "invalid:", ":invalid[]", "in:valid:", "invalid[1-3]:")) { - assertThrows(IllegalArgumentException.class, () -> shareConsumeBenchSpec(Collections.singletonList(invalidName)).expandTopicNames()); + for (String invalidName : List.of("In:valid", "invalid:", ":invalid[]", "in:valid:", "invalid[1-3]:")) { + assertThrows(IllegalArgumentException.class, () -> shareConsumeBenchSpec(List.of(invalidName)).expandTopicNames()); } } @Test public void testDefaultShareGroupName() { ShareConsumeBenchSpec shareConsumeBenchSpec = new ShareConsumeBenchSpec(0, 0, "node", "localhost", - 123, 1234, null, Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap(), 1, + 123, 1234, null, Map.of(), Map.of(), Map.of(), 1, Optional.empty(), List.of("abc")); assertEquals("share", shareConsumeBenchSpec.shareGroup()); } @@ -61,7 +60,7 @@ public void testDefaultShareGroupName() { private ShareConsumeBenchSpec shareConsumeBenchSpec(List activeTopics) { return new ShareConsumeBenchSpec(0, 0, "node", "localhost", 123, 1234, "sg-1", - Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap(), 1, + Map.of(), Map.of(), Map.of(), 1, Optional.empty(), activeTopics); } diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/workload/TopicsSpecTest.java b/trogdor/src/test/java/org/apache/kafka/trogdor/workload/TopicsSpecTest.java index d46585b869c33..4bd5018c98ea9 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/workload/TopicsSpecTest.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/workload/TopicsSpecTest.java @@ -22,7 +22,6 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; -import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -44,8 +43,8 @@ public class TopicsSpecTest { FOO.set("topicA[0-2]", PARTSA); Map> assignmentsB = new HashMap<>(); - assignmentsB.put(0, Arrays.asList(0, 1, 2)); - assignmentsB.put(1, Arrays.asList(2, 3, 4)); + assignmentsB.put(0, List.of(0, 1, 2)); + assignmentsB.put(1, List.of(2, 3, 4)); PARTSB = new PartitionsSpec(0, (short) 0, assignmentsB, null); FOO.set("topicB", PARTSB); }