KAFKA-18629: Utilize share group partition metadata for delete group. (apache#19363)

* Currently, the delete share group code flow uses
`group.subscribedTopicNames()` to fetch information about all the share
partitions to which a share group is subscribed to.
* However, this is incorrect since once the group is EMPTY, a
precondition for delete, the aforementioned method will return an empty
list.
* In this PR we have leveraged the `ShareGroupStatePartitionMetadata`
record to grab the `initialized` and `initializing` partitions to build
the delete candidates, which remedies the situation.

Reviewers: Andrew Schofield <[email protected]>

Author: smjn

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java

@@ -814,6 +814,21 @@ public static CoordinatorRecord newShareGroupCurrentAssignmentTombstoneRecord(
         );
     }
 
+    /**
+     * Creates a ShareGroupStatePartitionMetadata tombstone.
+     *
+     * @param groupId   The share group id.
+     * @return The record.
+     */
+    public static CoordinatorRecord newShareGroupStatePartitionMetadataTombstoneRecord(
+        String groupId
+    ) {
+        return CoordinatorRecord.tombstone(
+            new ShareGroupStatePartitionMetadataKey()
+                .setGroupId(groupId)
+        );
+    }
+
     /**
      * Creates a ShareGroupStatePartitionMetadata record.
      *

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java

@@ -1073,24 +1073,25 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
 
         groupsByTopicPartition.forEach((topicPartition, groupList) -> {
             CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future = deleteShareGroups(topicPartition, groupList).thenCompose(groupErrMap -> {
-                DeleteGroupsResponseData.DeletableGroupResultCollection collection = new DeleteGroupsResponseData.DeletableGroupResultCollection();
-                List<String> retainedGroupIds = deleteCandidateGroupIds(groupErrMap, groupList, collection);
+                DeleteGroupsResponseData.DeletableGroupResultCollection deletableGroupResults = new DeleteGroupsResponseData.DeletableGroupResultCollection();
+                List<String> retainedGroupIds = updateResponseAndGetNonErrorGroupList(groupErrMap, groupList, deletableGroupResults);
                 if (retainedGroupIds.isEmpty()) {
-                    return CompletableFuture.completedFuture(collection);
+                    return CompletableFuture.completedFuture(deletableGroupResults);
                 }
 
                 return handleDeleteGroups(context, topicPartition, retainedGroupIds)
-                    .whenComplete((resp, __) -> resp.forEach(result -> collection.add(result.duplicate())))
-                    .thenApply(__ -> collection);
+                    .whenComplete((resp, __) -> resp.forEach(result -> deletableGroupResults.add(result.duplicate())))
+                    .thenApply(__ -> deletableGroupResults);
             });
             // 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.
+            // All initialized and initializing share partitions are moved to deleting.
             // - Then the requests are passed to the persister.deleteState 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
+            // - The above map can be 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 -> general delete groups logic
             futures.add(future);
@@ -1102,17 +1103,26 @@ public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection
             (accumulator, newResults) -> newResults.forEach(result -> accumulator.add(result.duplicate())));
     }
 
-    private List<String> deleteCandidateGroupIds(
-        Map<String, Errors> groupErrMap,
+    /**
+     * Processes input shareGroupErrMap by retaining only those which do not contain an error.
+     * Also updates the result collection input arg with share groups containing errors.
+     *
+     * @param shareGroupErrMap      Map keyed on share groupId and value as the error (NONE for no error).
+     * @param groupList             Entire list of groups (all types)
+     * @param deletableGroupResults Collection of responses for delete groups request.
+     * @return A list of all non-error groupIds
+     */
+    private List<String> updateResponseAndGetNonErrorGroupList(
+        Map<String, Errors> shareGroupErrMap,
         List<String> groupList,
-        DeleteGroupsResponseData.DeletableGroupResultCollection collection
+        DeleteGroupsResponseData.DeletableGroupResultCollection deletableGroupResults
     ) {
         List<String> errGroupIds = new ArrayList<>();
-        groupErrMap.forEach((groupId, error) -> {
+        shareGroupErrMap.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(
+                deletableGroupResults.add(
                     new DeleteGroupsResponseData.DeletableGroupResult()
                         .setGroupId(groupId)
                         .setErrorCode(error.code())
@@ -1153,7 +1163,7 @@ private CompletableFuture<Map<String, Errors>> deleteShareGroups(
         TopicPartition topicPartition,
         List<String> groupList
     ) {
-        // topicPartition refers to internal topic __consumer_offsets
+        // topicPartition refers to internal topic __consumer_offsets.
         return runtime.scheduleWriteOperation(
             "delete-share-groups",
             topicPartition,

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java

@@ -589,18 +589,19 @@ public CoordinatorResult<DeleteGroupsResponseData.DeletableGroupResultCollection
      * Method returns a Map keyed on groupId and value as pair of {@link DeleteShareGroupStateParameters}
      * and any ERRORS while building the request corresponding
      * to the valid share groups passed as the input.
-     * <p></p>
+     * <p>
      * The groupIds are first filtered by type to restrict the list to share groups.
      * @param groupIds - A list of groupIds as string
-     * @return {@link CoordinatorResult} object always containing empty records and Map keyed on groupId and value pair (req, error)
+     * @return A result object containing a map keyed on groupId and value pair (req, error) and related coordinator records.
      */
     public CoordinatorResult<Map<String, Map.Entry<DeleteShareGroupStateParameters, Errors>>, CoordinatorRecord> sharePartitionDeleteRequests(List<String> groupIds) {
         Map<String, Map.Entry<DeleteShareGroupStateParameters, Errors>> responseMap = new HashMap<>();
+        List<CoordinatorRecord> records = new ArrayList<>();
         for (String groupId : groupIds) {
             try {
                 ShareGroup group = groupMetadataManager.shareGroup(groupId);
                 group.validateDeleteGroup();
-                groupMetadataManager.shareGroupBuildPartitionDeleteRequest(group)
+                groupMetadataManager.shareGroupBuildPartitionDeleteRequest(groupId, records)
                     .ifPresent(req -> responseMap.put(groupId, Map.entry(req, Errors.NONE)));
             } catch (GroupIdNotFoundException exception) {
                 log.debug("GroupId {} not found as a share group.", groupId);
@@ -609,7 +610,7 @@ public CoordinatorResult<Map<String, Map.Entry<DeleteShareGroupStateParameters,
                 responseMap.put(groupId, Map.entry(DeleteShareGroupStateParameters.EMPTY_PARAMS, Errors.forException(exception)));
             }
         }
-        return new CoordinatorResult<>(List.of(), responseMap);
+        return new CoordinatorResult<>(records, responseMap);
     }
 
     /**

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

@@ -2984,7 +2984,7 @@ private void addInitializingTopicsRecords(String groupId, List<CoordinatorRecord
                 groupId,
                 attachTopicName(finalInitializingMap),
                 attachTopicName(currentMap.initializedTopics()),
-                Map.of()
+                attachTopicName(currentMap.deletingTopics())
             )
         );
     }
@@ -4979,7 +4979,7 @@ public CoordinatorResult<Void, CoordinatorRecord> initializeShareGroupState(
                 group.groupId(),
                 attachTopicName(finalInitializingMap),
                 attachTopicName(finalInitializedMap),
-                Map.of()
+                attachTopicName(currentMap.deletingTopics())
             )),
             null
         );
@@ -5025,7 +5025,7 @@ public CoordinatorResult<Void, CoordinatorRecord> uninitializeShareGroupState(
                     groupId,
                     attachTopicName(finalInitializingTopics),
                     attachTopicName(info.initializedTopics()),
-                    Map.of()
+                    attachTopicName(info.deletingTopics())
                 )
             ),
             null
@@ -5057,6 +5057,13 @@ public List<InitializeShareGroupStateParameters> reconcileShareGroupStateInitial
         return requests;
     }
 
+    private Map<Uuid, String> attachTopicName(Set<Uuid> topicIds) {
+        TopicsImage topicsImage = metadataImage.topics();
+        return topicIds.stream()
+            .map(topicId -> Map.entry(topicId, topicsImage.getTopic(topicId).name()))
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    }
+
     private Map<Uuid, Map.Entry<String, Set<Integer>>> attachTopicName(Map<Uuid, Set<Integer>> initMap) {
         TopicsImage topicsImage = metadataImage.topics();
         Map<Uuid, Map.Entry<String, Set<Integer>>> finalMap = new HashMap<>();
@@ -8142,39 +8149,49 @@ public void createGroupTombstoneRecords(
     /**
      * 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
+     * @param shareGroupId      Share group id
+     * @param records           List of coordinator records to append to
      * @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();
-        List<TopicData<PartitionIdData>> topicDataList = new ArrayList<>(subscribedTopics.size());
-
-        for (String topic : subscribedTopics) {
-            TopicImage topicImage = topicsImage.getTopic(topic);
-            topicDataList.add(
-                new TopicData<>(
-                    topicImage.id(),
-                    topicImage.partitions().keySet().stream()
-                        .map(PartitionFactory::newPartitionIdData)
-                        .toList()
-                )
-            );
+    public Optional<DeleteShareGroupStateParameters> shareGroupBuildPartitionDeleteRequest(String shareGroupId, List<CoordinatorRecord> records) {
+        if (!shareGroupPartitionMetadata.containsKey(shareGroupId)) {
+            return Optional.empty();
         }
 
-        if (topicDataList.isEmpty()) {
+        Map<Uuid, Set<Integer>> deleteCandidates = mergeShareGroupInitMaps(
+            shareGroupPartitionMetadata.get(shareGroupId).initializedTopics(),
+            shareGroupPartitionMetadata.get(shareGroupId).initializingTopics()
+        );
+
+        if (deleteCandidates.isEmpty()) {
             return Optional.empty();
         }
 
-        return Optional.of(
-            new DeleteShareGroupStateParameters.Builder()
-                .setGroupTopicPartitionData(
-                    new GroupTopicPartitionData.Builder<PartitionIdData>()
-                        .setGroupId(shareGroup.groupId())
-                        .setTopicsData(topicDataList)
-                        .build()
-                )
-                .build()
+        List<TopicData<PartitionIdData>> topicDataList = new ArrayList<>(deleteCandidates.size());
+
+        for (Map.Entry<Uuid, Set<Integer>> entry : deleteCandidates.entrySet()) {
+            topicDataList.add(new TopicData<>(
+                entry.getKey(),
+                entry.getValue().stream()
+                    .map(PartitionFactory::newPartitionIdData)
+                    .toList()
+            ));
+        }
+
+        // Remove all initializing and initialized topic info from record and add deleting.
+        records.add(GroupCoordinatorRecordHelpers.newShareGroupStatePartitionMetadataRecord(
+            shareGroupId,
+            Map.of(),
+            Map.of(),
+            attachTopicName(deleteCandidates.keySet())
+        ));
+
+        return Optional.of(new DeleteShareGroupStateParameters.Builder()
+            .setGroupTopicPartitionData(new GroupTopicPartitionData.Builder<PartitionIdData>()
+                .setGroupId(shareGroupId)
+                .setTopicsData(topicDataList)
+                .build())
+            .build()
         );
     }
 

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java

@@ -236,6 +236,7 @@ public void createGroupTombstoneRecords(List<CoordinatorRecord> records) {
         );
 
         records.add(GroupCoordinatorRecordHelpers.newShareGroupSubscriptionMetadataTombstoneRecord(groupId()));
+        records.add(GroupCoordinatorRecordHelpers.newShareGroupStatePartitionMetadataTombstoneRecord(groupId()));
         records.add(GroupCoordinatorRecordHelpers.newShareGroupEpochTombstoneRecord(groupId()));
     }
 

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/Assertions.java

@@ -31,6 +31,7 @@
 import org.apache.kafka.coordinator.group.generated.ConsumerGroupTargetAssignmentMemberValue;
 import org.apache.kafka.coordinator.group.generated.GroupMetadataValue;
 import org.apache.kafka.coordinator.group.generated.ShareGroupPartitionMetadataValue;
+import org.apache.kafka.coordinator.group.generated.ShareGroupStatePartitionMetadataValue;
 import org.apache.kafka.server.common.ApiMessageAndVersion;
 
 import org.opentest4j.AssertionFailedError;
@@ -60,7 +61,8 @@ public class Assertions {
         ConsumerGroupPartitionMetadataValue.class, Assertions::assertConsumerGroupPartitionMetadataValue,
         GroupMetadataValue.class, Assertions::assertGroupMetadataValue,
         ConsumerGroupTargetAssignmentMemberValue.class, Assertions::assertConsumerGroupTargetAssignmentMemberValue,
-        ShareGroupPartitionMetadataValue.class, Assertions::assertShareGroupPartitionMetadataValue
+        ShareGroupPartitionMetadataValue.class, Assertions::assertShareGroupPartitionMetadataValue,
+        ShareGroupStatePartitionMetadataValue.class, Assertions::assertShareGroupStatePartitionMetadataValue
     );
 
     public static void assertResponseEquals(
@@ -285,6 +287,33 @@ private static void assertShareGroupPartitionMetadataValue(
         assertEquals(expected, actual);
     }
 
+    private static void assertShareGroupStatePartitionMetadataValue(
+        ApiMessage exp,
+        ApiMessage act
+    ) {
+        ShareGroupStatePartitionMetadataValue expected = (ShareGroupStatePartitionMetadataValue) exp.duplicate();
+        ShareGroupStatePartitionMetadataValue actual = (ShareGroupStatePartitionMetadataValue) act.duplicate();
+
+        Consumer<ShareGroupStatePartitionMetadataValue> normalize = message -> {
+            message.initializedTopics().sort(Comparator.comparing(ShareGroupStatePartitionMetadataValue.TopicPartitionsInfo::topicId));
+            message.initializedTopics().forEach(topic -> {
+                topic.partitions().sort(Comparator.naturalOrder());
+            });
+
+            message.initializingTopics().sort(Comparator.comparing(ShareGroupStatePartitionMetadataValue.TopicPartitionsInfo::topicId));
+            message.initializingTopics().forEach(topic -> {
+                topic.partitions().sort(Comparator.naturalOrder());
+            });
+
+            message.deletingTopics().sort(Comparator.comparing(ShareGroupStatePartitionMetadataValue.TopicInfo::topicId));
+        };
+
+        normalize.accept(expected);
+        normalize.accept(actual);
+
+        assertEquals(expected, actual);
+    }
+
     private static void assertGroupMetadataValue(
         ApiMessage exp,
         ApiMessage act