You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "smjn (via GitHub)" <gi...@apache.org> on 2025/02/10 12:33:21 UTC
[PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
smjn opened a new pull request, #18848:
URL: https://github.com/apache/kafka/pull/18848
* In this PR, we have added GC side impl to call the delete state share coord RPC using the persister.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1954182183
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -6280,6 +6281,29 @@ public void createGroupTombstoneRecords(
group.createGroupTombstoneRecords(records);
}
+ /**
+ * Returns all share partitions keys as a map from the input list of share groups.
+ * @param shareGroups - A list representing share groups.
+ * @return Map representing the share partition keys for all the groups in the input.
+ */
+ public Map<String, Map<Uuid, List<Integer>>> sharePartitionKeysMap(List<ShareGroup> shareGroups) {
+ Map<String, Map<Uuid, List<Integer>>> keyMap = new HashMap<>();
+ if (metadataImage == null) {
+ return Map.of();
+ }
+ TopicsImage topicsImage = metadataImage.topics();
+ for (ShareGroup shareGroup : shareGroups) {
+ String groupId = shareGroup.groupId();
+ for (String topic : shareGroup.subscribedTopicNames().keySet()) {
+ TopicImage topicImage = topicsImage.getTopic(topic);
Review Comment:
Not yet, but we do have another RPC planned
DeleteShareGroupOffsets which will be utilized to do just that.
https://cwiki.apache.org/confluence/display/KAFKA/KIP-932%3A+Queues+for+Kafka#KIP932:QueuesforKafka-DeleteShareGroupOffsetsAPI
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1963643829
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -6855,6 +6861,45 @@ public void createGroupTombstoneRecords(
group.createGroupTombstoneRecords(records);
}
+ /**
+ * Returns an optional of delete share group request object to be used with the persister.
+ * Empty if no subscribed topics or if the share group is empty.
+ * @param shareGroup - A share group
+ * @return Optional of object representing the share group state delete request.
+ */
+ public Optional<DeleteShareGroupStateParameters> shareGroupBuildPartitionDeleteRequest(ShareGroup shareGroup) {
+ TopicsImage topicsImage = metadataImage.topics();
+ Set<String> subscribedTopics = shareGroup.subscribedTopicNames().keySet();
Review Comment:
Yes correct, we will be replacing the `shareGroup.subscribedTopicNames` call with fetching data from a new timeline hashmap which would be populated using the initialize state RPC - [KAFKA-18827](https://issues.apache.org/jira/browse/KAFKA-18827).
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1954416039
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
Review Comment:
I really don't like that fact that we will obliterate the share-group state before checking whether the group is actually empty. This seems like a fundamental flaw to arranging the code like this.
Can the persister being invoked from the `GroupCoordinatorShard` instead?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1956698676
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -6280,6 +6286,48 @@ public void createGroupTombstoneRecords(
group.createGroupTombstoneRecords(records);
}
+ /**
+ * Returns an optional of delete share group request object to be used with the persister.
Review Comment:
`shareGroupCheckEmptyAndBuildPartitionDeleteRequest` - thats a mouthful
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1954185387
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -476,6 +479,33 @@ public CoordinatorResult<DeleteGroupsResponseData.DeletableGroupResultCollection
return new CoordinatorResult<>(records, resultCollection);
}
+ /**
+ * Method returns all share partition keys corresponding to a list of groupIds.
+ * The groupIds are first filtered by type to restrict the list to share groups.
+ * @param groupIds - A list of groupIds as string
+ * @param committedOffset - The last committedOffset for the internal topic partition
+ * @return A map representing the share partition structure.
+ */
+ public Map<String, Map<Uuid, List<Integer>>> sharePartitions(List<String> groupIds, long committedOffset) {
+ List<ShareGroup> shareGroups = new ArrayList<>();
+ for (String groupId : groupIds) {
+ try {
+ Group group = groupMetadataManager.group(groupId);
+ if (group instanceof ShareGroup) {
+ shareGroups.add((ShareGroup) group);
+ }
+ } catch (ApiException exception) {
+ // We needn't do anything more than logging here as deleteGroups
+ // method is handling these cases.
+ // Even if some groups cannot be found, we
+ // must check the entire list.
+ log.error("Failed to find group {}", groupId, exception);
Review Comment:
will change to debug
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1954506706
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
Review Comment:
It was discouraged - we can do it ...
So, first we will send a list of all groups to GroupCoordinatorShard which will return a list of delete requests only for VALID SHARE GROUPS. Other groups will be ignored. This is the only extra work - we are only reading state from the a hashmap.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #18848:
URL: https://github.com/apache/kafka/pull/18848
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #18848:
URL: https://github.com/apache/kafka/pull/18848#issuecomment-2671651121
@dajac Thanks for the review
incorporated comments. Replied to Qs
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #18848:
URL: https://github.com/apache/kafka/pull/18848#issuecomment-2660230440
> A couple of small comments remain. I know that this code is being reworked as once the `ShareGroupStatePartitionMetadataRecord` is introduced, so it's up to @smjn whether to address the comments on this PR or save them up for the next PR. Let me know.
@AndrewJSchofield will address shortly
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1956713206
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -6280,6 +6286,48 @@ public void createGroupTombstoneRecords(
group.createGroupTombstoneRecords(records);
}
+ /**
+ * Returns an optional of delete share group request object to be used with the persister.
Review Comment:
Ok - the empty check actually does not belong here. I will move the check to the `GroupCoordinatorShard.sharePartitionDeleteRequests`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #18848:
URL: https://github.com/apache/kafka/pull/18848#issuecomment-2654353784
> Thanks for the PR. Only a superficial review as this point so I understand the overall flow and can do a deeper review when I have a few hours to spare :)
Thanks for the review - I understand that its non-trivial. To help with the review I've added a pseudo-code style explanation of the code flow.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1954524147
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
Review Comment:
@dajac Thanks for the response. We have a few other interactions which are a similar shape, so this is helpful guidance for those too.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #18848:
URL: https://github.com/apache/kafka/pull/18848#issuecomment-2656614375
Thank you @dajac @AndrewJSchofield for the helpful reviews. Have incorporated most items.
@dajac (there were some incorrect indent settings in my IDE - have fixed them now).
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1954410000
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
+ .thenCompose(groupErrMap -> {
+ DeleteGroupsResponseData.DeletableGroupResultCollection collection = new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ List<String> errGroupIds = new ArrayList<>();
+ groupErrMap.forEach((groupId, error) -> {
+ if (error.code() != Errors.NONE.code()) {
+ log.error("Error deleting share group {} due to error {}", groupId, error);
+ errGroupIds.add(groupId);
+ collection.add(
+ new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(groupId)
+ .setErrorCode(error.code())
+ );
+ }
+ });
+
+ Set<String> groupSet = new HashSet<>(groupList);
+ // Remove all share group ids which have errored out
+ // when deleting with persister.
+ errGroupIds.forEach(groupSet::remove);
Review Comment:
It's probably relatively quick for these small sets though.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1963519484
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -6855,6 +6861,45 @@ public void createGroupTombstoneRecords(
group.createGroupTombstoneRecords(records);
}
+ /**
+ * Returns an optional of delete share group request object to be used with the persister.
+ * Empty if no subscribed topics or if the share group is empty.
+ * @param shareGroup - A share group
+ * @return Optional of object representing the share group state delete request.
+ */
+ public Optional<DeleteShareGroupStateParameters> shareGroupBuildPartitionDeleteRequest(ShareGroup shareGroup) {
+ TopicsImage topicsImage = metadataImage.topics();
+ Set<String> subscribedTopics = shareGroup.subscribedTopicNames().keySet();
Review Comment:
For my understanding, this set is always empty given that we require the group to be empty, right? This will be addressed in a follow-up PR if I understood correctly.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -1627,6 +1643,378 @@ public void testDeleteGroups() throws Exception {
assertEquals(expectedResultCollection, future.get());
}
+ @Test
+ public void testDeleteWithShareGroups() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1");
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection2 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // non-share group
+ DeleteGroupsResponseData.DeletableGroupResult result2 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("group-id-2");
+ resultCollection2.add(result2);
+
+ // null
+ DeleteGroupsResponseData.DeletableGroupResult result3 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(null)
+ .setErrorCode(Errors.INVALID_GROUP_ID.code());
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.addAll(List.of(
+ result3.duplicate(),
+ result2.duplicate(),
+ result1.duplicate()
+ ));
+
+ Uuid shareGroupTopicId = Uuid.randomUuid();
+
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-1", Map.entry(createDeleteShareRequest("share-group-id-1", shareGroupTopicId, List.of(0, 1)), Errors.NONE))
+ )).thenReturn(CompletableFuture.completedFuture(Map.of())); // non-share group
+
+ when(persister.deleteState(any())).thenReturn(CompletableFuture.completedFuture(
+ new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(List.of(
+ new TopicData<>(
+ shareGroupTopicId,
+ List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message()),
+ PartitionFactory.newPartitionErrorData(1, Errors.NONE.code(), Errors.NONE.message())
+ ))
+ ))
+ .build()
+ ));
+
+ // share-group-id-1
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 1)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection1));
+
+ // group-id-2
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 0)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection2));
+
+ List<String> groupIds = Arrays.asList("share-group-id-1", "group-id-2", null);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
+ service.deleteGroups(requestContext(ApiKeys.DELETE_GROUPS), groupIds, BufferSupplier.NO_CACHING);
+
+ try {
+ future.getNow(null);
+ } catch (Exception e) {
+ fail(e);
+ }
Review Comment:
nit: This is not necessary because the exception will be caught anyway by the test running.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -1627,6 +1643,378 @@ public void testDeleteGroups() throws Exception {
assertEquals(expectedResultCollection, future.get());
}
+ @Test
+ public void testDeleteWithShareGroups() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1");
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection2 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // non-share group
+ DeleteGroupsResponseData.DeletableGroupResult result2 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("group-id-2");
+ resultCollection2.add(result2);
+
+ // null
+ DeleteGroupsResponseData.DeletableGroupResult result3 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(null)
+ .setErrorCode(Errors.INVALID_GROUP_ID.code());
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.addAll(List.of(
+ result3.duplicate(),
+ result2.duplicate(),
+ result1.duplicate()
+ ));
+
+ Uuid shareGroupTopicId = Uuid.randomUuid();
+
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-1", Map.entry(createDeleteShareRequest("share-group-id-1", shareGroupTopicId, List.of(0, 1)), Errors.NONE))
+ )).thenReturn(CompletableFuture.completedFuture(Map.of())); // non-share group
+
+ when(persister.deleteState(any())).thenReturn(CompletableFuture.completedFuture(
+ new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(List.of(
+ new TopicData<>(
+ shareGroupTopicId,
+ List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message()),
+ PartitionFactory.newPartitionErrorData(1, Errors.NONE.code(), Errors.NONE.message())
+ ))
+ ))
+ .build()
+ ));
+
+ // share-group-id-1
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 1)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection1));
+
+ // group-id-2
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 0)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection2));
+
+ List<String> groupIds = Arrays.asList("share-group-id-1", "group-id-2", null);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
+ service.deleteGroups(requestContext(ApiKeys.DELETE_GROUPS), groupIds, BufferSupplier.NO_CACHING);
+
+ try {
+ future.getNow(null);
+ } catch (Exception e) {
+ fail(e);
+ }
+ assertEquals(expectedResultCollection, future.get());
+ verify(persister, times(1)).deleteState(any());
+ }
+
+ @Test
+ public void testDeleteShareGroupPersisterError() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group err
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1")
+ .setErrorCode(Errors.UNKNOWN_SERVER_ERROR.code());
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection2 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group success
+ DeleteGroupsResponseData.DeletableGroupResult result2 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-2");
+ resultCollection2.add(result2);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.addAll(Arrays.asList(
+ result1.duplicate(),
+ result2.duplicate()));
+
+ Uuid shareGroupTopicId = Uuid.randomUuid();
+ Uuid shareGroupTopicId2 = Uuid.randomUuid();
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-1", Map.entry(createDeleteShareRequest("share-group-id-1", shareGroupTopicId, List.of(0, 1)), Errors.NONE))
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-2", Map.entry(createDeleteShareRequest("share-group-id-2", shareGroupTopicId2, List.of(0, 1)), Errors.NONE))
+ ));
+
+ when(persister.deleteState(any())).thenReturn(CompletableFuture.completedFuture(
+ new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(List.of(
+ new TopicData<>(
+ shareGroupTopicId,
+ List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.UNKNOWN_SERVER_ERROR.code(), Errors.UNKNOWN_SERVER_ERROR.message()),
+ PartitionFactory.newPartitionErrorData(1, Errors.UNKNOWN_SERVER_ERROR.code(), Errors.UNKNOWN_SERVER_ERROR.message())
+ ))
+ ))
+ .build()
+ )).thenReturn(CompletableFuture.completedFuture(
+ new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(List.of(
+ new TopicData<>(
+ shareGroupTopicId2,
+ List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message()),
+ PartitionFactory.newPartitionErrorData(1, Errors.NONE.code(), Errors.NONE.message())
+ ))
+ ))
+ .build()
+ ));
+
+ // share-group-id-1
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 1)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection1));
+
+ // share-group-id-2
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 2)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection2));
+
+ List<String> groupIds = List.of("share-group-id-1", "share-group-id-2");
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
+ service.deleteGroups(requestContext(ApiKeys.DELETE_GROUPS), groupIds, BufferSupplier.NO_CACHING);
+
+ try {
+ future.getNow(null);
+ } catch (Exception e) {
+ fail(e);
+ }
+ assertEquals(expectedResultCollection, future.get());
+ verify(persister, times(2)).deleteState(any());
+ }
+
+ @Test
+ public void testDeleteShareGroupCoordinatorShareSpecificWriteError() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group err
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1")
+ .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code());
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.add(
+ result1.duplicate()
+ );
+
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.failedFuture(
+ Errors.COORDINATOR_NOT_AVAILABLE.exception()
+ ));
+
+ // share-group-id-1
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 1)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection1));
+
+ List<String> groupIds = List.of("share-group-id-1");
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
+ service.deleteGroups(requestContext(ApiKeys.DELETE_GROUPS), groupIds, BufferSupplier.NO_CACHING);
+
+ try {
+ future.getNow(null);
+ } catch (Exception e) {
+ fail(e);
+ }
+ assertEquals(expectedResultCollection, future.get());
+ verify(persister, times(0)).deleteState(any());
+ }
+
+ @Test
+ public void testDeleteShareGroupNotEmptyError() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group err
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1")
+ .setErrorCode(Errors.forException(new GroupNotEmptyException("bad stuff")).code());
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.add(
+ result1.duplicate()
+ );
+
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-1", Map.entry(EMPTY_PARAMS, Errors.forException(new GroupNotEmptyException("bad stuff"))))
+ ));
+
+ List<String> groupIds = List.of("share-group-id-1");
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
+ service.deleteGroups(requestContext(ApiKeys.DELETE_GROUPS), groupIds, BufferSupplier.NO_CACHING);
+
+ try {
+ future.getNow(null);
+ } catch (Exception e) {
+ fail(e);
+ }
Review Comment:
ditto.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -846,6 +861,152 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
(accumulator, newResults) -> newResults.forEach(result -> accumulator.add(result.duplicate())));
}
+ List<String> deleteCandidateGroupIds(
Review Comment:
Should it be private? If we use it from tests, let's add `// package-private for testing` please.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -1627,6 +1643,378 @@ public void testDeleteGroups() throws Exception {
assertEquals(expectedResultCollection, future.get());
}
+ @Test
+ public void testDeleteWithShareGroups() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1");
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection2 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // non-share group
+ DeleteGroupsResponseData.DeletableGroupResult result2 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("group-id-2");
+ resultCollection2.add(result2);
+
+ // null
+ DeleteGroupsResponseData.DeletableGroupResult result3 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(null)
+ .setErrorCode(Errors.INVALID_GROUP_ID.code());
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.addAll(List.of(
+ result3.duplicate(),
+ result2.duplicate(),
+ result1.duplicate()
+ ));
+
+ Uuid shareGroupTopicId = Uuid.randomUuid();
+
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-1", Map.entry(createDeleteShareRequest("share-group-id-1", shareGroupTopicId, List.of(0, 1)), Errors.NONE))
+ )).thenReturn(CompletableFuture.completedFuture(Map.of())); // non-share group
+
+ when(persister.deleteState(any())).thenReturn(CompletableFuture.completedFuture(
+ new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(List.of(
+ new TopicData<>(
+ shareGroupTopicId,
+ List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message()),
+ PartitionFactory.newPartitionErrorData(1, Errors.NONE.code(), Errors.NONE.message())
+ ))
+ ))
+ .build()
+ ));
+
+ // share-group-id-1
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 1)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection1));
+
+ // group-id-2
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 0)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection2));
+
+ List<String> groupIds = Arrays.asList("share-group-id-1", "group-id-2", null);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
+ service.deleteGroups(requestContext(ApiKeys.DELETE_GROUPS), groupIds, BufferSupplier.NO_CACHING);
+
+ try {
+ future.getNow(null);
+ } catch (Exception e) {
+ fail(e);
+ }
+ assertEquals(expectedResultCollection, future.get());
+ verify(persister, times(1)).deleteState(any());
+ }
+
+ @Test
+ public void testDeleteShareGroupPersisterError() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group err
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1")
+ .setErrorCode(Errors.UNKNOWN_SERVER_ERROR.code());
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection2 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group success
+ DeleteGroupsResponseData.DeletableGroupResult result2 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-2");
+ resultCollection2.add(result2);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.addAll(Arrays.asList(
+ result1.duplicate(),
+ result2.duplicate()));
+
+ Uuid shareGroupTopicId = Uuid.randomUuid();
+ Uuid shareGroupTopicId2 = Uuid.randomUuid();
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-1", Map.entry(createDeleteShareRequest("share-group-id-1", shareGroupTopicId, List.of(0, 1)), Errors.NONE))
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-2", Map.entry(createDeleteShareRequest("share-group-id-2", shareGroupTopicId2, List.of(0, 1)), Errors.NONE))
+ ));
+
+ when(persister.deleteState(any())).thenReturn(CompletableFuture.completedFuture(
+ new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(List.of(
+ new TopicData<>(
+ shareGroupTopicId,
+ List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.UNKNOWN_SERVER_ERROR.code(), Errors.UNKNOWN_SERVER_ERROR.message()),
+ PartitionFactory.newPartitionErrorData(1, Errors.UNKNOWN_SERVER_ERROR.code(), Errors.UNKNOWN_SERVER_ERROR.message())
+ ))
+ ))
+ .build()
+ )).thenReturn(CompletableFuture.completedFuture(
+ new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(List.of(
+ new TopicData<>(
+ shareGroupTopicId2,
+ List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message()),
+ PartitionFactory.newPartitionErrorData(1, Errors.NONE.code(), Errors.NONE.message())
+ ))
+ ))
+ .build()
+ ));
+
+ // share-group-id-1
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 1)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection1));
+
+ // share-group-id-2
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 2)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection2));
+
+ List<String> groupIds = List.of("share-group-id-1", "share-group-id-2");
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
+ service.deleteGroups(requestContext(ApiKeys.DELETE_GROUPS), groupIds, BufferSupplier.NO_CACHING);
+
+ try {
+ future.getNow(null);
+ } catch (Exception e) {
+ fail(e);
+ }
+ assertEquals(expectedResultCollection, future.get());
+ verify(persister, times(2)).deleteState(any());
+ }
+
+ @Test
+ public void testDeleteShareGroupCoordinatorShareSpecificWriteError() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group err
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1")
+ .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code());
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.add(
+ result1.duplicate()
+ );
+
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.failedFuture(
+ Errors.COORDINATOR_NOT_AVAILABLE.exception()
+ ));
+
+ // share-group-id-1
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 1)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection1));
+
+ List<String> groupIds = List.of("share-group-id-1");
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
+ service.deleteGroups(requestContext(ApiKeys.DELETE_GROUPS), groupIds, BufferSupplier.NO_CACHING);
+
+ try {
+ future.getNow(null);
+ } catch (Exception e) {
+ fail(e);
+ }
+ assertEquals(expectedResultCollection, future.get());
+ verify(persister, times(0)).deleteState(any());
+ }
+
+ @Test
+ public void testDeleteShareGroupNotEmptyError() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group err
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1")
+ .setErrorCode(Errors.forException(new GroupNotEmptyException("bad stuff")).code());
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.add(
+ result1.duplicate()
+ );
+
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-1", Map.entry(EMPTY_PARAMS, Errors.forException(new GroupNotEmptyException("bad stuff"))))
+ ));
+
+ List<String> groupIds = List.of("share-group-id-1");
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
+ service.deleteGroups(requestContext(ApiKeys.DELETE_GROUPS), groupIds, BufferSupplier.NO_CACHING);
+
+ try {
+ future.getNow(null);
+ } catch (Exception e) {
+ fail(e);
+ }
+ assertEquals(expectedResultCollection, future.get());
+ // If there is error creating share group delete req
+ // neither persister call nor general delete groups call is made.
+ verify(persister, times(0)).deleteState(any());
+ verify(runtime, times(0)).scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ );
+ }
+
+ @Test
+ public void testDeleteShareGroupCoordinatorGeneralWriteError() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group err
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1")
+ .setErrorCode(Errors.CLUSTER_AUTHORIZATION_FAILED.code());
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.add(
+ result1.duplicate()
+ );
+
+ Uuid shareGroupTopicId = Uuid.randomUuid();
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-1", Map.entry(createDeleteShareRequest("share-group-id-1", shareGroupTopicId, List.of(0, 1)), Errors.NONE))
+ ));
+
+ // share-group-id-1
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 1)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.failedFuture(Errors.CLUSTER_AUTHORIZATION_FAILED.exception()));
+
+ when(persister.deleteState(any())).thenReturn(CompletableFuture.completedFuture(new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(List.of(
+ new TopicData<>(
+ shareGroupTopicId,
+ List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message()),
+ PartitionFactory.newPartitionErrorData(1, Errors.NONE.code(), Errors.NONE.message())
+ ))
+ ))
+ .build()
+ ));
+
+ List<String> groupIds = List.of("share-group-id-1");
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
+ service.deleteGroups(requestContext(ApiKeys.DELETE_GROUPS), groupIds, BufferSupplier.NO_CACHING);
+
+ try {
+ future.getNow(null);
+ } catch (Exception e) {
+ fail(e);
+ }
Review Comment:
ditto.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -846,6 +861,152 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
(accumulator, newResults) -> newResults.forEach(result -> accumulator.add(result.duplicate())));
}
+ List<String> deleteCandidateGroupIds(
+ Map<String, Errors> groupErrMap,
+ List<String> groupList,
+ DeleteGroupsResponseData.DeletableGroupResultCollection collection
+ ) {
+ List<String> errGroupIds = new ArrayList<>();
+ groupErrMap.forEach((groupId, error) -> {
+ if (error.code() != Errors.NONE.code()) {
+ log.error("Error deleting share group {} due to error {}", groupId, error);
+ errGroupIds.add(groupId);
+ collection.add(
+ new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(groupId)
+ .setErrorCode(error.code())
+ );
+ }
+ });
+
+ Set<String> groupSet = new HashSet<>(groupList);
+ // Remove all share group ids which have errored out
+ // when deleting with persister.
+ groupSet.removeAll(errGroupIds);
+
+ // Let us invoke the standard procedure of any non-share
+ // groups or successfully deleted share groups remaining.
+ return groupSet.stream().toList();
+ }
+
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> handleDeleteGroups(
Review Comment:
ditto.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -2532,4 +2927,17 @@ public GroupCoordinatorServiceBuilder setMetrics(GroupCoordinatorMetrics metrics
return this;
}
}
+
+ private DeleteShareGroupStateParameters createDeleteShareRequest(String groupId, Uuid topic, List<Integer> partitions) {
Review Comment:
nit: Should we make it static?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -823,20 +831,27 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future = deleteShareGroups(topicPartition, groupList).thenCompose(groupErrMap -> {
+ DeleteGroupsResponseData.DeletableGroupResultCollection collection = new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ List<String> retainedGroupIds = deleteCandidateGroupIds(groupErrMap, groupList, collection);
+ if (retainedGroupIds.isEmpty()) {
+ return CompletableFuture.completedFuture(collection);
+ }
+ return handleDeleteGroups(context, topicPartition, retainedGroupIds)
+ .whenComplete((resp, __) -> resp.forEach(result -> collection.add(result.duplicate())))
+ .thenApply(__ -> collection);
+ });
+ // deleteShareGroups has its own exceptionally block, so we don't need one here.
+
+ // This future object has the following stages:
+ // - First it invokes the share group delete flow where the shard sharePartitionDeleteRequests
+ // method is invoked, and it returns request objects for each valid share group passed to it.
+ // - Then the requests are passed to the persister.deleteStata method one at a time. The results
Review Comment:
nit: `deleteStata` -> `deleteState`?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -1627,6 +1643,378 @@ public void testDeleteGroups() throws Exception {
assertEquals(expectedResultCollection, future.get());
}
+ @Test
+ public void testDeleteWithShareGroups() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1");
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection2 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // non-share group
+ DeleteGroupsResponseData.DeletableGroupResult result2 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("group-id-2");
+ resultCollection2.add(result2);
+
+ // null
+ DeleteGroupsResponseData.DeletableGroupResult result3 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(null)
+ .setErrorCode(Errors.INVALID_GROUP_ID.code());
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.addAll(List.of(
+ result3.duplicate(),
+ result2.duplicate(),
+ result1.duplicate()
+ ));
+
+ Uuid shareGroupTopicId = Uuid.randomUuid();
+
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-1", Map.entry(createDeleteShareRequest("share-group-id-1", shareGroupTopicId, List.of(0, 1)), Errors.NONE))
+ )).thenReturn(CompletableFuture.completedFuture(Map.of())); // non-share group
+
+ when(persister.deleteState(any())).thenReturn(CompletableFuture.completedFuture(
+ new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(List.of(
+ new TopicData<>(
+ shareGroupTopicId,
+ List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message()),
+ PartitionFactory.newPartitionErrorData(1, Errors.NONE.code(), Errors.NONE.message())
+ ))
+ ))
+ .build()
+ ));
+
+ // share-group-id-1
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 1)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection1));
+
+ // group-id-2
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 0)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection2));
+
+ List<String> groupIds = Arrays.asList("share-group-id-1", "group-id-2", null);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
+ service.deleteGroups(requestContext(ApiKeys.DELETE_GROUPS), groupIds, BufferSupplier.NO_CACHING);
+
+ try {
+ future.getNow(null);
+ } catch (Exception e) {
+ fail(e);
+ }
+ assertEquals(expectedResultCollection, future.get());
+ verify(persister, times(1)).deleteState(any());
+ }
+
+ @Test
+ public void testDeleteShareGroupPersisterError() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group err
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1")
+ .setErrorCode(Errors.UNKNOWN_SERVER_ERROR.code());
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection2 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group success
+ DeleteGroupsResponseData.DeletableGroupResult result2 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-2");
+ resultCollection2.add(result2);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.addAll(Arrays.asList(
+ result1.duplicate(),
+ result2.duplicate()));
+
+ Uuid shareGroupTopicId = Uuid.randomUuid();
+ Uuid shareGroupTopicId2 = Uuid.randomUuid();
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-1", Map.entry(createDeleteShareRequest("share-group-id-1", shareGroupTopicId, List.of(0, 1)), Errors.NONE))
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-2", Map.entry(createDeleteShareRequest("share-group-id-2", shareGroupTopicId2, List.of(0, 1)), Errors.NONE))
+ ));
+
+ when(persister.deleteState(any())).thenReturn(CompletableFuture.completedFuture(
+ new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(List.of(
+ new TopicData<>(
+ shareGroupTopicId,
+ List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.UNKNOWN_SERVER_ERROR.code(), Errors.UNKNOWN_SERVER_ERROR.message()),
+ PartitionFactory.newPartitionErrorData(1, Errors.UNKNOWN_SERVER_ERROR.code(), Errors.UNKNOWN_SERVER_ERROR.message())
+ ))
+ ))
+ .build()
+ )).thenReturn(CompletableFuture.completedFuture(
+ new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(List.of(
+ new TopicData<>(
+ shareGroupTopicId2,
+ List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message()),
+ PartitionFactory.newPartitionErrorData(1, Errors.NONE.code(), Errors.NONE.message())
+ ))
+ ))
+ .build()
+ ));
+
+ // share-group-id-1
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 1)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection1));
+
+ // share-group-id-2
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 2)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection2));
+
+ List<String> groupIds = List.of("share-group-id-1", "share-group-id-2");
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
+ service.deleteGroups(requestContext(ApiKeys.DELETE_GROUPS), groupIds, BufferSupplier.NO_CACHING);
+
+ try {
+ future.getNow(null);
+ } catch (Exception e) {
+ fail(e);
+ }
+ assertEquals(expectedResultCollection, future.get());
+ verify(persister, times(2)).deleteState(any());
+ }
+
+ @Test
+ public void testDeleteShareGroupCoordinatorShareSpecificWriteError() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group err
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1")
+ .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code());
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.add(
+ result1.duplicate()
+ );
+
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.failedFuture(
+ Errors.COORDINATOR_NOT_AVAILABLE.exception()
+ ));
+
+ // share-group-id-1
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 1)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection1));
+
+ List<String> groupIds = List.of("share-group-id-1");
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
+ service.deleteGroups(requestContext(ApiKeys.DELETE_GROUPS), groupIds, BufferSupplier.NO_CACHING);
+
+ try {
+ future.getNow(null);
+ } catch (Exception e) {
+ fail(e);
+ }
Review Comment:
ditto.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -1627,6 +1643,378 @@ public void testDeleteGroups() throws Exception {
assertEquals(expectedResultCollection, future.get());
}
+ @Test
+ public void testDeleteWithShareGroups() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1");
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection2 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // non-share group
+ DeleteGroupsResponseData.DeletableGroupResult result2 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("group-id-2");
+ resultCollection2.add(result2);
+
+ // null
+ DeleteGroupsResponseData.DeletableGroupResult result3 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(null)
+ .setErrorCode(Errors.INVALID_GROUP_ID.code());
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.addAll(List.of(
+ result3.duplicate(),
+ result2.duplicate(),
+ result1.duplicate()
+ ));
+
+ Uuid shareGroupTopicId = Uuid.randomUuid();
+
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-1", Map.entry(createDeleteShareRequest("share-group-id-1", shareGroupTopicId, List.of(0, 1)), Errors.NONE))
+ )).thenReturn(CompletableFuture.completedFuture(Map.of())); // non-share group
+
+ when(persister.deleteState(any())).thenReturn(CompletableFuture.completedFuture(
+ new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(List.of(
+ new TopicData<>(
+ shareGroupTopicId,
+ List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message()),
+ PartitionFactory.newPartitionErrorData(1, Errors.NONE.code(), Errors.NONE.message())
+ ))
+ ))
+ .build()
+ ));
+
+ // share-group-id-1
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 1)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection1));
+
+ // group-id-2
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 0)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection2));
+
+ List<String> groupIds = Arrays.asList("share-group-id-1", "group-id-2", null);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
+ service.deleteGroups(requestContext(ApiKeys.DELETE_GROUPS), groupIds, BufferSupplier.NO_CACHING);
+
+ try {
+ future.getNow(null);
+ } catch (Exception e) {
+ fail(e);
+ }
+ assertEquals(expectedResultCollection, future.get());
+ verify(persister, times(1)).deleteState(any());
+ }
+
+ @Test
+ public void testDeleteShareGroupPersisterError() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group err
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1")
+ .setErrorCode(Errors.UNKNOWN_SERVER_ERROR.code());
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection2 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group success
+ DeleteGroupsResponseData.DeletableGroupResult result2 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-2");
+ resultCollection2.add(result2);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.addAll(Arrays.asList(
+ result1.duplicate(),
+ result2.duplicate()));
+
+ Uuid shareGroupTopicId = Uuid.randomUuid();
+ Uuid shareGroupTopicId2 = Uuid.randomUuid();
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-1", Map.entry(createDeleteShareRequest("share-group-id-1", shareGroupTopicId, List.of(0, 1)), Errors.NONE))
+ )).thenReturn(CompletableFuture.completedFuture(
+ Map.of("share-group-id-2", Map.entry(createDeleteShareRequest("share-group-id-2", shareGroupTopicId2, List.of(0, 1)), Errors.NONE))
+ ));
+
+ when(persister.deleteState(any())).thenReturn(CompletableFuture.completedFuture(
+ new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(List.of(
+ new TopicData<>(
+ shareGroupTopicId,
+ List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.UNKNOWN_SERVER_ERROR.code(), Errors.UNKNOWN_SERVER_ERROR.message()),
+ PartitionFactory.newPartitionErrorData(1, Errors.UNKNOWN_SERVER_ERROR.code(), Errors.UNKNOWN_SERVER_ERROR.message())
+ ))
+ ))
+ .build()
+ )).thenReturn(CompletableFuture.completedFuture(
+ new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(List.of(
+ new TopicData<>(
+ shareGroupTopicId2,
+ List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message()),
+ PartitionFactory.newPartitionErrorData(1, Errors.NONE.code(), Errors.NONE.message())
+ ))
+ ))
+ .build()
+ ));
+
+ // share-group-id-1
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 1)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection1));
+
+ // share-group-id-2
+ when(runtime.scheduleWriteOperation(
+ ArgumentMatchers.eq("delete-groups"),
+ ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 2)),
+ ArgumentMatchers.eq(Duration.ofMillis(5000)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(resultCollection2));
+
+ List<String> groupIds = List.of("share-group-id-1", "share-group-id-2");
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
+ service.deleteGroups(requestContext(ApiKeys.DELETE_GROUPS), groupIds, BufferSupplier.NO_CACHING);
+
+ try {
+ future.getNow(null);
+ } catch (Exception e) {
+ fail(e);
+ }
Review Comment:
ditto.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1952976685
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -476,6 +479,34 @@ public CoordinatorResult<DeleteGroupsResponseData.DeletableGroupResultCollection
return new CoordinatorResult<>(records, resultCollection);
}
+ /**
+ * Method returns all share partition keys corresponding to a list of groupIds.
+ * The groupIds are first filtered by type to restrict the list to share groups.
+ * @param groupIds - A list of groupIds as string
+ * @param committedOffset - The last committedOffset for the internal topic partition
+ * @return A map representing the share partition structure.
+ * @throws ApiException
+ */
+ public Map<String, Map<Uuid, List<Integer>>> sharePartitions(List<String> groupIds, long committedOffset) throws ApiException {
+ List<ShareGroup> shareGroups = new ArrayList<>();
+ for (String groupId : groupIds) {
+ try {
+ Group group = groupMetadataManager.group(groupId);
+ if (group instanceof ShareGroup) {
+ shareGroups.add((ShareGroup) group);
+ }
+ } catch (ApiException exception) {
+ // We needn't do anything more than logging here as deleteGroups
+ // method is handling these cases.
+ // Even if some groups cannot be found, we
+ // must check the entre list.
Review Comment:
nit: "entre" should be "entire"
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -476,6 +479,34 @@ public CoordinatorResult<DeleteGroupsResponseData.DeletableGroupResultCollection
return new CoordinatorResult<>(records, resultCollection);
}
+ /**
+ * Method returns all share partition keys corresponding to a list of groupIds.
+ * The groupIds are first filtered by type to restrict the list to share groups.
+ * @param groupIds - A list of groupIds as string
+ * @param committedOffset - The last committedOffset for the internal topic partition
+ * @return A map representing the share partition structure.
+ * @throws ApiException
Review Comment:
No, it doesn't.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1954242074
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
+ .thenCompose(groupErrMap -> {
+ DeleteGroupsResponseData.DeletableGroupResultCollection collection = new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ List<String> errGroupIds = new ArrayList<>();
+ groupErrMap.forEach((groupId, error) -> {
+ if (error.code() != Errors.NONE.code()) {
+ log.error("Error deleting share group {} due to error {}", groupId, error);
+ errGroupIds.add(groupId);
+ collection.add(
+ new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(groupId)
+ .setErrorCode(error.code())
+ );
+ }
+ });
+
+ Set<String> groupSet = new HashSet<>(groupList);
+ // Remove all share group ids which have errored out
+ // when deleting with persister.
+ errGroupIds.forEach(groupSet::remove);
Review Comment:
changed per IDE suggestion
```
Call to 'set.removeAll(list)' may work slowly
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1954506706
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
Review Comment:
It was discouraged - we can do it ...
So, first we will send a list of all groups to GroupCoordinatorShard which will return a list of delete requests only for VALID SHARE GROUPS. Other groups will be ignored. This is the only extra work - we are only reading state from the a hashmap.
We can alleviate the situation by checking `shareGroup.isEmpty()` in `GroupMetadataManager`. That way, the `persister.deleteState` will not be called for that group but the standard deleteGroups code will be called.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
Review Comment:
using persister in shard was discouraged - we can do it ...
So, first we will send a list of all groups to GroupCoordinatorShard which will return a list of delete requests only for VALID SHARE GROUPS. Other groups will be ignored. This is the only extra work - we are only reading state from the a hashmap.
We can alleviate the situation by checking `shareGroup.isEmpty()` in `GroupMetadataManager`. That way, the `persister.deleteState` will not be called for that group but the standard deleteGroups code will be called.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #18848:
URL: https://github.com/apache/kafka/pull/18848#issuecomment-2656604424
> 3\. I do not see any tombstone records ever being written to the `__share_group_state` topic. I've deleted many share groups using this code and the log dumper never shows any tombstones. There should be a tombstone per share-partition deleted.
That code is part of an earlier merged PR. The persister.deleteState call will result in ShareCoordinator writing those records into the __share_group_state topic. This PR only addresses group coordinator.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #18848:
URL: https://github.com/apache/kafka/pull/18848#issuecomment-2658896304
@AndrewJSchofield
> The deletion of the share group state is scheduled before the call to GroupCoordinatorShard.deleteGroups which actually checks the group type and whether the group is empty.
Have implemented the change in GroupMetadataManager added test too.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1954516380
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
Review Comment:
This is possible but it would require to make the persister work with the event based workflows. Basically, we can not wait on the future from the shard and there is no way at the moment to react to it neither. We could make it work by wrapping the persister into a class which understand how to interface with the runtime/shard but it is more complicated than doing it in the service layer.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1954063627
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -845,6 +896,113 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
(accumulator, newResults) -> newResults.forEach(result -> accumulator.add(result.duplicate())));
}
+ private CompletableFuture<Map<String, Errors>> deleteShareGroups(TopicPartition topicPartition, List<String> groupList) {
+ // topicPartition refers to internal topic __consumer_offsets
+ return runtime.scheduleReadOperation(
+ "delete-share-groups",
+ topicPartition,
+ (coordinator, offset) -> coordinator.sharePartitions(groupList, offset)
+ )
Review Comment:
nit: The indentation is off here. We usually use 4 spaces for the arguments.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -845,6 +896,113 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
(accumulator, newResults) -> newResults.forEach(result -> accumulator.add(result.duplicate())));
}
+ private CompletableFuture<Map<String, Errors>> deleteShareGroups(TopicPartition topicPartition, List<String> groupList) {
Review Comment:
nit: Let's put one argument per line to follow the format of the other methods in this file.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -845,6 +896,113 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
(accumulator, newResults) -> newResults.forEach(result -> accumulator.add(result.duplicate())));
}
+ private CompletableFuture<Map<String, Errors>> deleteShareGroups(TopicPartition topicPartition, List<String> groupList) {
+ // topicPartition refers to internal topic __consumer_offsets
+ return runtime.scheduleReadOperation(
Review Comment:
For consistency reason, I suggest to use a write operation to ensure that you read the last state. Otherwise, there is a change that you have a share group non-committed yet and you would not see it with a read.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -845,6 +896,113 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
(accumulator, newResults) -> newResults.forEach(result -> accumulator.add(result.duplicate())));
}
+ private CompletableFuture<Map<String, Errors>> deleteShareGroups(TopicPartition topicPartition, List<String> groupList) {
+ // topicPartition refers to internal topic __consumer_offsets
+ return runtime.scheduleReadOperation(
+ "delete-share-groups",
+ topicPartition,
+ (coordinator, offset) -> coordinator.sharePartitions(groupList, offset)
+ )
+ .thenCompose(this::performShareGroupsDeletion)
+ .exceptionally(exception -> handleOperationException(
+ "delete-share-groups",
+ groupList,
+ exception,
+ (error, __) -> {
+ Map<String, Errors> errors = new HashMap<>();
+ groupList.forEach(group -> errors.put(group, error));
+ return errors;
+ },
+ log
+ ));
+ }
+
+ private CompletableFuture<Map<String, Errors>> performShareGroupsDeletion(
+ Map<String, Map<Uuid, List<Integer>>> keys
+ ) {
+ List<CompletableFuture<AbstractMap.SimpleEntry<String, DeleteShareGroupStateResult>>> futures = new ArrayList<>();
+ for (Map.Entry<String, Map<Uuid, List<Integer>>> groupEntry : keys.entrySet()) {
+ List<TopicData<PartitionIdData>> topicData = new ArrayList<>();
+ for (Map.Entry<Uuid, List<Integer>> topicEntry : groupEntry.getValue().entrySet()) {
+ topicData.add(
+ new TopicData<>(
+ topicEntry.getKey(),
+ topicEntry.getValue().stream().map(PartitionFactory::newPartitionIdData).toList()
+ )
+ );
+ }
Review Comment:
I wonder whether it would make sense to directly return the data structure that the persister needs from the group metadata manager. We could avoid all those conversions.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
Review Comment:
It would be great if you could put an comment explaining the main flow here.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
+ .thenCompose(groupErrMap -> {
+ DeleteGroupsResponseData.DeletableGroupResultCollection collection = new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ List<String> errGroupIds = new ArrayList<>();
+ groupErrMap.forEach((groupId, error) -> {
+ if (error.code() != Errors.NONE.code()) {
+ log.error("Error deleting share group {} due to error {}", groupId, error);
+ errGroupIds.add(groupId);
+ collection.add(
+ new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(groupId)
+ .setErrorCode(error.code())
+ );
+ }
+ });
+
+ Set<String> groupSet = new HashSet<>(groupList);
+ // Remove all share group ids which have errored out
+ // when deleting with persister.
+ errGroupIds.forEach(groupSet::remove);
+
+ // If no non-share groupIds or non-error share group ids present
+ // return.
+ if (groupSet.isEmpty()) {
+ return CompletableFuture.completedFuture(collection);
+ }
+
+ // Let us invoke the standard procedure of any non-share
+ // groups or successfully deleted share groups remaining.
+ List<String> retainedGroupIds = groupSet.stream().toList();
+ return runtime.scheduleWriteOperation(
Review Comment:
nit: It may be worth extracting this into a method to reduce the code in the `thenCompose`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
Review Comment:
nit: future? shareFuture is not quite right because it is the result of the entire deletion.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -1627,6 +1637,374 @@ public void testDeleteGroups() throws Exception {
assertEquals(expectedResultCollection, future.get());
}
+ @Test
+ public void testDeleteWithShareGroups() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1");
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection2 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // non-share group
+ DeleteGroupsResponseData.DeletableGroupResult result2 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("group-id-2");
+ resultCollection2.add(result2);
+
+ // null
+ DeleteGroupsResponseData.DeletableGroupResult result3 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(null)
+ .setErrorCode(Errors.INVALID_GROUP_ID.code());
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.addAll(List.of(
+ result3.duplicate(),
+ result2.duplicate(),
+ result1.duplicate()
+ )
+ );
+
+ Uuid shareGroupTopicId = Uuid.randomUuid();
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("delete-share-groups"),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ ))
+ .thenReturn(CompletableFuture.completedFuture(
+ Map.of(
+ "share-group-id-1",
+ Map.of(
+ shareGroupTopicId,
+ List.of(0, 1)
+ )
+ )
+ )
+ )
Review Comment:
This looks weird. The indentation is also incorrect. There are many such cases in this file. I let you go through them.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -6280,6 +6281,29 @@ public void createGroupTombstoneRecords(
group.createGroupTombstoneRecords(records);
}
+ /**
+ * Returns all share partitions keys as a map from the input list of share groups.
+ * @param shareGroups - A list representing share groups.
+ * @return Map representing the share partition keys for all the groups in the input.
+ */
+ public Map<String, Map<Uuid, List<Integer>>> sharePartitionKeysMap(List<ShareGroup> shareGroups) {
+ Map<String, Map<Uuid, List<Integer>>> keyMap = new HashMap<>();
+ if (metadataImage == null) {
+ return Map.of();
+ }
+ TopicsImage topicsImage = metadataImage.topics();
+ for (ShareGroup shareGroup : shareGroups) {
+ String groupId = shareGroup.groupId();
+ for (String topic : shareGroup.subscribedTopicNames().keySet()) {
+ TopicImage topicImage = topicsImage.getTopic(topic);
Review Comment:
The topic may not exist any more. For my understanding, how do we handle this case? Does the share coordinator deletes offsets of deleted topics?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -476,6 +479,33 @@ public CoordinatorResult<DeleteGroupsResponseData.DeletableGroupResultCollection
return new CoordinatorResult<>(records, resultCollection);
}
+ /**
+ * Method returns all share partition keys corresponding to a list of groupIds.
+ * The groupIds are first filtered by type to restrict the list to share groups.
+ * @param groupIds - A list of groupIds as string
+ * @param committedOffset - The last committedOffset for the internal topic partition
+ * @return A map representing the share partition structure.
+ */
+ public Map<String, Map<Uuid, List<Integer>>> sharePartitions(List<String> groupIds, long committedOffset) {
+ List<ShareGroup> shareGroups = new ArrayList<>();
+ for (String groupId : groupIds) {
Review Comment:
nit: With the current structure of the code, you iterate on the group ids twice. It may be better to directly get the share partitions of the share group when you have it in hand.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -476,6 +479,33 @@ public CoordinatorResult<DeleteGroupsResponseData.DeletableGroupResultCollection
return new CoordinatorResult<>(records, resultCollection);
}
+ /**
+ * Method returns all share partition keys corresponding to a list of groupIds.
+ * The groupIds are first filtered by type to restrict the list to share groups.
+ * @param groupIds - A list of groupIds as string
+ * @param committedOffset - The last committedOffset for the internal topic partition
+ * @return A map representing the share partition structure.
+ */
+ public Map<String, Map<Uuid, List<Integer>>> sharePartitions(List<String> groupIds, long committedOffset) {
+ List<ShareGroup> shareGroups = new ArrayList<>();
+ for (String groupId : groupIds) {
+ try {
+ Group group = groupMetadataManager.group(groupId);
+ if (group instanceof ShareGroup) {
+ shareGroups.add((ShareGroup) group);
+ }
+ } catch (ApiException exception) {
+ // We needn't do anything more than logging here as deleteGroups
+ // method is handling these cases.
+ // Even if some groups cannot be found, we
+ // must check the entire list.
+ log.error("Failed to find group {}", groupId, exception);
Review Comment:
nit: I would remove this one because it may be spammy.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -476,6 +479,33 @@ public CoordinatorResult<DeleteGroupsResponseData.DeletableGroupResultCollection
return new CoordinatorResult<>(records, resultCollection);
}
+ /**
+ * Method returns all share partition keys corresponding to a list of groupIds.
+ * The groupIds are first filtered by type to restrict the list to share groups.
+ * @param groupIds - A list of groupIds as string
+ * @param committedOffset - The last committedOffset for the internal topic partition
+ * @return A map representing the share partition structure.
+ */
+ public Map<String, Map<Uuid, List<Integer>>> sharePartitions(List<String> groupIds, long committedOffset) {
+ List<ShareGroup> shareGroups = new ArrayList<>();
+ for (String groupId : groupIds) {
+ try {
+ Group group = groupMetadataManager.group(groupId);
+ if (group instanceof ShareGroup) {
+ shareGroups.add((ShareGroup) group);
+ }
+ } catch (ApiException exception) {
Review Comment:
Could we be specific about the exception that we handle here? I suppose that we care about the group id not found.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -845,6 +896,113 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
(accumulator, newResults) -> newResults.forEach(result -> accumulator.add(result.duplicate())));
}
+ private CompletableFuture<Map<String, Errors>> deleteShareGroups(TopicPartition topicPartition, List<String> groupList) {
+ // topicPartition refers to internal topic __consumer_offsets
+ return runtime.scheduleReadOperation(
+ "delete-share-groups",
+ topicPartition,
+ (coordinator, offset) -> coordinator.sharePartitions(groupList, offset)
+ )
+ .thenCompose(this::performShareGroupsDeletion)
+ .exceptionally(exception -> handleOperationException(
+ "delete-share-groups",
+ groupList,
+ exception,
+ (error, __) -> {
+ Map<String, Errors> errors = new HashMap<>();
+ groupList.forEach(group -> errors.put(group, error));
+ return errors;
+ },
+ log
+ ));
+ }
+
+ private CompletableFuture<Map<String, Errors>> performShareGroupsDeletion(
+ Map<String, Map<Uuid, List<Integer>>> keys
+ ) {
+ List<CompletableFuture<AbstractMap.SimpleEntry<String, DeleteShareGroupStateResult>>> futures = new ArrayList<>();
+ for (Map.Entry<String, Map<Uuid, List<Integer>>> groupEntry : keys.entrySet()) {
+ List<TopicData<PartitionIdData>> topicData = new ArrayList<>();
+ for (Map.Entry<Uuid, List<Integer>> topicEntry : groupEntry.getValue().entrySet()) {
+ topicData.add(
+ new TopicData<>(
+ topicEntry.getKey(),
+ topicEntry.getValue().stream().map(PartitionFactory::newPartitionIdData).toList()
+ )
+ );
+ }
+
+ futures.add(deleteShareGroup(groupEntry.getKey(), topicData));
+ }
+
+ return persisterDeleteToGroupIdErrorMap(futures);
+ }
+
+ private CompletableFuture<AbstractMap.SimpleEntry<String, DeleteShareGroupStateResult>> deleteShareGroup(
+ String groupId,
+ List<TopicData<PartitionIdData>> topicData
+ ) {
+ return persister.deleteState(
+ new DeleteShareGroupStateParameters.Builder()
Review Comment:
nit: Indentation seems to be off here too.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -6280,6 +6281,29 @@ public void createGroupTombstoneRecords(
group.createGroupTombstoneRecords(records);
}
+ /**
+ * Returns all share partitions keys as a map from the input list of share groups.
+ * @param shareGroups - A list representing share groups.
+ * @return Map representing the share partition keys for all the groups in the input.
+ */
+ public Map<String, Map<Uuid, List<Integer>>> sharePartitionKeysMap(List<ShareGroup> shareGroups) {
+ Map<String, Map<Uuid, List<Integer>>> keyMap = new HashMap<>();
+ if (metadataImage == null) {
+ return Map.of();
+ }
Review Comment:
nit: metadata image is never null.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
+ .thenCompose(groupErrMap -> {
+ DeleteGroupsResponseData.DeletableGroupResultCollection collection = new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ List<String> errGroupIds = new ArrayList<>();
+ groupErrMap.forEach((groupId, error) -> {
+ if (error.code() != Errors.NONE.code()) {
Review Comment:
What kind of errors can we get here? Are they all expected/allowed by the DeleteGroups API?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -845,6 +896,113 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
(accumulator, newResults) -> newResults.forEach(result -> accumulator.add(result.duplicate())));
}
+ private CompletableFuture<Map<String, Errors>> deleteShareGroups(TopicPartition topicPartition, List<String> groupList) {
+ // topicPartition refers to internal topic __consumer_offsets
+ return runtime.scheduleReadOperation(
+ "delete-share-groups",
+ topicPartition,
+ (coordinator, offset) -> coordinator.sharePartitions(groupList, offset)
+ )
+ .thenCompose(this::performShareGroupsDeletion)
+ .exceptionally(exception -> handleOperationException(
+ "delete-share-groups",
+ groupList,
+ exception,
+ (error, __) -> {
+ Map<String, Errors> errors = new HashMap<>();
+ groupList.forEach(group -> errors.put(group, error));
+ return errors;
+ },
+ log
+ ));
+ }
+
+ private CompletableFuture<Map<String, Errors>> performShareGroupsDeletion(
+ Map<String, Map<Uuid, List<Integer>>> keys
+ ) {
+ List<CompletableFuture<AbstractMap.SimpleEntry<String, DeleteShareGroupStateResult>>> futures = new ArrayList<>();
+ for (Map.Entry<String, Map<Uuid, List<Integer>>> groupEntry : keys.entrySet()) {
+ List<TopicData<PartitionIdData>> topicData = new ArrayList<>();
+ for (Map.Entry<Uuid, List<Integer>> topicEntry : groupEntry.getValue().entrySet()) {
+ topicData.add(
+ new TopicData<>(
+ topicEntry.getKey(),
+ topicEntry.getValue().stream().map(PartitionFactory::newPartitionIdData).toList()
+ )
+ );
+ }
+
+ futures.add(deleteShareGroup(groupEntry.getKey(), topicData));
+ }
+
+ return persisterDeleteToGroupIdErrorMap(futures);
+ }
+
+ private CompletableFuture<AbstractMap.SimpleEntry<String, DeleteShareGroupStateResult>> deleteShareGroup(
+ String groupId,
+ List<TopicData<PartitionIdData>> topicData
+ ) {
+ return persister.deleteState(
+ new DeleteShareGroupStateParameters.Builder()
+ .setGroupTopicPartitionData(new GroupTopicPartitionData.Builder<PartitionIdData>()
+ .setGroupId(groupId)
+ .setTopicsData(topicData)
+ .build()
+ )
+ .build()
+ )
+ .thenCompose(result -> CompletableFuture.completedFuture(new AbstractMap.SimpleEntry<>(groupId, result)))
+ .exceptionally(exception -> {
+ // In case the deleteState call fails,
+ // we should construct the appropriate response here
+ // so that the subsequent callbacks don't see runtime exceptions.
+ log.error("Unable to delete share group partition(s) - {}, {}", groupId, topicData);
+ List<TopicData<PartitionErrorData>> respTopicData = topicData.stream()
+ .map(reqTopicData -> new TopicData<>(
+ reqTopicData.topicId(),
+ reqTopicData.partitions().stream()
+ .map(reqPartData -> {
+ Errors err = Errors.forException(exception);
+ return PartitionFactory.newPartitionErrorData(reqPartData.partition(), err.code(), err.message());
+ })
+ .toList()
+ )
+ )
+ .toList();
+
+ return new AbstractMap.SimpleEntry<>(groupId, new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(respTopicData)
+ .build()
+ );
+ });
+ }
+
+ private CompletableFuture<Map<String, Errors>> persisterDeleteToGroupIdErrorMap(
+ List<CompletableFuture<AbstractMap.SimpleEntry<String, DeleteShareGroupStateResult>>> futures
+ ) {
+ return CompletableFuture.allOf(futures.toArray(new CompletableFuture[]{}))
+ .thenCompose(v -> {
+ Map<String, Errors> groupIds = new HashMap<>();
+ for (CompletableFuture<AbstractMap.SimpleEntry<String, DeleteShareGroupStateResult>> future : futures) {
+ Map.Entry<String, DeleteShareGroupStateResult> entry = future.getNow(null); // safe as within allOff
+ groupIds.putIfAbsent(entry.getKey(), Errors.NONE);
+ for (TopicData<PartitionErrorData> topicData : entry.getValue().topicsData()) {
+ Optional<PartitionErrorData> errItem = topicData.partitions().stream()
+ .filter(errData -> errData.errorCode() != Errors.NONE.code())
+ .findAny();
+
+ errItem.ifPresent(val -> {
+ log.error("Received error while deleting share group {} - {}", entry.getKey(), val);
+ groupIds.put(entry.getKey(), Errors.forCode(val.errorCode()));
+ }
+ );
Review Comment:
nit: Format is incorrect here too.
```
errItem.ifPresent(val -> {
log.error("Received error while deleting share group {} - {}", entry.getKey(), val);
groupIds.put(entry.getKey(), Errors.forCode(val.errorCode()));
});
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
+ .thenCompose(groupErrMap -> {
+ DeleteGroupsResponseData.DeletableGroupResultCollection collection = new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ List<String> errGroupIds = new ArrayList<>();
+ groupErrMap.forEach((groupId, error) -> {
+ if (error.code() != Errors.NONE.code()) {
+ log.error("Error deleting share group {} due to error {}", groupId, error);
+ errGroupIds.add(groupId);
+ collection.add(
+ new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(groupId)
+ .setErrorCode(error.code())
+ );
+ }
+ });
+
+ Set<String> groupSet = new HashSet<>(groupList);
+ // Remove all share group ids which have errored out
+ // when deleting with persister.
+ errGroupIds.forEach(groupSet::remove);
Review Comment:
nit: removeAll?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
+ .thenCompose(groupErrMap -> {
+ DeleteGroupsResponseData.DeletableGroupResultCollection collection = new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ List<String> errGroupIds = new ArrayList<>();
+ groupErrMap.forEach((groupId, error) -> {
+ if (error.code() != Errors.NONE.code()) {
+ log.error("Error deleting share group {} due to error {}", groupId, error);
+ errGroupIds.add(groupId);
+ collection.add(
+ new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(groupId)
+ .setErrorCode(error.code())
+ );
+ }
+ });
+
+ Set<String> groupSet = new HashSet<>(groupList);
+ // Remove all share group ids which have errored out
+ // when deleting with persister.
+ errGroupIds.forEach(groupSet::remove);
+
+ // If no non-share groupIds or non-error share group ids present
+ // return.
+ if (groupSet.isEmpty()) {
+ return CompletableFuture.completedFuture(collection);
+ }
+
+ // Let us invoke the standard procedure of any non-share
+ // groups or successfully deleted share groups remaining.
+ List<String> retainedGroupIds = groupSet.stream().toList();
+ return runtime.scheduleWriteOperation(
+ "delete-groups",
+ topicPartition,
+ Duration.ofMillis(config.offsetCommitTimeoutMs()),
+ coordinator -> coordinator.deleteGroups(context, retainedGroupIds)
+ ).thenApply(deletedCollection -> {
+ deletedCollection.forEach(item -> collection.add(item.duplicate()));
+ return collection;
+ })
+ .exceptionally(exception -> handleOperationException(
+ "delete-groups",
+ groupList,
+ exception,
+ (error, __) -> {
+ DeleteGroupsRequest.getErrorResultCollection(retainedGroupIds, error).forEach(item -> collection.add(item.duplicate()));
+ return collection;
Review Comment:
I don't like the fact that we add to collection in two places (L873 and L881). This is error prone. Could we handle it at the end for both?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
+ .thenCompose(groupErrMap -> {
+ DeleteGroupsResponseData.DeletableGroupResultCollection collection = new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ List<String> errGroupIds = new ArrayList<>();
+ groupErrMap.forEach((groupId, error) -> {
+ if (error.code() != Errors.NONE.code()) {
+ log.error("Error deleting share group {} due to error {}", groupId, error);
+ errGroupIds.add(groupId);
+ collection.add(
+ new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(groupId)
+ .setErrorCode(error.code())
+ );
+ }
+ });
+
+ Set<String> groupSet = new HashSet<>(groupList);
+ // Remove all share group ids which have errored out
+ // when deleting with persister.
+ errGroupIds.forEach(groupSet::remove);
+
+ // If no non-share groupIds or non-error share group ids present
+ // return.
+ if (groupSet.isEmpty()) {
+ return CompletableFuture.completedFuture(collection);
+ }
+
+ // Let us invoke the standard procedure of any non-share
+ // groups or successfully deleted share groups remaining.
+ List<String> retainedGroupIds = groupSet.stream().toList();
+ return runtime.scheduleWriteOperation(
+ "delete-groups",
Review Comment:
nit: Indentation is incorrect here too.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -16294,6 +16299,76 @@ barTopicName, new TopicMetadata(barTopicId, barTopicName, 3)
);
}
+ @Test
+ public void testSharePartitionKeyMap() {
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ assignor.prepareGroupAssignment(new GroupAssignment(Collections.emptyMap()));
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .build();
+
+ MetadataImage image = mock(MetadataImage.class);
+ TopicsImage topicsImage = mock(TopicsImage.class);
+ TopicImage t1image = mock(TopicImage.class);
+ TopicImage t2image = mock(TopicImage.class);
+ when(topicsImage.getTopic(anyString()))
+ .thenReturn(t1image)
+ .thenReturn(t2image);
+
+ ShareGroup shareGroup = mock(ShareGroup.class);
+ when(shareGroup.subscribedTopicNames())
+ .thenReturn(Map.of(
+ "t1", mock(SubscriptionCount.class),
+ "t2", mock(SubscriptionCount.class)
+ )
+ );
Review Comment:
nit: I would format it as follow:
```
when(shareGroup.subscribedTopicNames()).thenReturn(Map.of(
"t1", mock(SubscriptionCount.class),
"t2", mock(SubscriptionCount.class)
));
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -1627,6 +1637,374 @@ public void testDeleteGroups() throws Exception {
assertEquals(expectedResultCollection, future.get());
}
+ @Test
+ public void testDeleteWithShareGroups() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ Persister persister = mock(Persister.class);
+ GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ .setConfig(createConfig())
+ .setRuntime(runtime)
+ .setMetrics(mock(GroupCoordinatorMetrics.class))
+ .setPersister(persister)
+ .build();
+ service.startup(() -> 3);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection1 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // share group
+ DeleteGroupsResponseData.DeletableGroupResult result1 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("share-group-id-1");
+ resultCollection1.add(result1);
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection resultCollection2 =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ // non-share group
+ DeleteGroupsResponseData.DeletableGroupResult result2 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId("group-id-2");
+ resultCollection2.add(result2);
+
+ // null
+ DeleteGroupsResponseData.DeletableGroupResult result3 = new DeleteGroupsResponseData.DeletableGroupResult()
+ .setGroupId(null)
+ .setErrorCode(Errors.INVALID_GROUP_ID.code());
+
+ DeleteGroupsResponseData.DeletableGroupResultCollection expectedResultCollection =
+ new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ expectedResultCollection.addAll(List.of(
+ result3.duplicate(),
+ result2.duplicate(),
+ result1.duplicate()
+ )
+ );
Review Comment:
nit:
```
expectedResultCollection.addAll(List.of(
result3.duplicate(),
result2.duplicate(),
result1.duplicate()
));
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -16294,6 +16299,76 @@ barTopicName, new TopicMetadata(barTopicId, barTopicName, 3)
);
}
+ @Test
+ public void testSharePartitionKeyMap() {
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ assignor.prepareGroupAssignment(new GroupAssignment(Collections.emptyMap()));
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .build();
+
+ MetadataImage image = mock(MetadataImage.class);
+ TopicsImage topicsImage = mock(TopicsImage.class);
+ TopicImage t1image = mock(TopicImage.class);
+ TopicImage t2image = mock(TopicImage.class);
+ when(topicsImage.getTopic(anyString()))
+ .thenReturn(t1image)
+ .thenReturn(t2image);
Review Comment:
It may be easier to just build the MetadataImage that you need rather than mocking it. We have `MetadataImageBuilder` which is pretty handy for it.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -845,6 +896,113 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
(accumulator, newResults) -> newResults.forEach(result -> accumulator.add(result.duplicate())));
}
+ private CompletableFuture<Map<String, Errors>> deleteShareGroups(TopicPartition topicPartition, List<String> groupList) {
+ // topicPartition refers to internal topic __consumer_offsets
+ return runtime.scheduleReadOperation(
+ "delete-share-groups",
+ topicPartition,
+ (coordinator, offset) -> coordinator.sharePartitions(groupList, offset)
+ )
+ .thenCompose(this::performShareGroupsDeletion)
+ .exceptionally(exception -> handleOperationException(
+ "delete-share-groups",
+ groupList,
+ exception,
+ (error, __) -> {
+ Map<String, Errors> errors = new HashMap<>();
+ groupList.forEach(group -> errors.put(group, error));
+ return errors;
+ },
+ log
+ ));
+ }
+
+ private CompletableFuture<Map<String, Errors>> performShareGroupsDeletion(
+ Map<String, Map<Uuid, List<Integer>>> keys
+ ) {
+ List<CompletableFuture<AbstractMap.SimpleEntry<String, DeleteShareGroupStateResult>>> futures = new ArrayList<>();
+ for (Map.Entry<String, Map<Uuid, List<Integer>>> groupEntry : keys.entrySet()) {
+ List<TopicData<PartitionIdData>> topicData = new ArrayList<>();
+ for (Map.Entry<Uuid, List<Integer>> topicEntry : groupEntry.getValue().entrySet()) {
+ topicData.add(
+ new TopicData<>(
+ topicEntry.getKey(),
+ topicEntry.getValue().stream().map(PartitionFactory::newPartitionIdData).toList()
+ )
+ );
+ }
+
+ futures.add(deleteShareGroup(groupEntry.getKey(), topicData));
+ }
+
+ return persisterDeleteToGroupIdErrorMap(futures);
+ }
+
+ private CompletableFuture<AbstractMap.SimpleEntry<String, DeleteShareGroupStateResult>> deleteShareGroup(
+ String groupId,
+ List<TopicData<PartitionIdData>> topicData
+ ) {
+ return persister.deleteState(
+ new DeleteShareGroupStateParameters.Builder()
+ .setGroupTopicPartitionData(new GroupTopicPartitionData.Builder<PartitionIdData>()
+ .setGroupId(groupId)
+ .setTopicsData(topicData)
+ .build()
+ )
+ .build()
+ )
+ .thenCompose(result -> CompletableFuture.completedFuture(new AbstractMap.SimpleEntry<>(groupId, result)))
+ .exceptionally(exception -> {
+ // In case the deleteState call fails,
+ // we should construct the appropriate response here
+ // so that the subsequent callbacks don't see runtime exceptions.
+ log.error("Unable to delete share group partition(s) - {}, {}", groupId, topicData);
+ List<TopicData<PartitionErrorData>> respTopicData = topicData.stream()
+ .map(reqTopicData -> new TopicData<>(
+ reqTopicData.topicId(),
+ reqTopicData.partitions().stream()
+ .map(reqPartData -> {
+ Errors err = Errors.forException(exception);
+ return PartitionFactory.newPartitionErrorData(reqPartData.partition(), err.code(), err.message());
+ })
+ .toList()
+ )
+ )
+ .toList();
+
+ return new AbstractMap.SimpleEntry<>(groupId, new DeleteShareGroupStateResult.Builder()
+ .setTopicsData(respTopicData)
+ .build()
+ );
+ });
+ }
+
+ private CompletableFuture<Map<String, Errors>> persisterDeleteToGroupIdErrorMap(
+ List<CompletableFuture<AbstractMap.SimpleEntry<String, DeleteShareGroupStateResult>>> futures
+ ) {
+ return CompletableFuture.allOf(futures.toArray(new CompletableFuture[]{}))
+ .thenCompose(v -> {
Review Comment:
nit: We would usually format it as follow:
```
return CompletableFuture.allOf(futures.toArray(new CompletableFuture[]{})).thenCompose(v -> {
...
});
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -16294,6 +16299,76 @@ barTopicName, new TopicMetadata(barTopicId, barTopicName, 3)
);
}
+ @Test
+ public void testSharePartitionKeyMap() {
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ assignor.prepareGroupAssignment(new GroupAssignment(Collections.emptyMap()));
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .build();
+
+ MetadataImage image = mock(MetadataImage.class);
+ TopicsImage topicsImage = mock(TopicsImage.class);
+ TopicImage t1image = mock(TopicImage.class);
+ TopicImage t2image = mock(TopicImage.class);
+ when(topicsImage.getTopic(anyString()))
+ .thenReturn(t1image)
+ .thenReturn(t2image);
+
+ ShareGroup shareGroup = mock(ShareGroup.class);
+ when(shareGroup.subscribedTopicNames())
+ .thenReturn(Map.of(
+ "t1", mock(SubscriptionCount.class),
+ "t2", mock(SubscriptionCount.class)
+ )
+ );
+
+ when(shareGroup.groupId())
+ .thenReturn("share-group");
+ when(image.topics())
+ .thenReturn(topicsImage);
+ when(image.provenance())
+ .thenReturn(new MetadataProvenance(-1, -1, -1, true));
+
+ when(t1image.partitions())
+ .thenReturn(
+ Map.of(
+ 0, mock(PartitionRegistration.class),
+ 1, mock(PartitionRegistration.class)
+ )
+ );
+ Uuid t1Uuid = Uuid.randomUuid();
+ when(t1image.id()).thenReturn(t1Uuid);
+
+ when(t2image.partitions())
+ .thenReturn(
+ Map.of(
+ 0, mock(PartitionRegistration.class),
+ 1, mock(PartitionRegistration.class)
+ )
+ );
+ Uuid t2Uuid = Uuid.randomUuid();
+ when(t2image.id()).thenReturn(t2Uuid);
+
+ context.groupMetadataManager.onNewMetadataImage(image, mock(MetadataDelta.class));
+ Map<String, Map<Uuid, List<Integer>>> keyMap = context.groupMetadataManager.sharePartitionKeysMap(List.of(shareGroup));
+ assertEquals(1, keyMap.size());
+ assertEquals(2, keyMap.get("share-group").size());
+ for (Uuid topic : List.of(t1Uuid, t2Uuid)) {
+ assertEquals(2, keyMap.get("share-group").get(topic).size());
+ assertTrue(keyMap.get("share-group").get(topic).contains(0));
+ assertTrue(keyMap.get("share-group").get(topic).contains(1));
+ }
Review Comment:
We usually prefer to build the expected map and to use assertEquals(expectedMap, actualMap).
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1954243251
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,21 +834,60 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
-
- futures.add(future);
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> shareFuture = deleteShareGroups(topicPartition, groupList)
+ .thenCompose(groupErrMap -> {
+ DeleteGroupsResponseData.DeletableGroupResultCollection collection = new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ List<String> errGroupIds = new ArrayList<>();
+ groupErrMap.forEach((groupId, error) -> {
+ if (error.code() != Errors.NONE.code()) {
Review Comment:
Yes, delete responses use the coordinator-common handleOperationException 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #18848:
URL: https://github.com/apache/kafka/pull/18848#issuecomment-2656619100
> And this is a shortcoming in the KIP I think. Should we include GroupType in a new version of the DeleteGroups RPC? It's unfortunate that this code doesn't know the group type being deleted. It could do. lmk if you want me to address this in the KIP.
It would be unfortunate to require the caller to specify the group type. Knowing the group id is everything that the server needs to know in the end.
> The deletion of the share group state is scheduled before the call to GroupCoordinatorShard.deleteGroups which actually checks the group type and whether the group is empty.
I think that we could consider restructuring the code to start by fetching the groups from the state machines (shards). The response would include whether the groups are eligible to be deleted and their type. Then we could proceed with the flow based on the group type. This would resolve both points.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18629: Delete share group state RPC group coordinator impl. [3/N] [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18848:
URL: https://github.com/apache/kafka/pull/18848#discussion_r1956632577
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -822,20 +831,27 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
});
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
- CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future =
- runtime.scheduleWriteOperation(
- "delete-groups",
- topicPartition,
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.deleteGroups(context, groupList)
- ).exceptionally(exception -> handleOperationException(
- "delete-groups",
- groupList,
- exception,
- (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupList, error),
- log
- ));
+ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future = deleteShareGroups(topicPartition, groupList).thenCompose(groupErrMap -> {
+ DeleteGroupsResponseData.DeletableGroupResultCollection collection = new DeleteGroupsResponseData.DeletableGroupResultCollection();
+ List<String> retainedGroupIds = deleteCandidateGroupIds(groupErrMap, groupList, collection);
+ if (retainedGroupIds.isEmpty()) {
+ return CompletableFuture.completedFuture(collection);
+ }
+ return handleDeleteGroups(context, topicPartition, retainedGroupIds)
+ .whenComplete((resp, __) -> resp.forEach(result -> collection.add(result.duplicate())))
+ .thenApply(__ -> collection);
+ });
+ // deleteShareGroups has its own exceptionally block, so we don't need one here.
+
+ // This future object has the following stages:
+ // - First it invokes the share group delete flow where the shard sharePartitionDeleteRequests
+ // method is invoked, and it returns request objects for each valid share group passed to it.
+ // - Then the requests are passed to the persister.deleteStata method one at a time. The results
+ // are collated as a Map of groupId -> persister errors
+ // - The above map is then used to decide whether to invoke the group coordinator delete groups logic
+ // - Share groups with failed persister delete are NOT CONSIDERED for group coordinator delete.
+ // TLDR: DeleteShareGroups -> filter erroneous persister deletes -> Original delete groups logic
Review Comment:
I think "original" is going to date quickly. I think "general" might be better.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -6280,6 +6286,48 @@ public void createGroupTombstoneRecords(
group.createGroupTombstoneRecords(records);
}
+ /**
+ * Returns an optional of delete share group request object to be used with the persister.
Review Comment:
This doesn't really do the method justice. It seems to me that the salient points are:
* It checks whether the share group is empty. If it is not empty, the return value is empty.
* If there are no subscribed topics, again the return value is empty.
* Otherwise, it returns the parameters required by the persister to delete the share-partition state.
I would rename the method to something like `shareGroupCheckEmptyAndBuildPartitionDeleteRequest`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -6280,6 +6286,48 @@ public void createGroupTombstoneRecords(
group.createGroupTombstoneRecords(records);
}
+ /**
+ * Returns an optional of delete share group request object to be used with the persister.
Review Comment:
It could even throw `GroupNotEmptyException` if the group is not empty. That seems like the cleanest way out of this deletion process for a non-empty share group.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org