KAFKA-19115: Utilize initialized topics info to verify delete share group offsets (#19431)

Currently, in the deleteShareGroupOffsets method in
GroupCoordinatorService, the user request was simply forwarded to the
persister without checking if the requested share partitions were
initialized for the group or not. This PR introduces such a check to
make sure that the persister deleteState request is only called for
share partitions that have been initialized for the group.

Reviewers: Andrew Schofield <[email protected]>, Sushant Mahajan <[email protected]>

Author: chirag-wadhwa5

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

@@ -28,7 +28,6 @@
 import org.apache.kafka.common.message.DeleteGroupsResponseData;
 import org.apache.kafka.common.message.DeleteShareGroupOffsetsRequestData;
 import org.apache.kafka.common.message.DeleteShareGroupOffsetsResponseData;
-import org.apache.kafka.common.message.DeleteShareGroupStateRequestData;
 import org.apache.kafka.common.message.DescribeGroupsResponseData;
 import org.apache.kafka.common.message.DescribeShareGroupOffsetsRequestData;
 import org.apache.kafka.common.message.DescribeShareGroupOffsetsResponseData;
@@ -1258,45 +1257,40 @@ private CompletableFuture<Map<String, Errors>> persisterDeleteToGroupIdErrorMap(
         });
     }
 
-    private void populateDeleteShareGroupOffsetsFuture(
-        DeleteShareGroupOffsetsRequestData requestData,
-        CompletableFuture<DeleteShareGroupOffsetsResponseData> future,
-        Map<Uuid, String> requestTopicIdToNameMapping,
-        List<DeleteShareGroupStateRequestData.DeleteStateData> deleteShareGroupStateRequestTopicsData,
-        List<DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic> deleteShareGroupOffsetsResponseTopicList
-
+    private CompletableFuture<DeleteShareGroupOffsetsResponseData> persistDeleteShareGroupOffsets(
+        DeleteShareGroupStateParameters deleteStateRequestParameters,
+        List<DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic> errorTopicResponseList
     ) {
-        DeleteShareGroupStateRequestData deleteShareGroupStateRequestData = new DeleteShareGroupStateRequestData()
-            .setGroupId(requestData.groupId())
-            .setTopics(deleteShareGroupStateRequestTopicsData);
-
-        persister.deleteState(DeleteShareGroupStateParameters.from(deleteShareGroupStateRequestData))
-            .whenComplete((result, error) -> {
-                if (error != null) {
-                    log.error("Failed to delete share group state");
-                    future.completeExceptionally(error);
-                    return;
-                }
+        return persister.deleteState(deleteStateRequestParameters)
+            .thenCompose(result -> {
                 if (result == null || result.topicsData() == null) {
                     log.error("Result is null for the delete share group state");
-                    future.completeExceptionally(new IllegalStateException("Result is null for the delete share group state"));
-                    return;
+                    Exception exception = new IllegalStateException("Result is null for the delete share group state");
+                    return CompletableFuture.completedFuture(
+                        DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(Errors.forException(exception))
+                    );
                 }
                 result.topicsData().forEach(topicData ->
-                    deleteShareGroupOffsetsResponseTopicList.add(new DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic()
-                        .setTopicId(topicData.topicId())
-                        .setTopicName(requestTopicIdToNameMapping.get(topicData.topicId()))
-                        .setPartitions(topicData.partitions().stream().map(
-                            partitionData -> new DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponsePartition()
-                                .setPartitionIndex(partitionData.partition())
-                                .setErrorMessage(partitionData.errorCode() == Errors.NONE.code() ? null : Errors.forCode(partitionData.errorCode()).message())
-                                .setErrorCode(partitionData.errorCode())
-                        ).toList())
-                    ));
+                    errorTopicResponseList.add(
+                        new DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic()
+                            .setTopicId(topicData.topicId())
+                            .setTopicName(metadataImage.topics().topicIdToNameView().get(topicData.topicId()))
+                            .setPartitions(topicData.partitions().stream().map(
+                                partitionData -> new DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponsePartition()
+                                    .setPartitionIndex(partitionData.partition())
+                                    .setErrorMessage(partitionData.errorCode() == Errors.NONE.code() ? null : Errors.forCode(partitionData.errorCode()).message())
+                                    .setErrorCode(partitionData.errorCode())
+                            ).toList())
+                    )
+                );
 
-                future.complete(
+                return CompletableFuture.completedFuture(
                     new DeleteShareGroupOffsetsResponseData()
-                        .setResponses(deleteShareGroupOffsetsResponseTopicList));
+                        .setResponses(errorTopicResponseList)
+                );
+            }).exceptionally(throwable -> {
+                log.error("Failed to delete share group state");
+                return DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(Errors.forException(throwable));
             });
     }
 
@@ -1590,83 +1584,53 @@ public CompletableFuture<DeleteShareGroupOffsetsResponseData> deleteShareGroupOf
                 DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(Errors.INVALID_GROUP_ID));
         }
 
