chia7712 commented on code in PR #17626:
URL: https://github.com/apache/kafka/pull/17626#discussion_r1823307470
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -3490,6 +3490,138 @@ void handleFailure(Throwable throwable) {
return new DescribeDelegationTokenResult(tokensFuture);
}
+ private static final class ListGroupsResults {
+ private final List<Throwable> errors;
+ private final HashMap<String, GroupListing> listings;
+ private final HashSet<Node> remaining;
+ private final KafkaFutureImpl<Collection<Object>> future;
+
+ ListGroupsResults(Collection<Node> leaders,
+ KafkaFutureImpl<Collection<Object>> future) {
+ this.errors = new ArrayList<>();
+ this.listings = new HashMap<>();
+ this.remaining = new HashSet<>(leaders);
+ this.future = future;
+ tryComplete();
+ }
+
+ synchronized void addError(Throwable throwable, Node node) {
+ ApiError error = ApiError.fromThrowable(throwable);
+ if (error.message() == null || error.message().isEmpty()) {
+ errors.add(error.error().exception("Error listing groups on "
+ node));
+ } else {
+ errors.add(error.error().exception("Error listing groups on "
+ node + ": " + error.message()));
+ }
+ }
+
+ synchronized void addListing(GroupListing listing) {
+ listings.put(listing.groupId(), listing);
+ }
+
+ synchronized void tryComplete(Node leader) {
+ remaining.remove(leader);
+ tryComplete();
+ }
+
+ private synchronized void tryComplete() {
+ if (remaining.isEmpty()) {
+ ArrayList<Object> results = new ArrayList<>(listings.values());
+ results.addAll(errors);
+ future.complete(results);
+ }
+ }
+ }
+
+ @Override
+ public ListGroupsResult listGroups(ListGroupsOptions options) {
+ final KafkaFutureImpl<Collection<Object>> all = new
KafkaFutureImpl<>();
+ final long nowMetadata = time.milliseconds();
+ final long deadline = calcDeadlineMs(nowMetadata, options.timeoutMs());
+ runnable.call(new Call("findAllBrokers", deadline, new
LeastLoadedNodeProvider()) {
+ @Override
+ MetadataRequest.Builder createRequest(int timeoutMs) {
+ return new MetadataRequest.Builder(new MetadataRequestData()
+ .setTopics(Collections.emptyList())
+ .setAllowAutoTopicCreation(true));
+ }
+
+ @Override
+ void handleResponse(AbstractResponse abstractResponse) {
+ MetadataResponse metadataResponse = (MetadataResponse)
abstractResponse;
+ Collection<Node> nodes = metadataResponse.brokers();
+ if (nodes.isEmpty())
+ throw new StaleMetadataException("Metadata fetch failed
due to missing broker list");
+
+ HashSet<Node> allNodes = new HashSet<>(nodes);
+ final ListGroupsResults results = new
ListGroupsResults(allNodes, all);
+
+ for (final Node node : allNodes) {
+ final long nowList = time.milliseconds();
+ runnable.call(new Call("listGroups", deadline, new
ConstantNodeIdProvider(node.id())) {
+ @Override
+ ListGroupsRequest.Builder createRequest(int timeoutMs)
{
+ List<String> groupTypes = options.types()
+ .stream()
+ .map(GroupType::toString)
+ .collect(Collectors.toList());
+ return new ListGroupsRequest.Builder(new
ListGroupsRequestData()
+ .setTypesFilter(groupTypes)
+ );
+ }
+
+ private void
maybeAddGroup(ListGroupsResponseData.ListedGroup group) {
+ final String groupId = group.groupId();
+ final GroupType type = group.groupType().isEmpty()
+ ? GroupType.UNKNOWN
+ : GroupType.parse(group.groupType());
Review Comment:
`GroupType.parse` is able to handle both null and empty string, so we don't
need to check the `groupType` here.
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -3490,6 +3490,138 @@ void handleFailure(Throwable throwable) {
return new DescribeDelegationTokenResult(tokensFuture);
}
+ private static final class ListGroupsResults {
+ private final List<Throwable> errors;
+ private final HashMap<String, GroupListing> listings;
+ private final HashSet<Node> remaining;
+ private final KafkaFutureImpl<Collection<Object>> future;
+
+ ListGroupsResults(Collection<Node> leaders,
+ KafkaFutureImpl<Collection<Object>> future) {
+ this.errors = new ArrayList<>();
+ this.listings = new HashMap<>();
+ this.remaining = new HashSet<>(leaders);
+ this.future = future;
+ tryComplete();
+ }
+
+ synchronized void addError(Throwable throwable, Node node) {
+ ApiError error = ApiError.fromThrowable(throwable);
+ if (error.message() == null || error.message().isEmpty()) {
+ errors.add(error.error().exception("Error listing groups on "
+ node));
+ } else {
+ errors.add(error.error().exception("Error listing groups on "
+ node + ": " + error.message()));
+ }
+ }
+
+ synchronized void addListing(GroupListing listing) {
+ listings.put(listing.groupId(), listing);
+ }
+
+ synchronized void tryComplete(Node leader) {
+ remaining.remove(leader);
+ tryComplete();
+ }
+
+ private synchronized void tryComplete() {
+ if (remaining.isEmpty()) {
+ ArrayList<Object> results = new ArrayList<>(listings.values());
+ results.addAll(errors);
+ future.complete(results);
+ }
+ }
+ }
+
+ @Override
+ public ListGroupsResult listGroups(ListGroupsOptions options) {
+ final KafkaFutureImpl<Collection<Object>> all = new
KafkaFutureImpl<>();
+ final long nowMetadata = time.milliseconds();
+ final long deadline = calcDeadlineMs(nowMetadata, options.timeoutMs());
+ runnable.call(new Call("findAllBrokers", deadline, new
LeastLoadedNodeProvider()) {
+ @Override
+ MetadataRequest.Builder createRequest(int timeoutMs) {
+ return new MetadataRequest.Builder(new MetadataRequestData()
+ .setTopics(Collections.emptyList())
+ .setAllowAutoTopicCreation(true));
+ }
+
+ @Override
+ void handleResponse(AbstractResponse abstractResponse) {
+ MetadataResponse metadataResponse = (MetadataResponse)
abstractResponse;
+ Collection<Node> nodes = metadataResponse.brokers();
+ if (nodes.isEmpty())
+ throw new StaleMetadataException("Metadata fetch failed
due to missing broker list");
+
+ HashSet<Node> allNodes = new HashSet<>(nodes);
+ final ListGroupsResults results = new
ListGroupsResults(allNodes, all);
+
+ for (final Node node : allNodes) {
+ final long nowList = time.milliseconds();
+ runnable.call(new Call("listGroups", deadline, new
ConstantNodeIdProvider(node.id())) {
+ @Override
+ ListGroupsRequest.Builder createRequest(int timeoutMs)
{
+ List<String> groupTypes = options.types()
+ .stream()
+ .map(GroupType::toString)
+ .collect(Collectors.toList());
+ return new ListGroupsRequest.Builder(new
ListGroupsRequestData()
+ .setTypesFilter(groupTypes)
+ );
+ }
+
+ private void
maybeAddGroup(ListGroupsResponseData.ListedGroup group) {
+ final String groupId = group.groupId();
+ final GroupType type = group.groupType().isEmpty()
+ ? GroupType.UNKNOWN
+ : GroupType.parse(group.groupType());
+ final String protocolType = group.protocolType();
+ final GroupListing groupListing = new GroupListing(
+ groupId,
+ Optional.of(type),
Review Comment:
It seems that `type` is never null, so why do we need to wrap it in
`Optional`?
##########
clients/src/main/java/org/apache/kafka/clients/admin/GroupListing.java:
##########
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.GroupType;
+
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * A listing of a group in the cluster.
+ */
+public class GroupListing {
+ private final String groupId;
+ private final Optional<GroupType> type;
+ private final String protocol;
+
+ /**
+ * Create an instance with the specified parameters.
+ *
+ * @param groupId Group Id
+ * @param type Group type
+ * @param protocol Protocol
+ */
+ public GroupListing(String groupId, Optional<GroupType> type, String
protocol) {
+ this.groupId = groupId;
+ this.type = Objects.requireNonNull(type);
+ this.protocol = protocol;
+ }
+
+ /**
+ * The group Id.
+ *
+ * @return Group Id
+ */
+ public String groupId() {
+ return groupId;
+ }
+
+ /**
+ * The type of the group.
+ *
+ * @return An Optional containing the type, if available
+ */
+ public Optional<GroupType> type() {
+ return type;
+ }
+
+ /**
+ * The protocol of the group.
+ *
+ * @return The protocol
+ */
+ public String protocol() {
+ return protocol;
+ }
+
+ /**
+ * If the group is a simple consumer group or not.
+ */
+ public boolean isSimpleConsumerGroup() {
+ return type.filter(gt -> gt == GroupType.CLASSIC).isPresent() &&
protocol.isEmpty();
+ }
+
+ @Override
+ public String toString() {
+ return "(" +
+ "groupId='" + groupId + '\'' +
+ ", type=" + type +
+ ", protocol=" + protocol +
+ ')';
+ }
+
+ protected String toStringBase() {
+ return "groupId='" + groupId + '\'' +
+ ", type=" + type +
+ ", protocol=" + protocol;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(groupId, type);
Review Comment:
Should it include `protocol`?
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListGroupsOptions.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.GroupType;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Options for {@link Admin#listGroups()}.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListGroupsOptions extends AbstractOptions<ListGroupsOptions> {
+
+ private Set<GroupType> types = Collections.emptySet();
+
+ /**
+ * If types is set, only groups of these types will be returned by
listGroups().
+ * Otherwise, all groups are returned.
+ */
+ public ListGroupsOptions withTypes(Set<GroupType> types) {
+ this.types = (types == null || types.isEmpty()) ?
Collections.emptySet() : new HashSet<>(types);
Review Comment:
If there is no specific reason, keeping it immutable is a better way.
```java
this.types = (types == null || types.isEmpty()) ? Set.of() :
Set.copyOf(types);
```
##########
tools/src/main/java/org/apache/kafka/tools/GroupsCommand.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.GroupListing;
+import org.apache.kafka.common.GroupType;
+import org.apache.kafka.common.utils.Exit;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.CommandDefaultOptions;
+import org.apache.kafka.server.util.CommandLineUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import joptsimple.ArgumentAcceptingOptionSpec;
+import joptsimple.OptionSpec;
+import joptsimple.OptionSpecBuilder;
+
+public class GroupsCommand {
+ private static final Logger LOG =
LoggerFactory.getLogger(GroupsCommand.class);
+
+ public static void main(String... args) {
+ Exit.exit(mainNoExit(args));
+ }
+
+ static int mainNoExit(String... args) {
+ try {
+ execute(args);
+ return 0;
+ } catch (Throwable e) {
+ System.err.println(e.getMessage());
+ System.err.println(Utils.stackTrace(e));
+ return 1;
+ }
+ }
+
+ static void execute(String... args) throws Exception {
+ GroupsCommandOptions opts = new GroupsCommandOptions(args);
+
+ Properties config = opts.commandConfig();
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
opts.bootstrapServer());
+
+ int exitCode = 0;
+ try (GroupsService service = new GroupsService(config)) {
+ if (opts.hasListOption()) {
+ service.listGroups(opts);
+ }
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause != null) {
+ printException(cause);
+ } else {
+ printException(e);
+ }
+ exitCode = 1;
+ } catch (Throwable t) {
+ printException(t);
+ exitCode = 1;
+ } finally {
+ Exit.exit(exitCode);
+ }
+ }
+
+ public static class GroupsService implements AutoCloseable {
+ private final Admin adminClient;
+
+ public GroupsService(Properties config) {
+ this.adminClient = Admin.create(config);
+ }
+
+ // Visible for testing
+ GroupsService(Admin adminClient) {
+ this.adminClient = adminClient;
+ }
+
+ public void listGroups(GroupsCommandOptions opts) throws Exception {
+ Collection<GroupListing> resources = adminClient.listGroups()
+ .all().get(30, TimeUnit.SECONDS);
+ printGroupDetails(resources, opts.groupType(), opts.protocol(),
opts.hasConsumerOption(), opts.hasShareOption());
+ }
+
+ private void printGroupDetails(Collection<GroupListing> groups,
+ Optional<GroupType> groupTypeFilter,
+ Optional<String> protocolFilter,
+ boolean consumerGroupFilter,
+ boolean shareGroupFilter) {
+ List<List<String>> lineItems = new ArrayList<>();
+ int maxLen = 20;
+ for (GroupListing group : groups) {
+ if (combinedFilter(group, groupTypeFilter, protocolFilter,
consumerGroupFilter, shareGroupFilter)) {
+ List<String> lineItem = new ArrayList<>();
+ lineItem.add(group.groupId());
+
lineItem.add(group.type().map(GroupType::toString).orElse(""));
+ lineItem.add(group.protocol());
+ for (String item : lineItem) {
+ if (item != null) {
+ maxLen = Math.max(maxLen, item.length());
+ }
+ }
+ lineItems.add(lineItem);
+ }
+ }
+
+ String fmt = "%" + (-maxLen) + "s";
+ String header = fmt + " " + fmt + " " + fmt;
+ System.out.printf(header, "GROUP", "TYPE", "PROTOCOL");
+ System.out.println();
+ for (List<String> item : lineItems) {
+ for (String atom : item) {
+ System.out.printf(fmt + " ", atom);
+ }
+ System.out.println();
+ }
+ }
+
+ private boolean combinedFilter(GroupListing group,
+ Optional<GroupType> groupTypeFilter,
+ Optional<String> protocolFilter,
+ boolean consumerGroupFilter,
+ boolean shareGroupFilter) {
+ boolean pass = true;
+ Optional<GroupType> groupType = group.type();
+ String protocol = group.protocol();
+
+ if (groupTypeFilter.isPresent()) {
+ pass = protocolFilter.map(s -> protocol.equals(s) &&
groupType.filter(gt -> gt == groupTypeFilter.get()).isPresent())
+ .orElseGet(() -> groupType.filter(gt -> gt ==
groupTypeFilter.get()).isPresent());
+ } else if (protocolFilter.isPresent()) {
+ pass = protocol.equals(protocolFilter.get());
+ } else if (consumerGroupFilter) {
+ pass = protocol.equals("consumer") || protocol.isEmpty() ||
groupType.filter(gt -> gt == GroupType.CONSUMER).isPresent();
+ } else if (shareGroupFilter) {
+ pass = groupType.filter(gt -> gt ==
GroupType.SHARE).isPresent();
+ }
+ return pass;
+ }
+
+ @Override
+ public void close() throws Exception {
+ adminClient.close();
+ }
+ }
+
+ private static void printException(Throwable e) {
+ System.out.println("Error while executing groups command : " +
e.getMessage());
+ LOG.error(Utils.stackTrace(e));
+ }
+
+ public static final class GroupsCommandOptions extends
CommandDefaultOptions {
+ private final ArgumentAcceptingOptionSpec<String> bootstrapServerOpt;
+
+ private final ArgumentAcceptingOptionSpec<String> commandConfigOpt;
+
+ private final OptionSpecBuilder listOpt;
+
+ private final ArgumentAcceptingOptionSpec<String> groupTypeOpt;
+
+ private final ArgumentAcceptingOptionSpec<String> protocolOpt;
+
+ private final OptionSpecBuilder consumerOpt;
+
+ private final OptionSpecBuilder shareOpt;
+
+ public GroupsCommandOptions(String[] args) {
+ super(args);
+ bootstrapServerOpt = parser.accepts("bootstrap-server", "REQUIRED:
The Kafka server to connect to.")
+ .withRequiredArg()
+ .describedAs("server to connect to")
+ .ofType(String.class);
+ commandConfigOpt = parser.accepts("command-config", "Property file
containing configs to be passed to Admin Client.")
+ .withRequiredArg()
+ .describedAs("command config property file")
+ .ofType(String.class);
+
+ listOpt = parser.accepts("list", "List the groups.");
+
+ groupTypeOpt = parser.accepts("group-type", "Filter the groups
based on group type. "
+ + "Valid types are: 'classic', 'consumer' and
'share'.")
+ .withRequiredArg()
+ .describedAs("type")
+ .ofType(String.class);
+
+ protocolOpt = parser.accepts("protocol", "Filter the groups based
on protocol type.")
+ .withRequiredArg()
+ .describedAs("protocol")
+ .ofType(String.class);
+
+ consumerOpt = parser.accepts("consumer", "Filter the groups to
show all kinds of consumer groups, including classic and simple consumer
groups. "
+ + "This matches group type 'consumer', and group
type 'classic' where the protocol type is 'consumer' or empty.");
+ shareOpt = parser.accepts("share", "Filter the groups to show
share groups.");
+
+ options = parser.parse(args);
+
+ checkArgs();
+ }
+
+ public Boolean has(OptionSpec<?> builder) {
+ return options.has(builder);
+ }
+
+ public <A> Optional<A> valueAsOption(OptionSpec<A> option) {
+ return valueAsOption(option, Optional.empty());
+ }
+
+ public <A> Optional<A> valueAsOption(OptionSpec<A> option, Optional<A>
defaultValue) {
+ if (has(option)) {
+ return Optional.of(options.valueOf(option));
+ } else {
+ return defaultValue;
+ }
+ }
+
+ public String bootstrapServer() {
+ return options.valueOf(bootstrapServerOpt);
+ }
+
+ public Properties commandConfig() throws IOException {
+ if (has(commandConfigOpt)) {
+ return Utils.loadProps(options.valueOf(commandConfigOpt));
+ } else {
+ return new Properties();
+ }
+ }
+
+ public Optional<GroupType> groupType() {
+ return valueAsOption(groupTypeOpt).map(GroupType::parse).filter(gt
-> gt != GroupType.UNKNOWN);
+ }
+
+ public Optional<String> protocol() {
+ return valueAsOption(protocolOpt);
+ }
+
+ public boolean hasConsumerOption() {
+ return has(consumerOpt);
+ }
+
+ public boolean hasListOption() {
+ return has(listOpt);
+ }
+
+ public boolean hasShareOption() {
+ return has(shareOpt);
+ }
+
+ public void checkArgs() {
+ if (args.length == 0)
+ CommandLineUtils.printUsageAndExit(parser, "This tool helps to
list groups of all types.");
+
+ CommandLineUtils.maybePrintHelpOrVersion(this, "This tool helps to
list groups of all types.");
+
+ // should have exactly one action
+ long actions = Stream.of(listOpt).filter(options::has).count();
+ if (actions != 1)
+ CommandLineUtils.printUsageAndExit(parser, "Command must
include exactly one action: --list.");
+
+ // check required args
+ if (!has(bootstrapServerOpt))
Review Comment:
Maybe this check can be replaced by `.required()`
```java
bootstrapServerOpt = parser.accepts("bootstrap-server",
"REQUIRED: The Kafka server to connect to.")
.withRequiredArg()
.describedAs("server to connect to")
.required()
.ofType(String.class);
```
##########
tools/src/main/java/org/apache/kafka/tools/GroupsCommand.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.GroupListing;
+import org.apache.kafka.common.GroupType;
+import org.apache.kafka.common.utils.Exit;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.CommandDefaultOptions;
+import org.apache.kafka.server.util.CommandLineUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import joptsimple.ArgumentAcceptingOptionSpec;
+import joptsimple.OptionSpec;
+import joptsimple.OptionSpecBuilder;
+
+public class GroupsCommand {
+ private static final Logger LOG =
LoggerFactory.getLogger(GroupsCommand.class);
+
+ public static void main(String... args) {
+ Exit.exit(mainNoExit(args));
+ }
+
+ static int mainNoExit(String... args) {
+ try {
+ execute(args);
+ return 0;
+ } catch (Throwable e) {
+ System.err.println(e.getMessage());
+ System.err.println(Utils.stackTrace(e));
+ return 1;
+ }
+ }
+
+ static void execute(String... args) throws Exception {
+ GroupsCommandOptions opts = new GroupsCommandOptions(args);
+
+ Properties config = opts.commandConfig();
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
opts.bootstrapServer());
+
+ int exitCode = 0;
+ try (GroupsService service = new GroupsService(config)) {
+ if (opts.hasListOption()) {
+ service.listGroups(opts);
+ }
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause != null) {
+ printException(cause);
+ } else {
+ printException(e);
+ }
+ exitCode = 1;
+ } catch (Throwable t) {
+ printException(t);
+ exitCode = 1;
+ } finally {
+ Exit.exit(exitCode);
+ }
+ }
+
+ public static class GroupsService implements AutoCloseable {
+ private final Admin adminClient;
+
+ public GroupsService(Properties config) {
+ this.adminClient = Admin.create(config);
+ }
+
+ // Visible for testing
+ GroupsService(Admin adminClient) {
+ this.adminClient = adminClient;
+ }
+
+ public void listGroups(GroupsCommandOptions opts) throws Exception {
+ Collection<GroupListing> resources = adminClient.listGroups()
+ .all().get(30, TimeUnit.SECONDS);
+ printGroupDetails(resources, opts.groupType(), opts.protocol(),
opts.hasConsumerOption(), opts.hasShareOption());
+ }
+
+ private void printGroupDetails(Collection<GroupListing> groups,
+ Optional<GroupType> groupTypeFilter,
+ Optional<String> protocolFilter,
+ boolean consumerGroupFilter,
+ boolean shareGroupFilter) {
+ List<List<String>> lineItems = new ArrayList<>();
+ int maxLen = 20;
+ for (GroupListing group : groups) {
+ if (combinedFilter(group, groupTypeFilter, protocolFilter,
consumerGroupFilter, shareGroupFilter)) {
+ List<String> lineItem = new ArrayList<>();
+ lineItem.add(group.groupId());
+
lineItem.add(group.type().map(GroupType::toString).orElse(""));
+ lineItem.add(group.protocol());
+ for (String item : lineItem) {
+ if (item != null) {
+ maxLen = Math.max(maxLen, item.length());
+ }
+ }
+ lineItems.add(lineItem);
+ }
+ }
+
+ String fmt = "%" + (-maxLen) + "s";
+ String header = fmt + " " + fmt + " " + fmt;
+ System.out.printf(header, "GROUP", "TYPE", "PROTOCOL");
+ System.out.println();
+ for (List<String> item : lineItems) {
+ for (String atom : item) {
+ System.out.printf(fmt + " ", atom);
+ }
+ System.out.println();
+ }
+ }
+
+ private boolean combinedFilter(GroupListing group,
+ Optional<GroupType> groupTypeFilter,
+ Optional<String> protocolFilter,
+ boolean consumerGroupFilter,
+ boolean shareGroupFilter) {
+ boolean pass = true;
+ Optional<GroupType> groupType = group.type();
+ String protocol = group.protocol();
+
+ if (groupTypeFilter.isPresent()) {
+ pass = protocolFilter.map(s -> protocol.equals(s) &&
groupType.filter(gt -> gt == groupTypeFilter.get()).isPresent())
+ .orElseGet(() -> groupType.filter(gt -> gt ==
groupTypeFilter.get()).isPresent());
+ } else if (protocolFilter.isPresent()) {
+ pass = protocol.equals(protocolFilter.get());
+ } else if (consumerGroupFilter) {
+ pass = protocol.equals("consumer") || protocol.isEmpty() ||
groupType.filter(gt -> gt == GroupType.CONSUMER).isPresent();
+ } else if (shareGroupFilter) {
+ pass = groupType.filter(gt -> gt ==
GroupType.SHARE).isPresent();
+ }
+ return pass;
+ }
+
+ @Override
+ public void close() throws Exception {
+ adminClient.close();
+ }
+ }
+
+ private static void printException(Throwable e) {
+ System.out.println("Error while executing groups command : " +
e.getMessage());
+ LOG.error(Utils.stackTrace(e));
+ }
+
+ public static final class GroupsCommandOptions extends
CommandDefaultOptions {
+ private final ArgumentAcceptingOptionSpec<String> bootstrapServerOpt;
+
+ private final ArgumentAcceptingOptionSpec<String> commandConfigOpt;
+
+ private final OptionSpecBuilder listOpt;
+
+ private final ArgumentAcceptingOptionSpec<String> groupTypeOpt;
+
+ private final ArgumentAcceptingOptionSpec<String> protocolOpt;
+
+ private final OptionSpecBuilder consumerOpt;
+
+ private final OptionSpecBuilder shareOpt;
+
+ public GroupsCommandOptions(String[] args) {
+ super(args);
+ bootstrapServerOpt = parser.accepts("bootstrap-server", "REQUIRED:
The Kafka server to connect to.")
+ .withRequiredArg()
+ .describedAs("server to connect to")
+ .ofType(String.class);
+ commandConfigOpt = parser.accepts("command-config", "Property file
containing configs to be passed to Admin Client.")
+ .withRequiredArg()
+ .describedAs("command config property file")
+ .ofType(String.class);
+
+ listOpt = parser.accepts("list", "List the groups.");
+
+ groupTypeOpt = parser.accepts("group-type", "Filter the groups
based on group type. "
+ + "Valid types are: 'classic', 'consumer' and
'share'.")
+ .withRequiredArg()
+ .describedAs("type")
+ .ofType(String.class);
+
+ protocolOpt = parser.accepts("protocol", "Filter the groups based
on protocol type.")
+ .withRequiredArg()
+ .describedAs("protocol")
+ .ofType(String.class);
+
+ consumerOpt = parser.accepts("consumer", "Filter the groups to
show all kinds of consumer groups, including classic and simple consumer
groups. "
+ + "This matches group type 'consumer', and group
type 'classic' where the protocol type is 'consumer' or empty.");
+ shareOpt = parser.accepts("share", "Filter the groups to show
share groups.");
Review Comment:
Does it equal to `group-type=share`?
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -3490,6 +3490,138 @@ void handleFailure(Throwable throwable) {
return new DescribeDelegationTokenResult(tokensFuture);
}
+ private static final class ListGroupsResults {
+ private final List<Throwable> errors;
+ private final HashMap<String, GroupListing> listings;
+ private final HashSet<Node> remaining;
+ private final KafkaFutureImpl<Collection<Object>> future;
+
+ ListGroupsResults(Collection<Node> leaders,
+ KafkaFutureImpl<Collection<Object>> future) {
+ this.errors = new ArrayList<>();
+ this.listings = new HashMap<>();
+ this.remaining = new HashSet<>(leaders);
+ this.future = future;
+ tryComplete();
+ }
+
+ synchronized void addError(Throwable throwable, Node node) {
+ ApiError error = ApiError.fromThrowable(throwable);
+ if (error.message() == null || error.message().isEmpty()) {
+ errors.add(error.error().exception("Error listing groups on "
+ node));
+ } else {
+ errors.add(error.error().exception("Error listing groups on "
+ node + ": " + error.message()));
+ }
+ }
+
+ synchronized void addListing(GroupListing listing) {
+ listings.put(listing.groupId(), listing);
+ }
+
+ synchronized void tryComplete(Node leader) {
+ remaining.remove(leader);
+ tryComplete();
+ }
+
+ private synchronized void tryComplete() {
+ if (remaining.isEmpty()) {
+ ArrayList<Object> results = new ArrayList<>(listings.values());
+ results.addAll(errors);
+ future.complete(results);
+ }
+ }
+ }
+
+ @Override
+ public ListGroupsResult listGroups(ListGroupsOptions options) {
+ final KafkaFutureImpl<Collection<Object>> all = new
KafkaFutureImpl<>();
+ final long nowMetadata = time.milliseconds();
+ final long deadline = calcDeadlineMs(nowMetadata, options.timeoutMs());
+ runnable.call(new Call("findAllBrokers", deadline, new
LeastLoadedNodeProvider()) {
+ @Override
+ MetadataRequest.Builder createRequest(int timeoutMs) {
+ return new MetadataRequest.Builder(new MetadataRequestData()
+ .setTopics(Collections.emptyList())
+ .setAllowAutoTopicCreation(true));
+ }
+
+ @Override
+ void handleResponse(AbstractResponse abstractResponse) {
+ MetadataResponse metadataResponse = (MetadataResponse)
abstractResponse;
+ Collection<Node> nodes = metadataResponse.brokers();
+ if (nodes.isEmpty())
+ throw new StaleMetadataException("Metadata fetch failed
due to missing broker list");
+
+ HashSet<Node> allNodes = new HashSet<>(nodes);
+ final ListGroupsResults results = new
ListGroupsResults(allNodes, all);
+
+ for (final Node node : allNodes) {
+ final long nowList = time.milliseconds();
+ runnable.call(new Call("listGroups", deadline, new
ConstantNodeIdProvider(node.id())) {
+ @Override
+ ListGroupsRequest.Builder createRequest(int timeoutMs)
{
+ List<String> groupTypes = options.types()
+ .stream()
+ .map(GroupType::toString)
+ .collect(Collectors.toList());
+ return new ListGroupsRequest.Builder(new
ListGroupsRequestData()
+ .setTypesFilter(groupTypes)
Review Comment:
Could you clarify why StatesFilter isn't supported
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListGroupsResult.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.ArrayList;
+import java.util.Collection;
+
+/**
+ * The result of the {@link Admin#listGroups()} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListGroupsResult {
+ private final KafkaFutureImpl<Collection<GroupListing>> all;
+ private final KafkaFutureImpl<Collection<GroupListing>> valid;
+ private final KafkaFutureImpl<Collection<Throwable>> errors;
+
+ ListGroupsResult(KafkaFuture<Collection<Object>> future) {
+ this.all = new KafkaFutureImpl<>();
+ this.valid = new KafkaFutureImpl<>();
+ this.errors = new KafkaFutureImpl<>();
+ future.thenApply((KafkaFuture.BaseFunction<Collection<Object>, Void>)
results -> {
+ ArrayList<Throwable> curErrors = new ArrayList<>();
+ ArrayList<GroupListing> curValid = new ArrayList<>();
+ for (Object resultObject : results) {
+ if (resultObject instanceof Throwable) {
+ curErrors.add((Throwable) resultObject);
+ } else {
+ curValid.add((GroupListing) resultObject);
+ }
+ }
+ if (!curErrors.isEmpty()) {
+ all.completeExceptionally(curErrors.get(0));
+ } else {
+ all.complete(curValid);
+ }
+ valid.complete(curValid);
Review Comment:
Could we make those collections immutable?
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -3490,6 +3490,138 @@ void handleFailure(Throwable throwable) {
return new DescribeDelegationTokenResult(tokensFuture);
}
+ private static final class ListGroupsResults {
+ private final List<Throwable> errors;
+ private final HashMap<String, GroupListing> listings;
+ private final HashSet<Node> remaining;
+ private final KafkaFutureImpl<Collection<Object>> future;
+
+ ListGroupsResults(Collection<Node> leaders,
+ KafkaFutureImpl<Collection<Object>> future) {
+ this.errors = new ArrayList<>();
+ this.listings = new HashMap<>();
+ this.remaining = new HashSet<>(leaders);
+ this.future = future;
+ tryComplete();
+ }
+
+ synchronized void addError(Throwable throwable, Node node) {
+ ApiError error = ApiError.fromThrowable(throwable);
+ if (error.message() == null || error.message().isEmpty()) {
+ errors.add(error.error().exception("Error listing groups on "
+ node));
+ } else {
+ errors.add(error.error().exception("Error listing groups on "
+ node + ": " + error.message()));
+ }
+ }
+
+ synchronized void addListing(GroupListing listing) {
+ listings.put(listing.groupId(), listing);
Review Comment:
Why do we need to use a map? All types of groups should share the same
namespace, so we could simplify the data structure to a single `List<Object>`.
I mean, errors and listings could be merged.
##########
clients/src/main/java/org/apache/kafka/clients/admin/GroupListing.java:
##########
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.GroupType;
+
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * A listing of a group in the cluster.
+ */
+public class GroupListing {
+ private final String groupId;
+ private final Optional<GroupType> type;
+ private final String protocol;
+
+ /**
+ * Create an instance with the specified parameters.
+ *
+ * @param groupId Group Id
+ * @param type Group type
+ * @param protocol Protocol
+ */
+ public GroupListing(String groupId, Optional<GroupType> type, String
protocol) {
+ this.groupId = groupId;
+ this.type = Objects.requireNonNull(type);
+ this.protocol = protocol;
+ }
+
+ /**
+ * The group Id.
+ *
+ * @return Group Id
+ */
+ public String groupId() {
+ return groupId;
+ }
+
+ /**
+ * The type of the group.
+ *
+ * @return An Optional containing the type, if available
+ */
+ public Optional<GroupType> type() {
+ return type;
+ }
+
+ /**
+ * The protocol of the group.
+ *
+ * @return The protocol
+ */
+ public String protocol() {
+ return protocol;
+ }
+
+ /**
+ * If the group is a simple consumer group or not.
+ */
+ public boolean isSimpleConsumerGroup() {
+ return type.filter(gt -> gt == GroupType.CLASSIC).isPresent() &&
protocol.isEmpty();
+ }
+
+ @Override
+ public String toString() {
+ return "(" +
+ "groupId='" + groupId + '\'' +
+ ", type=" + type +
+ ", protocol=" + protocol +
+ ')';
+ }
+
+ protected String toStringBase() {
Review Comment:
pardon me, what is the purpose of this method?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
