AndrewJSchofield commented on code in PR #18928:
URL: https://github.com/apache/kafka/pull/18928#discussion_r1965421190
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -208,6 +222,67 @@ public void describeGroups() throws ExecutionException,
InterruptedException {
}
}
+ Map<String, Throwable> deleteShareGroups() {
+ List<GroupListing> shareGroupIds = listDetailedShareGroups();
+ List<String> groupIds = opts.options.has(opts.allGroupsOpt)
+ ? shareGroupIds.stream().map(GroupListing::groupId).toList()
+ : opts.options.valuesOf(opts.groupOpt);
+
+ if (groupIds.isEmpty()) {
+ throw new IllegalArgumentException("--groups or --all-groups
argument is mandatory");
+ }
+
+ for (String groupId : groupIds) {
+ Optional<GroupListing> listing =
shareGroupIds.stream().filter(item -> item.groupId().equals(groupId)).findAny();
+ if (listing.isEmpty()) {
+ throw new IllegalArgumentException("Group '" + groupId +
"' is not a share group.");
+ } else {
+ Optional<GroupState> state = listing.get().groupState();
+ if (state.isPresent() &&
!state.get().equals(GroupState.EMPTY)) {
+ throw new IllegalStateException("Share group '" +
groupId + "' is not EMPTY.");
+ }
+ }
+ }
+
+ Map<String, KafkaFuture<Void>> groupsToDelete =
adminClient.deleteShareGroups(
+ groupIds,
+ withTimeoutMs(new DeleteShareGroupsOptions())
+ ).deletedGroups();
+
+ Map<String, Throwable> success = new HashMap<>();
+ Map<String, Throwable> failed = new HashMap<>();
+
+ groupsToDelete.forEach((g, f) -> {
+ try {
+ f.get();
+ success.put(g, null);
+ } catch (InterruptedException ie) {
+ failed.put(g, ie);
+ } catch (ExecutionException e) {
+ failed.put(g, e.getCause());
+ }
+ });
+
+ if (failed.isEmpty())
+ System.out.println("Deletion of requested share groups (" +
"'" +
success.keySet().stream().map(Object::toString).collect(Collectors.joining(",
")) + "'" + ") was successful.");
+ else {
+ printError("Deletion of some share groups failed:",
Optional.empty());
+ failed.forEach((group, error) -> System.out.println("* Share
Group '" + group + "' could not be deleted due to: " + error));
Review Comment:
nit: `Share group` not `Share Group` please.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteShareGroupsHandler.java:
##########
@@ -0,0 +1,33 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.utils.LogContext;
+
+public class DeleteShareGroupsHandler extends DeleteConsumerGroupsHandler {
Review Comment:
Again, in a similar vein, maybe having a common `DeleteGroupsHandler` and
having 2 subclasses would be sensible here. Having the share group handler
inherit from the consumer group one is strange. I'm pretty sure that there will
be an `Admin.deleteGroups` method before long too.
##########
clients/src/main/java/org/apache/kafka/clients/admin/Admin.java:
##########
@@ -953,6 +953,14 @@ default ListConsumerGroupOffsetsResult
listConsumerGroupOffsets(Map<String, List
*/
DeleteConsumerGroupsResult deleteConsumerGroups(Collection<String>
groupIds, DeleteConsumerGroupsOptions options);
+ /**
Review Comment:
Personally, I'd prefer to see these two methods declared with the other
share group methods, not mixed in with the consumer group methods.
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -208,6 +222,67 @@ public void describeGroups() throws ExecutionException,
InterruptedException {
}
}
+ Map<String, Throwable> deleteShareGroups() {
+ List<GroupListing> shareGroupIds = listDetailedShareGroups();
+ List<String> groupIds = opts.options.has(opts.allGroupsOpt)
+ ? shareGroupIds.stream().map(GroupListing::groupId).toList()
+ : opts.options.valuesOf(opts.groupOpt);
+
+ if (groupIds.isEmpty()) {
+ throw new IllegalArgumentException("--groups or --all-groups
argument is mandatory");
Review Comment:
I would have said that the checking that there is either `--groups` or
`--all-groups` should be done in the command line validation, not here.
As it stands, if the user specified `--all-groups` and there are no share
groups, the error message is `--groups or --all-groups is mandatory`. Really, I
would have said that this should be treated as successful deletion of no
groups. That's what `kafka-consumer-groups.sh` does. It says "Deletion of
requested share groups ('') was successful." Slight inelegant, but effective.
##########
clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupsResult.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * 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 java.util.Collection;
+import java.util.Map;
+
+/**
+ * The result of the {@link Admin#deleteShareGroups(Collection <String>,
DeleteShareGroupsOptions)} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class DeleteShareGroupsResult extends DeleteConsumerGroupsResult {
+ DeleteShareGroupsResult(final Map<String, KafkaFuture<Void>> futures) {
+ super(futures);
+ }
+}
Review Comment:
Yes, OK. I suggest making a common superclass `DeleteGroupsResult` and
having both `DeleteConsumerGroupsResult` and `DeleteShareGroupsResult`
inheriting from that.
--
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