-        Map<Uuid, String> requestTopicIdToNameMapping = new HashMap<>();
-        List<DeleteShareGroupStateRequestData.DeleteStateData> deleteShareGroupStateRequestTopicsData = new ArrayList<>(requestData.topics().size());
-        List<DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic> deleteShareGroupOffsetsResponseTopicList = new ArrayList<>(requestData.topics().size());
-
-        requestData.topics().forEach(topic -> {
-            Uuid topicId = metadataImage.topics().topicNameToIdView().get(topic.topicName());
-            if (topicId != null) {
-                requestTopicIdToNameMapping.put(topicId, topic.topicName());
-                deleteShareGroupStateRequestTopicsData.add(new DeleteShareGroupStateRequestData.DeleteStateData()
-                    .setTopicId(topicId)
-                    .setPartitions(
-                        topic.partitions().stream().map(
-                            partitionIndex -> new DeleteShareGroupStateRequestData.PartitionData().setPartition(partitionIndex)
-                        ).toList()
-                    ));
-            } else {
-                deleteShareGroupOffsetsResponseTopicList.add(new DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic()
-                    .setTopicName(topic.topicName())
-                    .setPartitions(topic.partitions().stream().map(
-                        partition -> new DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponsePartition()
-                            .setPartitionIndex(partition)
-                            .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code())
-                            .setErrorMessage(Errors.UNKNOWN_TOPIC_OR_PARTITION.message())
-                    ).toList()));
-            }
-        });
-
-        // If the request for the persister is empty, just complete the operation right away.
-        if (deleteShareGroupStateRequestTopicsData.isEmpty()) {
+        if (requestData.topics() == null || requestData.topics().isEmpty()) {
             return CompletableFuture.completedFuture(
                 new DeleteShareGroupOffsetsResponseData()
-                    .setResponses(deleteShareGroupOffsetsResponseTopicList));
+            );
         }
 
