You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2025/08/19 14:24:21 UTC
(kafka) branch trunk updated: MINOR: Cleanups in Tools Module (3/n) (#20332)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 0da9cacffab MINOR: Cleanups in Tools Module (3/n) (#20332)
0da9cacffab is described below
commit 0da9cacffaba276e1bc6c1301aed449aaff788b9
Author: Sanskar Jhajharia <sj...@confluent.io>
AuthorDate: Tue Aug 19 19:54:13 2025 +0530
MINOR: Cleanups in Tools Module (3/n) (#20332)
This PR aims at cleaning up the tools module further by getting rid of
some extra code which can be replaced by `record`
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../java/org/apache/kafka/tools/AclCommand.java | 168 ++++++++++-----------
.../org/apache/kafka/tools/ConnectPluginPath.java | 35 +----
.../apache/kafka/tools/MetadataQuorumCommand.java | 3 +-
.../apache/kafka/tools/OAuthCompatibilityTool.java | 9 +-
.../java/org/apache/kafka/tools/OffsetsUtils.java | 11 +-
.../kafka/tools/PushHttpMetricsReporter.java | 85 ++---------
.../kafka/tools/ReplicaVerificationTool.java | 13 +-
.../apache/kafka/tools/TransactionsCommand.java | 31 ++--
.../tools/consumer/group/ConsumerGroupCommand.java | 96 ++++++------
.../tools/consumer/group/GroupInformation.java | 28 +---
.../consumer/group/MemberAssignmentState.java | 42 +-----
.../consumer/group/PartitionAssignmentState.java | 42 +-----
.../tools/consumer/group/ShareGroupCommand.java | 22 +--
.../kafka/tools/reassign/ActiveMoveState.java | 37 +----
.../kafka/tools/reassign/CancelledMoveState.java | 32 +---
.../kafka/tools/reassign/CompletedMoveState.java | 27 +---
.../tools/reassign/MissingLogDirMoveState.java | 27 +---
.../tools/reassign/MissingReplicaMoveState.java | 27 +---
.../tools/reassign/ReassignPartitionsCommand.java | 10 +-
.../apache/kafka/tools/ConnectPluginPathTest.java | 27 +---
.../consumer/group/ConsumerGroupServiceTest.java | 8 +-
.../consumer/group/DeleteConsumerGroupsTest.java | 2 +-
.../consumer/group/DescribeConsumerGroupTest.java | 92 +++++------
.../group/ResetConsumerGroupOffsetTest.java | 4 +-
.../reassign/ReassignPartitionsCommandTest.java | 11 +-
25 files changed, 242 insertions(+), 647 deletions(-)
diff --git a/tools/src/main/java/org/apache/kafka/tools/AclCommand.java b/tools/src/main/java/org/apache/kafka/tools/AclCommand.java
index 791397b2405..fd62e8f3b56 100644
--- a/tools/src/main/java/org/apache/kafka/tools/AclCommand.java
+++ b/tools/src/main/java/org/apache/kafka/tools/AclCommand.java
@@ -73,14 +73,13 @@ public class AclCommand {
public static void main(String[] args) {
AclCommandOptions opts = new AclCommandOptions(args);
- AdminClientService aclCommandService = new AdminClientService(opts);
try (Admin admin = Admin.create(adminConfigs(opts))) {
if (opts.options.has(opts.addOpt)) {
- aclCommandService.addAcls(admin);
+ addAcls(admin, opts);
} else if (opts.options.has(opts.removeOpt)) {
- aclCommandService.removeAcls(admin);
+ removeAcls(admin, opts);
} else if (opts.options.has(opts.listOpt)) {
- aclCommandService.listAcls(admin);
+ listAcls(admin, opts);
}
} catch (Throwable e) {
System.out.println("Error while executing ACL command: " + e.getMessage());
@@ -102,106 +101,97 @@ public class AclCommand {
return props;
}
- private static class AdminClientService {
-
- private final AclCommandOptions opts;
-
- AdminClientService(AclCommandOptions opts) {
- this.opts = opts;
- }
-
- void addAcls(Admin admin) throws ExecutionException, InterruptedException {
- Map<ResourcePattern, Set<AccessControlEntry>> resourceToAcl = getResourceToAcls(opts);
- for (Map.Entry<ResourcePattern, Set<AccessControlEntry>> entry : resourceToAcl.entrySet()) {
- ResourcePattern resource = entry.getKey();
- Set<AccessControlEntry> acls = entry.getValue();
- System.out.println("Adding ACLs for resource `" + resource + "`: " + NL + " " + acls.stream().map(a -> "\t" + a).collect(Collectors.joining(NL)) + NL);
- Collection<AclBinding> aclBindings = acls.stream().map(acl -> new AclBinding(resource, acl)).collect(Collectors.toList());
- admin.createAcls(aclBindings).all().get();
- }
+ private static void addAcls(Admin admin, AclCommandOptions opts) throws ExecutionException, InterruptedException {
+ Map<ResourcePattern, Set<AccessControlEntry>> resourceToAcl = getResourceToAcls(opts);
+ for (Map.Entry<ResourcePattern, Set<AccessControlEntry>> entry : resourceToAcl.entrySet()) {
+ ResourcePattern resource = entry.getKey();
+ Set<AccessControlEntry> acls = entry.getValue();
+ System.out.println("Adding ACLs for resource `" + resource + "`: " + NL + " " + acls.stream().map(a -> "\t" + a).collect(Collectors.joining(NL)) + NL);
+ Collection<AclBinding> aclBindings = acls.stream().map(acl -> new AclBinding(resource, acl)).collect(Collectors.toList());
+ admin.createAcls(aclBindings).all().get();
}
+ }
- void removeAcls(Admin admin) throws ExecutionException, InterruptedException {
- Map<ResourcePatternFilter, Set<AccessControlEntry>> filterToAcl = getResourceFilterToAcls(opts);
- for (Map.Entry<ResourcePatternFilter, Set<AccessControlEntry>> entry : filterToAcl.entrySet()) {
- ResourcePatternFilter filter = entry.getKey();
- Set<AccessControlEntry> acls = entry.getValue();
- if (acls.isEmpty()) {
- if (confirmAction(opts, "Are you sure you want to delete all ACLs for resource filter `" + filter + "`? (y/n)")) {
- removeAcls(admin, acls, filter);
- }
- } else {
- String msg = "Are you sure you want to remove ACLs: " + NL +
- " " + acls.stream().map(a -> "\t" + a).collect(Collectors.joining(NL)) + NL +
- " from resource filter `" + filter + "`? (y/n)";
- if (confirmAction(opts, msg)) {
- removeAcls(admin, acls, filter);
- }
+ private static void removeAcls(Admin admin, AclCommandOptions opts) throws ExecutionException, InterruptedException {
+ Map<ResourcePatternFilter, Set<AccessControlEntry>> filterToAcl = getResourceFilterToAcls(opts);
+ for (Map.Entry<ResourcePatternFilter, Set<AccessControlEntry>> entry : filterToAcl.entrySet()) {
+ ResourcePatternFilter filter = entry.getKey();
+ Set<AccessControlEntry> acls = entry.getValue();
+ if (acls.isEmpty()) {
+ if (confirmAction(opts, "Are you sure you want to delete all ACLs for resource filter `" + filter + "`? (y/n)")) {
+ removeAcls(admin, acls, filter);
+ }
+ } else {
+ String msg = "Are you sure you want to remove ACLs: " + NL +
+ " " + acls.stream().map(a -> "\t" + a).collect(Collectors.joining(NL)) + NL +
+ " from resource filter `" + filter + "`? (y/n)";
+ if (confirmAction(opts, msg)) {
+ removeAcls(admin, acls, filter);
}
}
}
+ }
- private void listAcls(Admin admin) throws ExecutionException, InterruptedException {
- Set<ResourcePatternFilter> filters = getResourceFilter(opts, false);
- Set<KafkaPrincipal> listPrincipals = getPrincipals(opts, opts.listPrincipalsOpt);
- Map<ResourcePattern, Set<AccessControlEntry>> resourceToAcls = getAcls(admin, filters);
+ private static void listAcls(Admin admin, AclCommandOptions opts) throws ExecutionException, InterruptedException {
+ Set<ResourcePatternFilter> filters = getResourceFilter(opts, false);
+ Set<KafkaPrincipal> listPrincipals = getPrincipals(opts, opts.listPrincipalsOpt);
+ Map<ResourcePattern, Set<AccessControlEntry>> resourceToAcls = getAcls(admin, filters);
- if (listPrincipals.isEmpty()) {
- printResourceAcls(resourceToAcls);
- } else {
- listPrincipals.forEach(principal -> {
- System.out.println("ACLs for principal `" + principal + "`");
- Map<ResourcePattern, Set<AccessControlEntry>> filteredResourceToAcls = resourceToAcls.entrySet().stream()
- .map(entry -> {
- ResourcePattern resource = entry.getKey();
- Set<AccessControlEntry> acls = entry.getValue().stream()
- .filter(acl -> principal.toString().equals(acl.principal()))
- .collect(Collectors.toSet());
- return new AbstractMap.SimpleEntry<>(resource, acls);
- })
- .filter(entry -> !entry.getValue().isEmpty())
- .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
- printResourceAcls(filteredResourceToAcls);
- });
- }
+ if (listPrincipals.isEmpty()) {
+ printResourceAcls(resourceToAcls);
+ } else {
+ listPrincipals.forEach(principal -> {
+ System.out.println("ACLs for principal `" + principal + "`");
+ Map<ResourcePattern, Set<AccessControlEntry>> filteredResourceToAcls = resourceToAcls.entrySet().stream()
+ .map(entry -> {
+ ResourcePattern resource = entry.getKey();
+ Set<AccessControlEntry> acls = entry.getValue().stream()
+ .filter(acl -> principal.toString().equals(acl.principal()))
+ .collect(Collectors.toSet());
+ return new AbstractMap.SimpleEntry<>(resource, acls);
+ })
+ .filter(entry -> !entry.getValue().isEmpty())
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ printResourceAcls(filteredResourceToAcls);
+ });
}
+ }
- private static void printResourceAcls(Map<ResourcePattern, Set<AccessControlEntry>> resourceToAcls) {
- resourceToAcls.forEach((resource, acls) ->
- System.out.println("Current ACLs for resource `" + resource + "`:" + NL +
- acls.stream().map(acl -> "\t" + acl).collect(Collectors.joining(NL)) + NL)
- );
- }
+ private static void printResourceAcls(Map<ResourcePattern, Set<AccessControlEntry>> resourceToAcls) {
+ resourceToAcls.forEach((resource, acls) ->
+ System.out.println("Current ACLs for resource `" + resource + "`:" + NL +
+ acls.stream().map(acl -> "\t" + acl).collect(Collectors.joining(NL)) + NL)
+ );
+ }
- private static void removeAcls(Admin adminClient, Set<AccessControlEntry> acls, ResourcePatternFilter filter) throws ExecutionException, InterruptedException {
- if (acls.isEmpty()) {
- adminClient.deleteAcls(List.of(new AclBindingFilter(filter, AccessControlEntryFilter.ANY))).all().get();
- } else {
- List<AclBindingFilter> aclBindingFilters = acls.stream().map(acl -> new AclBindingFilter(filter, acl.toFilter())).collect(Collectors.toList());
- adminClient.deleteAcls(aclBindingFilters).all().get();
- }
+ private static void removeAcls(Admin adminClient, Set<AccessControlEntry> acls, ResourcePatternFilter filter) throws ExecutionException, InterruptedException {
+ if (acls.isEmpty()) {
+ adminClient.deleteAcls(List.of(new AclBindingFilter(filter, AccessControlEntryFilter.ANY))).all().get();
+ } else {
+ List<AclBindingFilter> aclBindingFilters = acls.stream().map(acl -> new AclBindingFilter(filter, acl.toFilter())).collect(Collectors.toList());
+ adminClient.deleteAcls(aclBindingFilters).all().get();
}
+ }
- private Map<ResourcePattern, Set<AccessControlEntry>> getAcls(Admin adminClient, Set<ResourcePatternFilter> filters) throws ExecutionException, InterruptedException {
- Collection<AclBinding> aclBindings;
- if (filters.isEmpty()) {
- aclBindings = adminClient.describeAcls(AclBindingFilter.ANY).values().get();
- } else {
- aclBindings = new ArrayList<>();
- for (ResourcePatternFilter filter : filters) {
- aclBindings.addAll(adminClient.describeAcls(new AclBindingFilter(filter, AccessControlEntryFilter.ANY)).values().get());
- }
+ private static Map<ResourcePattern, Set<AccessControlEntry>> getAcls(Admin adminClient, Set<ResourcePatternFilter> filters) throws ExecutionException, InterruptedException {
+ Collection<AclBinding> aclBindings;
+ if (filters.isEmpty()) {
+ aclBindings = adminClient.describeAcls(AclBindingFilter.ANY).values().get();
+ } else {
+ aclBindings = new ArrayList<>();
+ for (ResourcePatternFilter filter : filters) {
+ aclBindings.addAll(adminClient.describeAcls(new AclBindingFilter(filter, AccessControlEntryFilter.ANY)).values().get());
}
+ }
- Map<ResourcePattern, Set<AccessControlEntry>> resourceToAcls = new HashMap<>();
- for (AclBinding aclBinding : aclBindings) {
- ResourcePattern resource = aclBinding.pattern();
- Set<AccessControlEntry> acls = resourceToAcls.getOrDefault(resource, new HashSet<>());
- acls.add(aclBinding.entry());
- resourceToAcls.put(resource, acls);
- }
- return resourceToAcls;
+ Map<ResourcePattern, Set<AccessControlEntry>> resourceToAcls = new HashMap<>();
+ for (AclBinding aclBinding : aclBindings) {
+ ResourcePattern resource = aclBinding.pattern();
+ Set<AccessControlEntry> acls = resourceToAcls.getOrDefault(resource, new HashSet<>());
+ acls.add(aclBinding.entry());
+ resourceToAcls.put(resource, acls);
}
+ return resourceToAcls;
}
private static Map<ResourcePattern, Set<AccessControlEntry>> getResourceToAcls(AclCommandOptions opts) {
diff --git a/tools/src/main/java/org/apache/kafka/tools/ConnectPluginPath.java b/tools/src/main/java/org/apache/kafka/tools/ConnectPluginPath.java
index c86638e846a..a3da1d3a9c6 100644
--- a/tools/src/main/java/org/apache/kafka/tools/ConnectPluginPath.java
+++ b/tools/src/main/java/org/apache/kafka/tools/ConnectPluginPath.java
@@ -196,22 +196,8 @@ public class ConnectPluginPath {
LIST, SYNC_MANIFESTS
}
- private static class Config {
- private final Command command;
- private final Set<Path> locations;
- private final boolean dryRun;
- private final boolean keepNotFound;
- private final PrintStream out;
- private final PrintStream err;
-
- private Config(Command command, Set<Path> locations, boolean dryRun, boolean keepNotFound, PrintStream out, PrintStream err) {
- this.command = command;
- this.locations = locations;
- this.dryRun = dryRun;
- this.keepNotFound = keepNotFound;
- this.out = out;
- this.err = err;
- }
+ private record Config(Command command, Set<Path> locations, boolean dryRun, boolean keepNotFound, PrintStream out,
+ PrintStream err) {
@Override
public String toString() {
@@ -262,16 +248,9 @@ public class ConnectPluginPath {
* <p>This is unique to the (source, class, type) tuple, and contains additional pre-computed information
* that pertains to this specific plugin.
*/
- private static class Row {
- private final ManifestWorkspace.SourceWorkspace<?> workspace;
- private final String className;
- private final PluginType type;
- private final String version;
- private final List<String> aliases;
- private final boolean loadable;
- private final boolean hasManifest;
-
- public Row(ManifestWorkspace.SourceWorkspace<?> workspace, String className, PluginType type, String version, List<String> aliases, boolean loadable, boolean hasManifest) {
+ private record Row(ManifestWorkspace.SourceWorkspace<?> workspace, String className, PluginType type,
+ String version, List<String> aliases, boolean loadable, boolean hasManifest) {
+ private Row(ManifestWorkspace.SourceWorkspace<?> workspace, String className, PluginType type, String version, List<String> aliases, boolean loadable, boolean hasManifest) {
this.workspace = Objects.requireNonNull(workspace, "workspace must be non-null");
this.className = Objects.requireNonNull(className, "className must be non-null");
this.version = Objects.requireNonNull(version, "version must be non-null");
@@ -281,10 +260,6 @@ public class ConnectPluginPath {
this.hasManifest = hasManifest;
}
- private boolean loadable() {
- return loadable;
- }
-
private boolean compatible() {
return loadable && hasManifest;
}
diff --git a/tools/src/main/java/org/apache/kafka/tools/MetadataQuorumCommand.java b/tools/src/main/java/org/apache/kafka/tools/MetadataQuorumCommand.java
index b65655a8c89..4abe443b82b 100644
--- a/tools/src/main/java/org/apache/kafka/tools/MetadataQuorumCommand.java
+++ b/tools/src/main/java/org/apache/kafka/tools/MetadataQuorumCommand.java
@@ -65,7 +65,6 @@ import java.util.stream.Stream;
import static java.lang.String.format;
import static java.lang.String.valueOf;
-import static java.util.Arrays.asList;
/**
* A tool for describing quorum status
@@ -206,7 +205,7 @@ public class MetadataQuorumCommand {
rows.addAll(quorumInfoToRows(leader, quorumInfo.observers().stream(), "Observer", humanReadable));
ToolsUtils.prettyPrintTable(
- asList("NodeId", "DirectoryId", "LogEndOffset", "Lag", "LastFetchTimestamp", "LastCaughtUpTimestamp", "Status"),
+ List.of("NodeId", "DirectoryId", "LogEndOffset", "Lag", "LastFetchTimestamp", "LastCaughtUpTimestamp", "Status"),
rows,
System.out
);
diff --git a/tools/src/main/java/org/apache/kafka/tools/OAuthCompatibilityTool.java b/tools/src/main/java/org/apache/kafka/tools/OAuthCompatibilityTool.java
index 40f3100054b..59dbb47daec 100644
--- a/tools/src/main/java/org/apache/kafka/tools/OAuthCompatibilityTool.java
+++ b/tools/src/main/java/org/apache/kafka/tools/OAuthCompatibilityTool.java
@@ -292,14 +292,7 @@ public class OAuthCompatibilityTool {
}
- private static class ConfigHandler {
-
- private final Namespace namespace;
-
-
- private ConfigHandler(Namespace namespace) {
- this.namespace = namespace;
- }
+ private record ConfigHandler(Namespace namespace) {
private Map<String, ?> getConfigs() {
Map<String, Object> m = new HashMap<>();
diff --git a/tools/src/main/java/org/apache/kafka/tools/OffsetsUtils.java b/tools/src/main/java/org/apache/kafka/tools/OffsetsUtils.java
index e37bf2804d7..269cd53875d 100644
--- a/tools/src/main/java/org/apache/kafka/tools/OffsetsUtils.java
+++ b/tools/src/main/java/org/apache/kafka/tools/OffsetsUtils.java
@@ -500,16 +500,7 @@ public class OffsetsUtils {
public interface LogOffsetResult { }
- public static class LogOffset implements LogOffsetResult {
- final long value;
-
- public LogOffset(long value) {
- this.value = value;
- }
-
- public long value() {
- return value;
- }
+ public record LogOffset(long value) implements LogOffsetResult {
}
public static class Unknown implements LogOffsetResult { }
diff --git a/tools/src/main/java/org/apache/kafka/tools/PushHttpMetricsReporter.java b/tools/src/main/java/org/apache/kafka/tools/PushHttpMetricsReporter.java
index 86ea19f0623..9b423b78947 100644
--- a/tools/src/main/java/org/apache/kafka/tools/PushHttpMetricsReporter.java
+++ b/tools/src/main/java/org/apache/kafka/tools/PushHttpMetricsReporter.java
@@ -240,86 +240,19 @@ public class PushHttpMetricsReporter implements MetricsReporter {
}
}
- private static class MetricsReport {
- private final MetricClientInfo client;
- private final Collection<MetricValue> metrics;
-
- MetricsReport(MetricClientInfo client, Collection<MetricValue> metrics) {
- this.client = client;
- this.metrics = metrics;
- }
-
- @JsonProperty
- public MetricClientInfo client() {
- return client;
- }
-
- @JsonProperty
- public Collection<MetricValue> metrics() {
- return metrics;
- }
+ private record MetricsReport(@JsonProperty("client") MetricClientInfo client,
+ @JsonProperty("metrics") Collection<MetricValue> metrics) {
}
- private static class MetricClientInfo {
- private final String host;
- private final String clientId;
- private final long time;
-
- MetricClientInfo(String host, String clientId, long time) {
- this.host = host;
- this.clientId = clientId;
- this.time = time;
- }
-
- @JsonProperty
- public String host() {
- return host;
- }
-
- @JsonProperty("client_id")
- public String clientId() {
- return clientId;
- }
-
- @JsonProperty
- public long time() {
- return time;
- }
+ private record MetricClientInfo(@JsonProperty("host") String host,
+ @JsonProperty("client_id") String clientId,
+ @JsonProperty("time") long time) {
}
- private static class MetricValue {
-
- private final String name;
- private final String group;
- private final Map<String, String> tags;
- private final Object value;
-
- MetricValue(String name, String group, Map<String, String> tags, Object value) {
- this.name = name;
- this.group = group;
- this.tags = tags;
- this.value = value;
- }
-
- @JsonProperty
- public String name() {
- return name;
- }
-
- @JsonProperty
- public String group() {
- return group;
- }
-
- @JsonProperty
- public Map<String, String> tags() {
- return tags;
- }
-
- @JsonProperty
- public Object value() {
- return value;
- }
+ private record MetricValue(@JsonProperty("name") String name,
+ @JsonProperty("group") String group,
+ @JsonProperty("tags") Map<String, String> tags,
+ @JsonProperty("value") Object value) {
}
// The signature for getInt changed from returning int to Integer so to remain compatible with 0.8.2.2 jars
diff --git a/tools/src/main/java/org/apache/kafka/tools/ReplicaVerificationTool.java b/tools/src/main/java/org/apache/kafka/tools/ReplicaVerificationTool.java
index c5f1ddcc2d4..937e1cc5955 100644
--- a/tools/src/main/java/org/apache/kafka/tools/ReplicaVerificationTool.java
+++ b/tools/src/main/java/org/apache/kafka/tools/ReplicaVerificationTool.java
@@ -348,18 +348,7 @@ public class ReplicaVerificationTool {
}
}
- private static class MessageInfo {
- final int replicaId;
- final long offset;
- final long nextOffset;
- final long checksum;
-
- MessageInfo(int replicaId, long offset, long nextOffset, long checksum) {
- this.replicaId = replicaId;
- this.offset = offset;
- this.nextOffset = nextOffset;
- this.checksum = checksum;
- }
+ private record MessageInfo(int replicaId, long offset, long nextOffset, long checksum) {
}
protected static class ReplicaBuffer {
diff --git a/tools/src/main/java/org/apache/kafka/tools/TransactionsCommand.java b/tools/src/main/java/org/apache/kafka/tools/TransactionsCommand.java
index 9c5323104fe..5b59fbee4a1 100644
--- a/tools/src/main/java/org/apache/kafka/tools/TransactionsCommand.java
+++ b/tools/src/main/java/org/apache/kafka/tools/TransactionsCommand.java
@@ -63,7 +63,6 @@ import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;
-import static java.util.Arrays.asList;
import static net.sourceforge.argparse4j.impl.Arguments.store;
public abstract class TransactionsCommand {
@@ -288,7 +287,7 @@ public abstract class TransactionsCommand {
}
static class DescribeProducersCommand extends TransactionsCommand {
- static final List<String> HEADERS = asList(
+ static final List<String> HEADERS = List.of(
"ProducerId",
"ProducerEpoch",
"LatestCoordinatorEpoch",
@@ -360,7 +359,7 @@ public abstract class TransactionsCommand {
String.valueOf(producerState.currentTransactionStartOffset().getAsLong()) :
"None";
- return asList(
+ return List.of(
String.valueOf(producerState.producerId()),
String.valueOf(producerState.producerEpoch()),
String.valueOf(producerState.coordinatorEpoch().orElse(-1)),
@@ -375,7 +374,7 @@ public abstract class TransactionsCommand {
}
static class DescribeTransactionsCommand extends TransactionsCommand {
- static final List<String> HEADERS = asList(
+ static final List<String> HEADERS = List.of(
"CoordinatorId",
"TransactionalId",
"ProducerId",
@@ -436,7 +435,7 @@ public abstract class TransactionsCommand {
transactionDurationMsColumnValue = "None";
}
- List<String> row = asList(
+ List<String> row = List.of(
String.valueOf(result.coordinatorId()),
transactionalId,
String.valueOf(result.producerId()),
@@ -453,7 +452,7 @@ public abstract class TransactionsCommand {
}
static class ListTransactionsCommand extends TransactionsCommand {
- static final List<String> HEADERS = asList(
+ static final List<String> HEADERS = List.of(
"TransactionalId",
"Coordinator",
"ProducerId",
@@ -510,7 +509,7 @@ public abstract class TransactionsCommand {
Collection<TransactionListing> listings = brokerListingsEntry.getValue();
for (TransactionListing listing : listings) {
- rows.add(asList(
+ rows.add(List.of(
listing.transactionalId(),
coordinatorIdString,
String.valueOf(listing.producerId()),
@@ -526,7 +525,7 @@ public abstract class TransactionsCommand {
static class FindHangingTransactionsCommand extends TransactionsCommand {
private static final int MAX_BATCH_SIZE = 500;
- static final List<String> HEADERS = asList(
+ static final List<String> HEADERS = List.of(
"Topic",
"Partition",
"ProducerId",
@@ -709,7 +708,7 @@ public abstract class TransactionsCommand {
long transactionDurationMinutes = TimeUnit.MILLISECONDS.toMinutes(
currentTimeMs - transaction.producerState.lastTimestamp());
- rows.add(asList(
+ rows.add(List.of(
transaction.topicPartition.topic(),
String.valueOf(transaction.topicPartition.partition()),
String.valueOf(transaction.producerState.producerId()),
@@ -848,17 +847,7 @@ public abstract class TransactionsCommand {
return candidateTransactions;
}
- private static class OpenTransaction {
- private final TopicPartition topicPartition;
- private final ProducerState producerState;
-
- private OpenTransaction(
- TopicPartition topicPartition,
- ProducerState producerState
- ) {
- this.topicPartition = topicPartition;
- this.producerState = producerState;
- }
+ private record OpenTransaction(TopicPartition topicPartition, ProducerState producerState) {
}
private void collectCandidateOpenTransactions(
@@ -1024,7 +1013,7 @@ public abstract class TransactionsCommand {
PrintStream out,
Time time
) throws Exception {
- List<TransactionsCommand> commands = asList(
+ List<TransactionsCommand> commands = List.of(
new ListTransactionsCommand(time),
new DescribeTransactionsCommand(time),
new DescribeProducersCommand(time),
diff --git a/tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java b/tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java
index 1f29cdd8156..cfdef8f7518 100644
--- a/tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java
+++ b/tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java
@@ -347,20 +347,20 @@ public class ConsumerGroupCommand {
for (PartitionAssignmentState consumerAssignment : consumerAssignments) {
if (verbose) {
System.out.printf(format,
- consumerAssignment.group,
- consumerAssignment.topic.orElse(MISSING_COLUMN_VALUE), consumerAssignment.partition.map(Object::toString).orElse(MISSING_COLUMN_VALUE),
- consumerAssignment.leaderEpoch.map(Object::toString).orElse(MISSING_COLUMN_VALUE),
- consumerAssignment.offset.map(Object::toString).orElse(MISSING_COLUMN_VALUE), consumerAssignment.logEndOffset.map(Object::toString).orElse(MISSING_COLUMN_VALUE),
- consumerAssignment.lag.map(Object::toString).orElse(MISSING_COLUMN_VALUE), consumerAssignment.consumerId.orElse(MISSING_COLUMN_VALUE),
- consumerAssignment.host.orElse(MISSING_COLUMN_VALUE), consumerAssignment.clientId.orElse(MISSING_COLUMN_VALUE)
+ consumerAssignment.group(),
+ consumerAssignment.topic().orElse(MISSING_COLUMN_VALUE), consumerAssignment.partition().map(Object::toString).orElse(MISSING_COLUMN_VALUE),
+ consumerAssignment.leaderEpoch().map(Object::toString).orElse(MISSING_COLUMN_VALUE),
+ consumerAssignment.offset().map(Object::toString).orElse(MISSING_COLUMN_VALUE), consumerAssignment.logEndOffset().map(Object::toString).orElse(MISSING_COLUMN_VALUE),
+ consumerAssignment.lag().map(Object::toString).orElse(MISSING_COLUMN_VALUE), consumerAssignment.consumerId().orElse(MISSING_COLUMN_VALUE),
+ consumerAssignment.host().orElse(MISSING_COLUMN_VALUE), consumerAssignment.clientId().orElse(MISSING_COLUMN_VALUE)
);
} else {
System.out.printf(format,
- consumerAssignment.group,
- consumerAssignment.topic.orElse(MISSING_COLUMN_VALUE), consumerAssignment.partition.map(Object::toString).orElse(MISSING_COLUMN_VALUE),
- consumerAssignment.offset.map(Object::toString).orElse(MISSING_COLUMN_VALUE), consumerAssignment.logEndOffset.map(Object::toString).orElse(MISSING_COLUMN_VALUE),
- consumerAssignment.lag.map(Object::toString).orElse(MISSING_COLUMN_VALUE), consumerAssignment.consumerId.orElse(MISSING_COLUMN_VALUE),
- consumerAssignment.host.orElse(MISSING_COLUMN_VALUE), consumerAssignment.clientId.orElse(MISSING_COLUMN_VALUE)
+ consumerAssignment.group(),
+ consumerAssignment.topic().orElse(MISSING_COLUMN_VALUE), consumerAssignment.partition().map(Object::toString).orElse(MISSING_COLUMN_VALUE),
+ consumerAssignment.offset().map(Object::toString).orElse(MISSING_COLUMN_VALUE), consumerAssignment.logEndOffset().map(Object::toString).orElse(MISSING_COLUMN_VALUE),
+ consumerAssignment.lag().map(Object::toString).orElse(MISSING_COLUMN_VALUE), consumerAssignment.consumerId().orElse(MISSING_COLUMN_VALUE),
+ consumerAssignment.host().orElse(MISSING_COLUMN_VALUE), consumerAssignment.clientId().orElse(MISSING_COLUMN_VALUE)
);
}
}
@@ -379,10 +379,10 @@ public class ConsumerGroupCommand {
if (assignments.isPresent()) {
Collection<PartitionAssignmentState> consumerAssignments = assignments.get();
for (PartitionAssignmentState consumerAssignment : consumerAssignments) {
- maxGroupLen = Math.max(maxGroupLen, consumerAssignment.group.length());
- maxTopicLen = Math.max(maxTopicLen, consumerAssignment.topic.orElse(MISSING_COLUMN_VALUE).length());
- maxConsumerIdLen = Math.max(maxConsumerIdLen, consumerAssignment.consumerId.orElse(MISSING_COLUMN_VALUE).length());
- maxHostLen = Math.max(maxHostLen, consumerAssignment.host.orElse(MISSING_COLUMN_VALUE).length());
+ maxGroupLen = Math.max(maxGroupLen, consumerAssignment.group().length());
+ maxTopicLen = Math.max(maxTopicLen, consumerAssignment.topic().orElse(MISSING_COLUMN_VALUE).length());
+ maxConsumerIdLen = Math.max(maxConsumerIdLen, consumerAssignment.consumerId().orElse(MISSING_COLUMN_VALUE).length());
+ maxHostLen = Math.max(maxHostLen, consumerAssignment.host().orElse(MISSING_COLUMN_VALUE).length());
}
}
@@ -408,20 +408,20 @@ public class ConsumerGroupCommand {
// find proper columns width
if (assignments.isPresent()) {
for (MemberAssignmentState memberAssignment : assignments.get()) {
- maxGroupLen = Math.max(maxGroupLen, memberAssignment.group.length());
- maxConsumerIdLen = Math.max(maxConsumerIdLen, memberAssignment.consumerId.length());
- maxGroupInstanceIdLen = Math.max(maxGroupInstanceIdLen, memberAssignment.groupInstanceId.length());
- maxHostLen = Math.max(maxHostLen, memberAssignment.host.length());
- maxClientIdLen = Math.max(maxClientIdLen, memberAssignment.clientId.length());
- includeGroupInstanceId = includeGroupInstanceId || !memberAssignment.groupInstanceId.isEmpty();
- String currentAssignment = memberAssignment.assignment.isEmpty() ?
- MISSING_COLUMN_VALUE : getAssignmentString(memberAssignment.assignment);
- String targetAssignment = memberAssignment.targetAssignment.isEmpty() ?
- MISSING_COLUMN_VALUE : getAssignmentString(memberAssignment.targetAssignment);
+ maxGroupLen = Math.max(maxGroupLen, memberAssignment.group().length());
+ maxConsumerIdLen = Math.max(maxConsumerIdLen, memberAssignment.consumerId().length());
+ maxGroupInstanceIdLen = Math.max(maxGroupInstanceIdLen, memberAssignment.groupInstanceId().length());
+ maxHostLen = Math.max(maxHostLen, memberAssignment.host().length());
+ maxClientIdLen = Math.max(maxClientIdLen, memberAssignment.clientId().length());
+ includeGroupInstanceId = includeGroupInstanceId || !memberAssignment.groupInstanceId().isEmpty();
+ String currentAssignment = memberAssignment.assignment().isEmpty() ?
+ MISSING_COLUMN_VALUE : getAssignmentString(memberAssignment.assignment());
+ String targetAssignment = memberAssignment.targetAssignment().isEmpty() ?
+ MISSING_COLUMN_VALUE : getAssignmentString(memberAssignment.targetAssignment());
maxCurrentAssignment = Math.max(maxCurrentAssignment, currentAssignment.length());
maxTargetAssignment = Math.max(maxTargetAssignment, targetAssignment.length());
- hasClassicMember = hasClassicMember || (memberAssignment.upgraded.isPresent() && !memberAssignment.upgraded.get());
- hasConsumerMember = hasConsumerMember || (memberAssignment.upgraded.isPresent() && memberAssignment.upgraded.get());
+ hasClassicMember = hasClassicMember || (memberAssignment.upgraded().isPresent() && !memberAssignment.upgraded().get());
+ hasConsumerMember = hasConsumerMember || (memberAssignment.upgraded().isPresent() && memberAssignment.upgraded().get());
}
}
}
@@ -465,23 +465,23 @@ public class ConsumerGroupCommand {
) {
for (MemberAssignmentState memberAssignment : memberAssignments) {
if (includeGroupInstanceId) {
- System.out.printf(formatWithGroupInstanceId, memberAssignment.group, memberAssignment.consumerId,
- memberAssignment.groupInstanceId, memberAssignment.host, memberAssignment.clientId,
- memberAssignment.numPartitions);
+ System.out.printf(formatWithGroupInstanceId, memberAssignment.group(), memberAssignment.consumerId(),
+ memberAssignment.groupInstanceId(), memberAssignment.host(), memberAssignment.clientId(),
+ memberAssignment.numPartitions());
} else {
- System.out.printf(formatWithoutGroupInstanceId, memberAssignment.group, memberAssignment.consumerId,
- memberAssignment.host, memberAssignment.clientId, memberAssignment.numPartitions);
+ System.out.printf(formatWithoutGroupInstanceId, memberAssignment.group(), memberAssignment.consumerId(),
+ memberAssignment.host(), memberAssignment.clientId(), memberAssignment.numPartitions());
}
if (verbose) {
- String currentEpoch = memberAssignment.currentEpoch.map(Object::toString).orElse(MISSING_COLUMN_VALUE);
- String currentAssignment = memberAssignment.assignment.isEmpty() ?
- MISSING_COLUMN_VALUE : getAssignmentString(memberAssignment.assignment);
- String targetEpoch = memberAssignment.targetEpoch.map(Object::toString).orElse(MISSING_COLUMN_VALUE);
- String targetAssignment = memberAssignment.targetAssignment.isEmpty() ?
- MISSING_COLUMN_VALUE : getAssignmentString(memberAssignment.targetAssignment);
+ String currentEpoch = memberAssignment.currentEpoch().map(Object::toString).orElse(MISSING_COLUMN_VALUE);
+ String currentAssignment = memberAssignment.assignment().isEmpty() ?
+ MISSING_COLUMN_VALUE : getAssignmentString(memberAssignment.assignment());
+ String targetEpoch = memberAssignment.targetEpoch().map(Object::toString).orElse(MISSING_COLUMN_VALUE);
+ String targetAssignment = memberAssignment.targetAssignment().isEmpty() ?
+ MISSING_COLUMN_VALUE : getAssignmentString(memberAssignment.targetAssignment());
if (hasMigrationMember) {
System.out.printf(formatWithUpgrade, currentEpoch, currentAssignment, targetEpoch, targetAssignment,
- memberAssignment.upgraded.map(Object::toString).orElse(MISSING_COLUMN_VALUE));
+ memberAssignment.upgraded().map(Object::toString).orElse(MISSING_COLUMN_VALUE));
} else {
System.out.printf(formatWithoutUpgrade, currentEpoch, currentAssignment, targetEpoch, targetAssignment);
}
@@ -511,23 +511,23 @@ public class ConsumerGroupCommand {
private void printStates(Map<String, GroupInformation> states, boolean verbose) {
states.forEach((groupId, state) -> {
- if (shouldPrintMemberState(groupId, Optional.of(state.groupState), Optional.of(1))) {
- String coordinator = state.coordinator.host() + ":" + state.coordinator.port() + " (" + state.coordinator.idString() + ")";
+ if (shouldPrintMemberState(groupId, Optional.of(state.groupState()), Optional.of(1))) {
+ String coordinator = state.coordinator().host() + ":" + state.coordinator().port() + " (" + state.coordinator().idString() + ")";
int coordinatorColLen = Math.max(25, coordinator.length());
- int groupColLen = Math.max(15, state.group.length());
+ int groupColLen = Math.max(15, state.group().length());
- String assignmentStrategy = state.assignmentStrategy.isEmpty() ? MISSING_COLUMN_VALUE : state.assignmentStrategy;
+ String assignmentStrategy = state.assignmentStrategy().isEmpty() ? MISSING_COLUMN_VALUE : state.assignmentStrategy();
if (verbose) {
String format = "\n%" + -groupColLen + "s %" + -coordinatorColLen + "s %-20s %-20s %-15s %-25s %s";
System.out.printf(format, "GROUP", "COORDINATOR (ID)", "ASSIGNMENT-STRATEGY", "STATE",
"GROUP-EPOCH", "TARGET-ASSIGNMENT-EPOCH", "#MEMBERS");
- System.out.printf(format, state.group, coordinator, assignmentStrategy, state.groupState,
- state.groupEpoch.map(Object::toString).orElse(MISSING_COLUMN_VALUE), state.targetAssignmentEpoch.map(Object::toString).orElse(MISSING_COLUMN_VALUE), state.numMembers);
+ System.out.printf(format, state.group(), coordinator, assignmentStrategy, state.groupState(),
+ state.groupEpoch().map(Object::toString).orElse(MISSING_COLUMN_VALUE), state.targetAssignmentEpoch().map(Object::toString).orElse(MISSING_COLUMN_VALUE), state.numMembers());
} else {
String format = "\n%" + -groupColLen + "s %" + -coordinatorColLen + "s %-20s %-20s %s";
System.out.printf(format, "GROUP", "COORDINATOR (ID)", "ASSIGNMENT-STRATEGY", "STATE", "#MEMBERS");
- System.out.printf(format, state.group, coordinator, assignmentStrategy, state.groupState, state.numMembers);
+ System.out.printf(format, state.group(), coordinator, assignmentStrategy, state.groupState(), state.numMembers());
}
System.out.println();
}
@@ -623,8 +623,8 @@ public class ConsumerGroupCommand {
// concat the data and then sort them
return Stream.concat(existLeaderAssignments.stream(), noneLeaderAssignments.stream())
.sorted(Comparator.<PartitionAssignmentState, String>comparing(
- state -> state.topic.orElse(""), String::compareTo)
- .thenComparingInt(state -> state.partition.orElse(-1)))
+ state -> state.topic().orElse(""), String::compareTo)
+ .thenComparingInt(state -> state.partition().orElse(-1)))
.collect(Collectors.toList());
}
diff --git a/tools/src/main/java/org/apache/kafka/tools/consumer/group/GroupInformation.java b/tools/src/main/java/org/apache/kafka/tools/consumer/group/GroupInformation.java
index 9fde352cdda..d313390b6a9 100644
--- a/tools/src/main/java/org/apache/kafka/tools/consumer/group/GroupInformation.java
+++ b/tools/src/main/java/org/apache/kafka/tools/consumer/group/GroupInformation.java
@@ -21,30 +21,6 @@ import org.apache.kafka.common.Node;
import java.util.Optional;
-class GroupInformation {
- final String group;
- final Node coordinator;
- final String assignmentStrategy;
- final GroupState groupState;
- final int numMembers;
- final Optional<Integer> groupEpoch;
- final Optional<Integer> targetAssignmentEpoch;
-
- GroupInformation(
- String group,
- Node coordinator,
- String assignmentStrategy,
- GroupState groupState,
- int numMembers,
- Optional<Integer> groupEpoch,
- Optional<Integer> targetAssignmentEpoch
- ) {
- this.group = group;
- this.coordinator = coordinator;
- this.assignmentStrategy = assignmentStrategy;
- this.groupState = groupState;
- this.numMembers = numMembers;
- this.groupEpoch = groupEpoch;
- this.targetAssignmentEpoch = targetAssignmentEpoch;
- }
+record GroupInformation(String group, Node coordinator, String assignmentStrategy, GroupState groupState,
+ int numMembers, Optional<Integer> groupEpoch, Optional<Integer> targetAssignmentEpoch) {
}
diff --git a/tools/src/main/java/org/apache/kafka/tools/consumer/group/MemberAssignmentState.java b/tools/src/main/java/org/apache/kafka/tools/consumer/group/MemberAssignmentState.java
index 420e640419e..56feb8b029f 100644
--- a/tools/src/main/java/org/apache/kafka/tools/consumer/group/MemberAssignmentState.java
+++ b/tools/src/main/java/org/apache/kafka/tools/consumer/group/MemberAssignmentState.java
@@ -21,42 +21,8 @@ import org.apache.kafka.common.TopicPartition;
import java.util.List;
import java.util.Optional;
-class MemberAssignmentState {
- final String group;
- final String consumerId;
- final String host;
- final String clientId;
- final String groupInstanceId;
- final int numPartitions;
- final List<TopicPartition> assignment;
- final List<TopicPartition> targetAssignment;
- final Optional<Integer> currentEpoch;
- final Optional<Integer> targetEpoch;
- final Optional<Boolean> upgraded;
-
- MemberAssignmentState(
- String group,
- String consumerId,
- String host,
- String clientId,
- String groupInstanceId,
- int numPartitions,
- List<TopicPartition> assignment,
- List<TopicPartition> targetAssignment,
- Optional<Integer> currentEpoch,
- Optional<Integer> targetEpoch,
- Optional<Boolean> upgraded
- ) {
- this.group = group;
- this.consumerId = consumerId;
- this.host = host;
- this.clientId = clientId;
- this.groupInstanceId = groupInstanceId;
- this.numPartitions = numPartitions;
- this.assignment = assignment;
- this.targetAssignment = targetAssignment;
- this.currentEpoch = currentEpoch;
- this.targetEpoch = targetEpoch;
- this.upgraded = upgraded;
- }
+record MemberAssignmentState(String group, String consumerId, String host, String clientId, String groupInstanceId,
+ int numPartitions, List<TopicPartition> assignment, List<TopicPartition> targetAssignment,
+ Optional<Integer> currentEpoch, Optional<Integer> targetEpoch,
+ Optional<Boolean> upgraded) {
}
diff --git a/tools/src/main/java/org/apache/kafka/tools/consumer/group/PartitionAssignmentState.java b/tools/src/main/java/org/apache/kafka/tools/consumer/group/PartitionAssignmentState.java
index 53f5c391100..d42377ca029 100644
--- a/tools/src/main/java/org/apache/kafka/tools/consumer/group/PartitionAssignmentState.java
+++ b/tools/src/main/java/org/apache/kafka/tools/consumer/group/PartitionAssignmentState.java
@@ -20,42 +20,8 @@ import org.apache.kafka.common.Node;
import java.util.Optional;
-class PartitionAssignmentState {
- final String group;
- final Optional<Node> coordinator;
- final Optional<String> topic;
- final Optional<Integer> partition;
- final Optional<Long> offset;
- final Optional<Long> lag;
- final Optional<String> consumerId;
- final Optional<String> host;
- final Optional<String> clientId;
- final Optional<Long> logEndOffset;
- final Optional<Integer> leaderEpoch;
-
- PartitionAssignmentState(
- String group,
- Optional<Node> coordinator,
- Optional<String> topic,
- Optional<Integer> partition,
- Optional<Long> offset,
- Optional<Long> lag,
- Optional<String> consumerId,
- Optional<String> host,
- Optional<String> clientId,
- Optional<Long> logEndOffset,
- Optional<Integer> leaderEpoch
- ) {
- this.group = group;
- this.coordinator = coordinator;
- this.topic = topic;
- this.partition = partition;
- this.offset = offset;
- this.lag = lag;
- this.consumerId = consumerId;
- this.host = host;
- this.clientId = clientId;
- this.logEndOffset = logEndOffset;
- this.leaderEpoch = leaderEpoch;
- }
+record PartitionAssignmentState(String group, Optional<Node> coordinator, Optional<String> topic,
+ Optional<Integer> partition, Optional<Long> offset, Optional<Long> lag,
+ Optional<String> consumerId, Optional<String> host, Optional<String> clientId,
+ Optional<Long> logEndOffset, Optional<Integer> leaderEpoch) {
}
diff --git a/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java b/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java
index 95723665032..df4353c467a 100644
--- a/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java
+++ b/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java
@@ -656,25 +656,7 @@ public class ShareGroupCommand {
}
}
- static class SharePartitionOffsetInformation {
- final String group;
- final String topic;
- final int partition;
- final Optional<Long> offset;
- final Optional<Integer> leaderEpoch;
-
- SharePartitionOffsetInformation(
- String group,
- String topic,
- int partition,
- Optional<Long> offset,
- Optional<Integer> leaderEpoch
- ) {
- this.group = group;
- this.topic = topic;
- this.partition = partition;
- this.offset = offset;
- this.leaderEpoch = leaderEpoch;
- }
+ record SharePartitionOffsetInformation(String group, String topic, int partition, Optional<Long> offset,
+ Optional<Integer> leaderEpoch) {
}
}
diff --git a/tools/src/main/java/org/apache/kafka/tools/reassign/ActiveMoveState.java b/tools/src/main/java/org/apache/kafka/tools/reassign/ActiveMoveState.java
index 842d46ec587..58e2cf9b80f 100644
--- a/tools/src/main/java/org/apache/kafka/tools/reassign/ActiveMoveState.java
+++ b/tools/src/main/java/org/apache/kafka/tools/reassign/ActiveMoveState.java
@@ -17,44 +17,15 @@
package org.apache.kafka.tools.reassign;
-import java.util.Objects;
-
/**
* A replica log directory move state where the move is in progress.
+ * @param currentLogDir The current log directory.
+ * @param futureLogDir The log directory that the replica is moving to.
+ * @param targetLogDir The log directory that we wanted the replica to move to.
*/
-final class ActiveMoveState implements LogDirMoveState {
- public final String currentLogDir;
-
- public final String targetLogDir;
-
- public final String futureLogDir;
-
- /**
- * @param currentLogDir The current log directory.
- * @param futureLogDir The log directory that the replica is moving to.
- * @param targetLogDir The log directory that we wanted the replica to move to.
- */
- public ActiveMoveState(String currentLogDir, String targetLogDir, String futureLogDir) {
- this.currentLogDir = currentLogDir;
- this.targetLogDir = targetLogDir;
- this.futureLogDir = futureLogDir;
- }
-
+record ActiveMoveState(String currentLogDir, String targetLogDir, String futureLogDir) implements LogDirMoveState {
@Override
public boolean done() {
return false;
}
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- ActiveMoveState that = (ActiveMoveState) o;
- return Objects.equals(currentLogDir, that.currentLogDir) && Objects.equals(targetLogDir, that.targetLogDir) && Objects.equals(futureLogDir, that.futureLogDir);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(currentLogDir, targetLogDir, futureLogDir);
- }
}
diff --git a/tools/src/main/java/org/apache/kafka/tools/reassign/CancelledMoveState.java b/tools/src/main/java/org/apache/kafka/tools/reassign/CancelledMoveState.java
index f405eedd403..c7fa61af3d9 100644
--- a/tools/src/main/java/org/apache/kafka/tools/reassign/CancelledMoveState.java
+++ b/tools/src/main/java/org/apache/kafka/tools/reassign/CancelledMoveState.java
@@ -17,41 +17,15 @@
package org.apache.kafka.tools.reassign;
-import java.util.Objects;
-
/**
* A replica log directory move state where there is no move in progress, but we did not
* reach the target log directory.
+ * @param currentLogDir The current log directory.
+ * @param targetLogDir The log directory that we wanted the replica to move to.
*/
-final class CancelledMoveState implements LogDirMoveState {
- public final String currentLogDir;
-
- public final String targetLogDir;
-
- /**
- * @param currentLogDir The current log directory.
- * @param targetLogDir The log directory that we wanted the replica to move to.
- */
- public CancelledMoveState(String currentLogDir, String targetLogDir) {
- this.currentLogDir = currentLogDir;
- this.targetLogDir = targetLogDir;
- }
-
+record CancelledMoveState(String currentLogDir, String targetLogDir) implements LogDirMoveState {
@Override
public boolean done() {
return true;
}
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- CancelledMoveState that = (CancelledMoveState) o;
- return Objects.equals(currentLogDir, that.currentLogDir) && Objects.equals(targetLogDir, that.targetLogDir);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(currentLogDir, targetLogDir);
- }
}
diff --git a/tools/src/main/java/org/apache/kafka/tools/reassign/CompletedMoveState.java b/tools/src/main/java/org/apache/kafka/tools/reassign/CompletedMoveState.java
index df5fb890148..28b017b7cdc 100644
--- a/tools/src/main/java/org/apache/kafka/tools/reassign/CompletedMoveState.java
+++ b/tools/src/main/java/org/apache/kafka/tools/reassign/CompletedMoveState.java
@@ -17,36 +17,13 @@
package org.apache.kafka.tools.reassign;
-import java.util.Objects;
-
/**
* The completed replica log directory move state.
+ * @param targetLogDir The log directory that we wanted the replica to move to.
*/
-final class CompletedMoveState implements LogDirMoveState {
- public final String targetLogDir;
-
- /**
- * @param targetLogDir The log directory that we wanted the replica to move to.
- */
- public CompletedMoveState(String targetLogDir) {
- this.targetLogDir = targetLogDir;
- }
-
+record CompletedMoveState(String targetLogDir) implements LogDirMoveState {
@Override
public boolean done() {
return true;
}
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- CompletedMoveState that = (CompletedMoveState) o;
- return Objects.equals(targetLogDir, that.targetLogDir);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(targetLogDir);
- }
}
diff --git a/tools/src/main/java/org/apache/kafka/tools/reassign/MissingLogDirMoveState.java b/tools/src/main/java/org/apache/kafka/tools/reassign/MissingLogDirMoveState.java
index eb3f592841c..44982a8b712 100644
--- a/tools/src/main/java/org/apache/kafka/tools/reassign/MissingLogDirMoveState.java
+++ b/tools/src/main/java/org/apache/kafka/tools/reassign/MissingLogDirMoveState.java
@@ -17,36 +17,13 @@
package org.apache.kafka.tools.reassign;
-import java.util.Objects;
-
/**
* A replica log directory move state where the source replica is missing.
+ * @param targetLogDir The log directory that we wanted the replica to move to.
*/
-final class MissingLogDirMoveState implements LogDirMoveState {
- public final String targetLogDir;
-
- /**
- * @param targetLogDir The log directory that we wanted the replica to move to.
- */
- public MissingLogDirMoveState(String targetLogDir) {
- this.targetLogDir = targetLogDir;
- }
-
+record MissingLogDirMoveState(String targetLogDir) implements LogDirMoveState {
@Override
public boolean done() {
return false;
}
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- MissingLogDirMoveState that = (MissingLogDirMoveState) o;
- return Objects.equals(targetLogDir, that.targetLogDir);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(targetLogDir);
- }
}
diff --git a/tools/src/main/java/org/apache/kafka/tools/reassign/MissingReplicaMoveState.java b/tools/src/main/java/org/apache/kafka/tools/reassign/MissingReplicaMoveState.java
index eda9c22b829..b802275ceeb 100644
--- a/tools/src/main/java/org/apache/kafka/tools/reassign/MissingReplicaMoveState.java
+++ b/tools/src/main/java/org/apache/kafka/tools/reassign/MissingReplicaMoveState.java
@@ -17,36 +17,13 @@
package org.apache.kafka.tools.reassign;
-import java.util.Objects;
-
/**
* A replica log directory move state where the source log directory is missing.
+ * @param targetLogDir The log directory that we wanted the replica to move to.
*/
-final class MissingReplicaMoveState implements LogDirMoveState {
- public final String targetLogDir;
-
- /**
- * @param targetLogDir The log directory that we wanted the replica to move to.
- */
- public MissingReplicaMoveState(String targetLogDir) {
- this.targetLogDir = targetLogDir;
- }
-
+record MissingReplicaMoveState(String targetLogDir) implements LogDirMoveState {
@Override
public boolean done() {
return false;
}
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- MissingReplicaMoveState that = (MissingReplicaMoveState) o;
- return Objects.equals(targetLogDir, that.targetLogDir);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(targetLogDir);
- }
}
diff --git a/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java b/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java
index 591c127ae2f..4628c34fe18 100644
--- a/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java
+++ b/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java
@@ -467,8 +467,8 @@ public class ReassignPartitionsCommand {
bld.add("Partition " + replica.topic() + "-" + replica.partition() + " cannot be found " +
"in any live log directory on broker " + replica.brokerId() + ".");
} else if (state instanceof ActiveMoveState) {
- String targetLogDir = ((ActiveMoveState) state).targetLogDir;
- String futureLogDir = ((ActiveMoveState) state).futureLogDir;
+ String targetLogDir = ((ActiveMoveState) state).targetLogDir();
+ String futureLogDir = ((ActiveMoveState) state).futureLogDir();
if (targetLogDir.equals(futureLogDir)) {
bld.add("Reassignment of replica " + replica + " is still in progress.");
} else {
@@ -477,8 +477,8 @@ public class ReassignPartitionsCommand {
"instead of " + targetLogDir + ".");
}
} else if (state instanceof CancelledMoveState) {
- String targetLogDir = ((CancelledMoveState) state).targetLogDir;
- String currentLogDir = ((CancelledMoveState) state).currentLogDir;
+ String targetLogDir = ((CancelledMoveState) state).targetLogDir();
+ String currentLogDir = ((CancelledMoveState) state).currentLogDir();
bld.add("Partition " + replica.topic() + "-" + replica.partition() + " on broker " +
replica.brokerId() + " is not being moved from log dir " + currentLogDir + " to " +
targetLogDir + ".");
@@ -1292,7 +1292,7 @@ public class ReassignPartitionsCommand {
Map<TopicPartitionReplica, String> curMovingParts = new HashMap<>();
findLogDirMoveStates(adminClient, targetReplicas).forEach((part, moveState) -> {
if (moveState instanceof ActiveMoveState)
- curMovingParts.put(part, ((ActiveMoveState) moveState).currentLogDir);
+ curMovingParts.put(part, ((ActiveMoveState) moveState).currentLogDir());
});
if (curMovingParts.isEmpty()) {
System.out.print("None of the specified partition moves are active.");
diff --git a/tools/src/test/java/org/apache/kafka/tools/ConnectPluginPathTest.java b/tools/src/test/java/org/apache/kafka/tools/ConnectPluginPathTest.java
index 2b8666d57bc..d10cf7b2e45 100644
--- a/tools/src/test/java/org/apache/kafka/tools/ConnectPluginPathTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/ConnectPluginPathTest.java
@@ -444,13 +444,7 @@ public class ConnectPluginPathTest {
MULTI_JAR
}
- private static class PluginLocation {
- private final Path path;
-
- private PluginLocation(Path path) {
- this.path = path;
- }
-
+ private record PluginLocation(Path path) {
@Override
public String toString() {
return path.toString();
@@ -504,15 +498,7 @@ public class ConnectPluginPathTest {
}
}
- private static class PluginPathElement {
- private final Path root;
- private final List<PluginLocation> locations;
-
- private PluginPathElement(Path root, List<PluginLocation> locations) {
- this.root = root;
- this.locations = locations;
- }
-
+ private record PluginPathElement(Path root, List<PluginLocation> locations) {
@Override
public String toString() {
return root.toString();
@@ -535,14 +521,7 @@ public class ConnectPluginPathTest {
return new PluginPathElement(path, locations);
}
- private static class WorkerConfig {
- private final Path configFile;
- private final List<PluginPathElement> pluginPathElements;
-
- private WorkerConfig(Path configFile, List<PluginPathElement> pluginPathElements) {
- this.configFile = configFile;
- this.pluginPathElements = pluginPathElements;
- }
+ private record WorkerConfig(Path configFile, List<PluginPathElement> pluginPathElements) {
@Override
public String toString() {
diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupServiceTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupServiceTest.java
index d5c1b035a44..236d1ce51cc 100644
--- a/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupServiceTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupServiceTest.java
@@ -183,13 +183,13 @@ public class ConsumerGroupServiceTest {
Map<TopicPartition, Optional<Long>> returnedOffsets = assignments.map(results ->
results.stream().collect(Collectors.toMap(
- assignment -> new TopicPartition(assignment.topic.get(), assignment.partition.get()),
- assignment -> assignment.offset))
+ assignment -> new TopicPartition(assignment.topic().get(), assignment.partition().get()),
+ assignment -> assignment.offset()))
).orElse(Map.of());
Map<TopicPartition, Optional<Integer>> returnedLeaderEpoch = assignments.map(results ->
results.stream().collect(Collectors.toMap(
- assignment -> new TopicPartition(assignment.topic.get(), assignment.partition.get()),
- assignment -> assignment.leaderEpoch))
+ assignment -> new TopicPartition(assignment.topic().get(), assignment.partition().get()),
+ assignment -> assignment.leaderEpoch()))
).orElse(Map.of());
Map<TopicPartition, Optional<Long>> expectedOffsets = Map.of(
diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java
index d30c8081440..2866027e232 100644
--- a/tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java
@@ -322,7 +322,7 @@ public class DeleteConsumerGroupsTest {
}
private boolean checkGroupState(ConsumerGroupCommand.ConsumerGroupService service, String groupId, GroupState state) throws Exception {
- return Objects.equals(service.collectGroupState(groupId).groupState, state);
+ return Objects.equals(service.collectGroupState(groupId).groupState(), state);
}
private ConsumerGroupCommand.ConsumerGroupService getConsumerGroupService(String[] args) {
diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java
index b15b4fe9d45..9e5072576f4 100644
--- a/tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java
@@ -462,7 +462,7 @@ public class DescribeConsumerGroupTest {
Optional<GroupState> state = groupOffsets.getKey();
Optional<Collection<PartitionAssignmentState>> assignments = groupOffsets.getValue();
- Predicate<PartitionAssignmentState> isGrp = s -> Objects.equals(s.group, group);
+ Predicate<PartitionAssignmentState> isGrp = s -> Objects.equals(s.group(), group);
boolean res = state.map(s -> s.equals(GroupState.STABLE)).orElse(false) &&
assignments.isPresent() &&
@@ -477,9 +477,9 @@ public class DescribeConsumerGroupTest {
PartitionAssignmentState partitionState = maybePartitionState.get();
- return !partitionState.consumerId.map(s0 -> s0.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) &&
- !partitionState.clientId.map(s0 -> s0.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) &&
- !partitionState.host.map(h -> h.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false);
+ return !partitionState.consumerId().map(s0 -> s0.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) &&
+ !partitionState.clientId().map(s0 -> s0.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) &&
+ !partitionState.host().map(h -> h.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false);
}, "Expected a 'Stable' group status, rows and valid values for consumer id / client id / host columns in describe results for group " + group + ".");
}
}
@@ -506,7 +506,7 @@ public class DescribeConsumerGroupTest {
Entry<Optional<GroupState>, Optional<Collection<MemberAssignmentState>>> res = service.collectGroupMembers(group);
assertTrue(res.getValue().isPresent());
- assertTrue(res.getValue().get().size() == 1 && res.getValue().get().iterator().next().assignment.size() == 1,
+ assertTrue(res.getValue().get().size() == 1 && res.getValue().get().iterator().next().assignment().size() == 1,
"Expected a topic partition assigned to the single group member for group " + group);
}
}
@@ -526,10 +526,10 @@ public class DescribeConsumerGroupTest {
) {
TestUtils.waitForCondition(() -> {
GroupInformation state = service.collectGroupState(group);
- return Objects.equals(state.groupState, GroupState.STABLE) &&
- state.numMembers == 1 &&
- state.coordinator != null &&
- clusterInstance.brokerIds().contains(state.coordinator.id());
+ return Objects.equals(state.groupState(), GroupState.STABLE) &&
+ state.numMembers() == 1 &&
+ state.coordinator() != null &&
+ clusterInstance.brokerIds().contains(state.coordinator().id());
}, "Expected a 'Stable' group status, with one member for group " + group + ".");
}
}
@@ -558,11 +558,11 @@ public class DescribeConsumerGroupTest {
try (ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})) {
TestUtils.waitForCondition(() -> {
GroupInformation state = service.collectGroupState(group);
- return Objects.equals(state.groupState, GroupState.STABLE) &&
- state.numMembers == 1 &&
- Objects.equals(state.assignmentStrategy, expectedName) &&
- state.coordinator != null &&
- clusterInstance.brokerIds().contains(state.coordinator.id());
+ return Objects.equals(state.groupState(), GroupState.STABLE) &&
+ state.numMembers() == 1 &&
+ Objects.equals(state.assignmentStrategy(), expectedName) &&
+ state.coordinator() != null &&
+ clusterInstance.brokerIds().contains(state.coordinator().id());
}, "Expected a 'Stable' group status, with one member and " + expectedName + " assignment strategy for group " + group + ".");
}
} finally {
@@ -619,7 +619,7 @@ public class DescribeConsumerGroupTest {
TestUtils.waitForCondition(() -> {
Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> res = service.collectGroupOffsets(group);
return res.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false)
- && res.getValue().map(c -> c.stream().anyMatch(assignment -> Objects.equals(assignment.group, group) && assignment.offset.isPresent())).orElse(false);
+ && res.getValue().map(c -> c.stream().anyMatch(assignment -> Objects.equals(assignment.group(), group) && assignment.offset().isPresent())).orElse(false);
}, "Expected the group to initially become stable, and to find group in assignments after initial offset commit.");
// stop the consumer so the group has no active member anymore
@@ -629,13 +629,13 @@ public class DescribeConsumerGroupTest {
Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> offsets = service.collectGroupOffsets(group);
Optional<GroupState> state = offsets.getKey();
Optional<Collection<PartitionAssignmentState>> assignments = offsets.getValue();
- List<PartitionAssignmentState> testGroupAssignments = assignments.get().stream().filter(a -> Objects.equals(a.group, group)).toList();
+ List<PartitionAssignmentState> testGroupAssignments = assignments.get().stream().filter(a -> Objects.equals(a.group(), group)).toList();
PartitionAssignmentState assignment = testGroupAssignments.get(0);
return state.map(s -> s.equals(GroupState.EMPTY)).orElse(false) &&
testGroupAssignments.size() == 1 &&
- assignment.consumerId.map(c -> c.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) && // the member should be gone
- assignment.clientId.map(c -> c.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) &&
- assignment.host.map(c -> c.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false);
+ assignment.consumerId().map(c -> c.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) && // the member should be gone
+ assignment.clientId().map(c -> c.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) &&
+ assignment.host().map(c -> c.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false);
}, "failed to collect group offsets");
}
}
@@ -656,7 +656,7 @@ public class DescribeConsumerGroupTest {
TestUtils.waitForCondition(() -> {
Entry<Optional<GroupState>, Optional<Collection<MemberAssignmentState>>> res = service.collectGroupMembers(group);
return res.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false)
- && res.getValue().map(c -> c.stream().anyMatch(m -> Objects.equals(m.group, group))).orElse(false);
+ && res.getValue().map(c -> c.stream().anyMatch(m -> Objects.equals(m.group(), group))).orElse(false);
}, "Expected the group to initially become stable, and to find group in assignments after initial offset commit.");
// stop the consumer so the group has no active member anymore
@@ -684,10 +684,10 @@ public class DescribeConsumerGroupTest {
) {
TestUtils.waitForCondition(() -> {
GroupInformation state = service.collectGroupState(group);
- return Objects.equals(state.groupState, GroupState.STABLE) &&
- state.numMembers == 1 &&
- state.coordinator != null &&
- clusterInstance.brokerIds().contains(state.coordinator.id());
+ return Objects.equals(state.groupState(), GroupState.STABLE) &&
+ state.numMembers() == 1 &&
+ state.coordinator() != null &&
+ clusterInstance.brokerIds().contains(state.coordinator().id());
}, "Expected the group to initially become stable, and have a single member.");
// stop the consumer so the group has no active member anymore
@@ -695,7 +695,7 @@ public class DescribeConsumerGroupTest {
TestUtils.waitForCondition(() -> {
GroupInformation state = service.collectGroupState(group);
- return Objects.equals(state.groupState, GroupState.EMPTY) && state.numMembers == 0;
+ return Objects.equals(state.groupState(), GroupState.EMPTY) && state.numMembers() == 0;
}, "Expected the group to become empty after the only member leaving.");
}
}
@@ -744,8 +744,8 @@ public class DescribeConsumerGroupTest {
Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> res = service.collectGroupOffsets(group);
return res.getKey().map(s -> s.equals(GroupState.STABLE)).isPresent() &&
res.getValue().isPresent() &&
- res.getValue().get().stream().filter(s -> Objects.equals(s.group, group)).count() == 1 &&
- res.getValue().get().stream().filter(x -> Objects.equals(x.group, group) && x.partition.isPresent()).count() == 1;
+ res.getValue().get().stream().filter(s -> Objects.equals(s.group(), group)).count() == 1 &&
+ res.getValue().get().stream().filter(x -> Objects.equals(x.group(), group) && x.partition().isPresent()).count() == 1;
}, "Expected rows for consumers with no assigned partitions in describe group results");
}
}
@@ -767,15 +767,15 @@ public class DescribeConsumerGroupTest {
Entry<Optional<GroupState>, Optional<Collection<MemberAssignmentState>>> res = service.collectGroupMembers(group);
return res.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false) &&
res.getValue().isPresent() &&
- res.getValue().get().stream().filter(s -> Objects.equals(s.group, group)).count() == 2 &&
- res.getValue().get().stream().filter(x -> Objects.equals(x.group, group) && x.numPartitions == 1).count() == 1 &&
- res.getValue().get().stream().filter(x -> Objects.equals(x.group, group) && x.numPartitions == 0).count() == 1 &&
- res.getValue().get().stream().anyMatch(s -> !s.assignment.isEmpty());
+ res.getValue().get().stream().filter(s -> Objects.equals(s.group(), group)).count() == 2 &&
+ res.getValue().get().stream().filter(x -> Objects.equals(x.group(), group) && x.numPartitions() == 1).count() == 1 &&
+ res.getValue().get().stream().filter(x -> Objects.equals(x.group(), group) && x.numPartitions() == 0).count() == 1 &&
+ res.getValue().get().stream().anyMatch(s -> !s.assignment().isEmpty());
}, "Expected rows for consumers with no assigned partitions in describe group results");
Entry<Optional<GroupState>, Optional<Collection<MemberAssignmentState>>> res = service.collectGroupMembers(group);
assertTrue(res.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false)
- && res.getValue().map(c -> c.stream().anyMatch(s -> !s.assignment.isEmpty())).orElse(false),
+ && res.getValue().map(c -> c.stream().anyMatch(s -> !s.assignment().isEmpty())).orElse(false),
"Expected additional columns in verbose version of describe members");
}
}
@@ -795,7 +795,7 @@ public class DescribeConsumerGroupTest {
) {
TestUtils.waitForCondition(() -> {
GroupInformation state = service.collectGroupState(group);
- return Objects.equals(state.groupState, GroupState.STABLE) && state.numMembers == 2;
+ return Objects.equals(state.groupState(), GroupState.STABLE) && state.numMembers() == 2;
}, "Expected two consumers in describe group results");
}
}
@@ -844,9 +844,9 @@ public class DescribeConsumerGroupTest {
Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> res = service.collectGroupOffsets(group);
return res.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false) &&
res.getValue().isPresent() &&
- res.getValue().get().stream().filter(s -> Objects.equals(s.group, group)).count() == 2 &&
- res.getValue().get().stream().filter(x -> Objects.equals(x.group, group) && x.partition.isPresent()).count() == 2 &&
- res.getValue().get().stream().noneMatch(x -> Objects.equals(x.group, group) && x.partition.isEmpty());
+ res.getValue().get().stream().filter(s -> Objects.equals(s.group(), group)).count() == 2 &&
+ res.getValue().get().stream().filter(x -> Objects.equals(x.group(), group) && x.partition().isPresent()).count() == 2 &&
+ res.getValue().get().stream().noneMatch(x -> Objects.equals(x.group(), group) && x.partition().isEmpty());
}, "Expected two rows (one row per consumer) in describe group results.");
}
}
@@ -868,13 +868,13 @@ public class DescribeConsumerGroupTest {
Entry<Optional<GroupState>, Optional<Collection<MemberAssignmentState>>> res = service.collectGroupMembers(group);
return res.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false) &&
res.getValue().isPresent() &&
- res.getValue().get().stream().filter(s -> Objects.equals(s.group, group)).count() == 2 &&
- res.getValue().get().stream().filter(x -> Objects.equals(x.group, group) && x.numPartitions == 1).count() == 2 &&
- res.getValue().get().stream().noneMatch(x -> Objects.equals(x.group, group) && x.numPartitions == 0);
+ res.getValue().get().stream().filter(s -> Objects.equals(s.group(), group)).count() == 2 &&
+ res.getValue().get().stream().filter(x -> Objects.equals(x.group(), group) && x.numPartitions() == 1).count() == 2 &&
+ res.getValue().get().stream().noneMatch(x -> Objects.equals(x.group(), group) && x.numPartitions() == 0);
}, "Expected two rows (one row per consumer) in describe group members results.");
Entry<Optional<GroupState>, Optional<Collection<MemberAssignmentState>>> res = service.collectGroupMembers(group);
- assertTrue(res.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false) && res.getValue().map(s -> s.stream().filter(x -> x.assignment.isEmpty()).count()).orElse(0L) == 0,
+ assertTrue(res.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false) && res.getValue().map(s -> s.stream().filter(x -> x.assignment().isEmpty()).count()).orElse(0L) == 0,
"Expected additional columns in verbose version of describe members");
}
}
@@ -894,7 +894,7 @@ public class DescribeConsumerGroupTest {
) {
TestUtils.waitForCondition(() -> {
GroupInformation state = service.collectGroupState(group);
- return Objects.equals(state.groupState, GroupState.STABLE) && Objects.equals(state.group, group) && state.numMembers == 2;
+ return Objects.equals(state.groupState(), GroupState.STABLE) && Objects.equals(state.group(), group) && state.numMembers() == 2;
}, "Expected a stable group with two members in describe group state result.");
}
}
@@ -915,7 +915,7 @@ public class DescribeConsumerGroupTest {
TestUtils.waitForCondition(() -> {
Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> res = service.collectGroupOffsets(group);
return res.getKey().map(s -> s.equals(GroupState.EMPTY)).orElse(false)
- && res.getValue().isPresent() && res.getValue().get().stream().filter(s -> Objects.equals(s.group, group)).count() == 2;
+ && res.getValue().isPresent() && res.getValue().get().stream().filter(s -> Objects.equals(s.group(), group)).count() == 2;
}, "Expected a stable group with two members in describe group state result.");
}
}
@@ -1030,7 +1030,7 @@ public class DescribeConsumerGroupTest {
TestUtils.waitForCondition(() -> {
Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> groupOffsets = service.collectGroupOffsets(group);
- Predicate<PartitionAssignmentState> isGrp = s -> Objects.equals(s.group, group);
+ Predicate<PartitionAssignmentState> isGrp = s -> Objects.equals(s.group(), group);
boolean res = groupOffsets.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false) &&
groupOffsets.getValue().isPresent() &&
@@ -1045,9 +1045,9 @@ public class DescribeConsumerGroupTest {
PartitionAssignmentState assignmentState = maybeAssignmentState.get();
- return assignmentState.consumerId.map(c -> !c.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) &&
- assignmentState.clientId.map(c -> !c.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) &&
- assignmentState.host.map(h -> !h.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false);
+ return assignmentState.consumerId().map(c -> !c.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) &&
+ assignmentState.clientId().map(c -> !c.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) &&
+ assignmentState.host().map(h -> !h.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false);
}, "Expected a 'Stable' group status, rows and valid values for consumer id / client id / host columns in describe results for non-offset-committing group " + group + ".");
}
}
diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java
index 1c9ab9cf98b..bbcfb6e35c1 100644
--- a/tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java
@@ -896,10 +896,10 @@ public class ResetConsumerGroupOffsetTest {
private void awaitConsumerGroupInactive(ConsumerGroupCommand.ConsumerGroupService service,
String group) throws Exception {
TestUtils.waitForCondition(() -> {
- GroupState state = service.collectGroupState(group).groupState;
+ GroupState state = service.collectGroupState(group).groupState();
return Objects.equals(state, GroupState.EMPTY) || Objects.equals(state, GroupState.DEAD);
}, "Expected that consumer group is inactive. Actual state: " +
- service.collectGroupState(group).groupState);
+ service.collectGroupState(group).groupState());
}
private void resetAndAssertOffsetsCommitted(ClusterInstance cluster,
diff --git a/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommandTest.java b/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommandTest.java
index f1726721bfd..a8d830b4bd4 100644
--- a/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommandTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommandTest.java
@@ -646,16 +646,7 @@ public class ReassignPartitionsCommandTest {
}));
}
- static class LogDirReassignment {
- final String json;
- final String currentDir;
- final String targetDir;
-
- public LogDirReassignment(String json, String currentDir, String targetDir) {
- this.json = json;
- this.currentDir = currentDir;
- this.targetDir = targetDir;
- }
+ record LogDirReassignment(String json, String currentDir, String targetDir) {
}
private LogDirReassignment buildLogDirReassignment(TopicPartition topicPartition,