-        CompletableFuture<DeleteShareGroupOffsetsResponseData> future = new CompletableFuture<>();
+        return runtime.scheduleReadOperation(
+            "share-group-delete-offsets-request",
+            topicPartitionFor(groupId),
+            (coordinator, lastCommittedOffset) -> coordinator.shareGroupDeleteOffsetsRequest(groupId, requestData)
+        )
+            .thenCompose(resultHolder -> {
+                if (resultHolder == null) {
+                    log.error("Failed to retrieve deleteState request parameters from group coordinator for the group {}", groupId);
+                    return CompletableFuture.completedFuture(
+                        DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(Errors.UNKNOWN_SERVER_ERROR)
+                    );
+                }
+
+                if (resultHolder.topLevelErrorCode() != Errors.NONE.code()) {
+                    return CompletableFuture.completedFuture(
+                        DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(
+                            resultHolder.topLevelErrorCode(),
+                            resultHolder.topLevelErrorMessage()
+                        )
+                    );
+                }
 
-        TopicPartition topicPartition = topicPartitionFor(groupId);
+                List<DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic> errorTopicResponseList =
+                    resultHolder.errorTopicResponseList() == null ? new ArrayList<>() : new ArrayList<>(resultHolder.errorTopicResponseList());
 
-        // This is done to make sure the provided group is empty. Offsets can be deleted only for an empty share group.
-        CompletableFuture<List<ShareGroupDescribeResponseData.DescribedGroup>> describeGroupFuture =
-            runtime.scheduleReadOperation(
-                "share-group-describe",
-                topicPartition,
-                (coordinator, lastCommittedOffset) -> coordinator.shareGroupDescribe(List.of(groupId), lastCommittedOffset)
-            ).exceptionally(exception -> handleOperationException(
-                "share-group-describe",
-                List.of(groupId),
-                exception,
-                (error, __) -> ShareGroupDescribeRequest.getErrorDescribedGroupList(List.of(groupId), error),
-                log
-            ));
+                if (resultHolder.deleteStateRequestParameters() == null) {
+                    return CompletableFuture.completedFuture(
+                        new DeleteShareGroupOffsetsResponseData()
+                            .setResponses(errorTopicResponseList)
+                    );
+                }
 
-        describeGroupFuture.whenComplete((groups, throwable) -> {
-            if (throwable != null) {
-                log.error("Failed to describe the share group {}", groupId, throwable);
-                future.complete(DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(Errors.forException(throwable)));
-            } else if (groups == null || groups.isEmpty()) {
-                log.error("Describe share group resulted in empty response for group {}", groupId);
-                future.complete(DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(Errors.GROUP_ID_NOT_FOUND));
-            } else if (groups.get(0).errorCode() != Errors.NONE.code()) {
-                log.error("Failed to describe the share group {}", groupId);
-                future.complete(DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(groups.get(0).errorCode(), groups.get(0).errorMessage()));
-            } else if (groups.get(0).members() != null && !groups.get(0).members().isEmpty()) {
-                log.error("Provided group {} is not empty", groupId);
-                future.complete(DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(Errors.NON_EMPTY_GROUP));
-            } else {
-                populateDeleteShareGroupOffsetsFuture(
-                    requestData,
-                    future,
-                    requestTopicIdToNameMapping,
-                    deleteShareGroupStateRequestTopicsData,
-                    deleteShareGroupOffsetsResponseTopicList
+                return persistDeleteShareGroupOffsets(
+                    resultHolder.deleteStateRequestParameters(),
+                    errorTopicResponseList
                 );
-            }
-        });
-
-        return future;
+            })
+            .exceptionally(throwable -> {
+                log.error("Failed to retrieve deleteState request parameters from group coordinator for the group {}", groupId, throwable);
+                return DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(Errors.forException(throwable));
+            });
     }
 
     /**

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

@@ -26,6 +26,9 @@
 import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
 import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
 import org.apache.kafka.common.message.DeleteGroupsResponseData;
+import org.apache.kafka.common.message.DeleteShareGroupOffsetsRequestData;
+import org.apache.kafka.common.message.DeleteShareGroupOffsetsResponseData;
+import org.apache.kafka.common.message.DeleteShareGroupStateRequestData;
 import org.apache.kafka.common.message.DescribeGroupsResponseData;
 import org.apache.kafka.common.message.HeartbeatRequestData;
 import org.apache.kafka.common.message.HeartbeatResponseData;
@@ -131,6 +134,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -290,6 +294,69 @@ public GroupCoordinatorShard build() {
         }
     }
 
+    public static class DeleteShareGroupOffsetsResultHolder {
+        private final short topLevelErrorCode;
+        private final String topLevelErrorMessage;
+        private final List<DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic> errorTopicResponseList;
+        private final DeleteShareGroupStateParameters deleteStateRequestParameters;
+
+        DeleteShareGroupOffsetsResultHolder(short topLevelErrorCode, String topLevelErrorMessage) {
+            this(topLevelErrorCode, topLevelErrorMessage, null,  null);
+        }
+
+        DeleteShareGroupOffsetsResultHolder(
+            short topLevelErrorCode,
+            String topLevelErrorMessage,
+            List<DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic> errorTopicResponseList
+        ) {
+            this(topLevelErrorCode, topLevelErrorMessage, errorTopicResponseList, null);
+        }
+
+        DeleteShareGroupOffsetsResultHolder(
+            short topLevelErrorCode,
+            String topLevelErrorMessage,
+            List<DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic> errorTopicResponseList,
+            DeleteShareGroupStateParameters deleteStateRequestParameters
+        ) {
+            this.topLevelErrorCode = topLevelErrorCode;
+            this.topLevelErrorMessage = topLevelErrorMessage;
+            this.errorTopicResponseList = errorTopicResponseList;
+            this.deleteStateRequestParameters = deleteStateRequestParameters;
+        }
+
+        public short topLevelErrorCode() {
+            return this.topLevelErrorCode;
+        }
+
+        public String topLevelErrorMessage() {
+            return this.topLevelErrorMessage;
+        }
+
+        public List<DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic> errorTopicResponseList() {
+            return this.errorTopicResponseList;
+        }
+
+        public DeleteShareGroupStateParameters deleteStateRequestParameters() {
+            return this.deleteStateRequestParameters;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            DeleteShareGroupOffsetsResultHolder other = (DeleteShareGroupOffsetsResultHolder) o;
+            return topLevelErrorCode == other.topLevelErrorCode &&
+                Objects.equals(topLevelErrorMessage, other.topLevelErrorMessage) &&
+                Objects.equals(errorTopicResponseList, other.errorTopicResponseList) &&
+                Objects.equals(deleteStateRequestParameters, other.deleteStateRequestParameters);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(topLevelErrorCode, topLevelErrorMessage, errorTopicResponseList, deleteStateRequestParameters);
+        }
+    }
+
     /**
      * The group/offsets expiration key to schedule a timer task.
      *
@@ -613,6 +680,57 @@ public CoordinatorResult<Map<String, Map.Entry<DeleteShareGroupStateParameters,
         return new CoordinatorResult<>(records, responseMap);
     }
 
+    /**
+     * Does the following checks to make sure that a DeleteShareGroupOffsets request is valid and can be processed further
+     * 1. Checks whether the provided group is empty
+     * 2. Checks the requested topics are presented in the metadataImage
+     * 3. Checks the requested share partitions are initialized for the group
+     *
+     * @param groupId - The group ID
+     * @param requestData - The request data for DeleteShareGroupOffsetsRequest
+     * @return {@link DeleteShareGroupOffsetsResultHolder} an object containing top level error code, list of topic responses
+     *                                               and persister deleteState request parameters
+     */
+    public DeleteShareGroupOffsetsResultHolder shareGroupDeleteOffsetsRequest(
+        String groupId,
+        DeleteShareGroupOffsetsRequestData requestData
+    ) {
+        try {
+            ShareGroup group = groupMetadataManager.shareGroup(groupId);
+            group.validateDeleteGroup();
+
+            List<DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic> errorTopicResponseList = new ArrayList<>();
+            List<DeleteShareGroupStateRequestData.DeleteStateData> deleteShareGroupStateRequestTopicsData =
+                groupMetadataManager.sharePartitionsEligibleForOffsetDeletion(
+                    groupId,
+                    requestData,
+                    errorTopicResponseList
+                );
+
+            if (deleteShareGroupStateRequestTopicsData.isEmpty()) {
+                return new DeleteShareGroupOffsetsResultHolder(Errors.NONE.code(), null, errorTopicResponseList);
+            }
+
+            DeleteShareGroupStateRequestData deleteShareGroupStateRequestData = new DeleteShareGroupStateRequestData()
+                .setGroupId(requestData.groupId())
+                .setTopics(deleteShareGroupStateRequestTopicsData);
+
+            return new DeleteShareGroupOffsetsResultHolder(
+                Errors.NONE.code(),
+                null,
+                errorTopicResponseList,
+                DeleteShareGroupStateParameters.from(deleteShareGroupStateRequestData)
+            );
+
+        } catch (GroupIdNotFoundException exception) {
+            log.error("groupId {} not found", groupId, exception);
+            return new DeleteShareGroupOffsetsResultHolder(Errors.GROUP_ID_NOT_FOUND.code(), exception.getMessage());
+        } catch (GroupNotEmptyException exception) {
+            log.error("Provided group {} is not empty", groupId);
+            return new DeleteShareGroupOffsetsResultHolder(Errors.NON_EMPTY_GROUP.code(), exception.getMessage());
+        }
+    }
+
     /**
      * Fetch offsets for a given set of partitions and a given group.
      *