KAFKA-16718-3/n: Added the ShareGroupStatePartitionMetadata record during deletion of share group offsets (#19478)

This is a follow up PR for implementation of DeleteShareGroupOffsets
RPC. This PR adds the ShareGroupStatePartitionMetadata record to
__consumer__offsets topic to make sure the topic is removed from the
initializedTopics list. This PR also removes partitions from the request
and response schemas for DeleteShareGroupState RPC

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

Author: chirag-wadhwa5

clients/src/main/java/org/apache/kafka/clients/admin/Admin.java

@@ -1951,28 +1951,28 @@ default ListShareGroupOffsetsResult listShareGroupOffsets(Map<String, ListShareG
     }
 
     /**
-     * Delete offsets for a set of partitions in a share group.
+     * Delete offsets for a set of topics in a share group.
      *
      * @param groupId The group for which to delete offsets.
-     * @param partitions The topic-partitions.
+     * @param topics The topics for which to delete offsets.
      * @param options The options to use when deleting offsets in a share group.
      * @return The DeleteShareGroupOffsetsResult.
      */
-    DeleteShareGroupOffsetsResult deleteShareGroupOffsets(String groupId, Set<TopicPartition> partitions, DeleteShareGroupOffsetsOptions options);
+    DeleteShareGroupOffsetsResult deleteShareGroupOffsets(String groupId, Set<String> topics, DeleteShareGroupOffsetsOptions options);
 
     /**
-     * Delete offsets for a set of partitions in a share group with the default options.
+     * Delete offsets for a set of topics in a share group with the default options.
      *
      * <p>
      * This is a convenience method for {@link #deleteShareGroupOffsets(String, Set, DeleteShareGroupOffsetsOptions)} with default options.
      * See the overload for more details.
      *
      * @param groupId The group for which to delete offsets.
-     * @param partitions The topic-partitions.
+     * @param topics The topics for which to delete offsets.
      * @return The DeleteShareGroupOffsetsResult.
      */
-    default DeleteShareGroupOffsetsResult deleteShareGroupOffsets(String groupId, Set<TopicPartition> partitions) {
-        return deleteShareGroupOffsets(groupId, partitions, new DeleteShareGroupOffsetsOptions());
+    default DeleteShareGroupOffsetsResult deleteShareGroupOffsets(String groupId, Set<String> topics) {
+        return deleteShareGroupOffsets(groupId, topics, new DeleteShareGroupOffsetsOptions());
     }
 
     /**

clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupOffsetsResult.java

@@ -17,7 +17,6 @@
 package org.apache.kafka.clients.admin;
 
 import org.apache.kafka.common.KafkaFuture;
-import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.common.errors.ApiException;
 import org.apache.kafka.common.internals.KafkaFutureImpl;
@@ -33,27 +32,27 @@
 @InterfaceStability.Evolving
 public class DeleteShareGroupOffsetsResult {
 
-    private final KafkaFuture<Map<TopicPartition, ApiException>> future;
-    private final Set<TopicPartition> partitions;
+    private final KafkaFuture<Map<String, ApiException>> future;
+    private final Set<String> topics;
 
-    DeleteShareGroupOffsetsResult(KafkaFuture<Map<TopicPartition, ApiException>> future, Set<TopicPartition> partitions) {
+    DeleteShareGroupOffsetsResult(KafkaFuture<Map<String, ApiException>> future, Set<String> topics) {
         this.future = future;
-        this.partitions = partitions;
+        this.topics = topics;
     }
 
     /**
      * Return a future which succeeds only if all the deletions succeed.
-     * If not, the first partition error shall be returned.
+     * If not, the first topic error shall be returned.
      */
     public KafkaFuture<Void> all() {
         final KafkaFutureImpl<Void> result = new KafkaFutureImpl<>();
 
-        this.future.whenComplete((topicPartitions, throwable) -> {
+        this.future.whenComplete((topicResults, throwable) -> {
             if (throwable != null) {
                 result.completeExceptionally(throwable);
             } else {
-                for (TopicPartition partition : partitions) {
-                    if (maybeCompleteExceptionally(topicPartitions, partition, result)) {
+                for (String topic : topics) {
+                    if (maybeCompleteExceptionally(topicResults, topic, result)) {
                         return;
                     }
                 }
@@ -64,32 +63,32 @@ public KafkaFuture<Void> all() {
     }
 
     /**
-     * Return a future which can be used to check the result for a given partition.
+     * Return a future which can be used to check the result for a given topic.
      */
-    public KafkaFuture<Void> partitionResult(final TopicPartition partition) {
-        if (!partitions.contains(partition)) {
-            throw new IllegalArgumentException("Partition " + partition + " was not included in the original request");
+    public KafkaFuture<Void> topicResult(final String topic) {
+        if (!topics.contains(topic)) {
+            throw new IllegalArgumentException("Topic " + topic + " was not included in the original request");
         }
         final KafkaFutureImpl<Void> result = new KafkaFutureImpl<>();
 
-        this.future.whenComplete((topicPartitions, throwable) -> {
+        this.future.whenComplete((topicResults, throwable) -> {
             if (throwable != null) {
                 result.completeExceptionally(throwable);
-            } else if (!maybeCompleteExceptionally(topicPartitions, partition, result)) {
+            } else if (!maybeCompleteExceptionally(topicResults, topic, result)) {
                 result.complete(null);
             }
         });
         return result;
     }
 
-    private boolean maybeCompleteExceptionally(Map<TopicPartition, ApiException> partitionLevelErrors,
-                                               TopicPartition partition,
+    private boolean maybeCompleteExceptionally(Map<String, ApiException> topicLevelErrors,
+                                               String topic,
                                                KafkaFutureImpl<Void> result) {
         Throwable exception;
-        if (!partitionLevelErrors.containsKey(partition)) {
-            exception = new IllegalArgumentException("Offset deletion result for partition \"" + partition + "\" was not included in the response");
+        if (!topicLevelErrors.containsKey(topic)) {
+            exception = new IllegalArgumentException("Offset deletion result for topic \"" + topic + "\" was not included in the response");
         } else {
-            exception = partitionLevelErrors.get(partition);
+            exception = topicLevelErrors.get(topic);
         }
 
         if (exception != null) {

clients/src/main/java/org/apache/kafka/clients/admin/ForwardingAdmin.java

@@ -336,8 +336,8 @@ public ListShareGroupOffsetsResult listShareGroupOffsets(Map<String, ListShareGr
     }
 
     @Override
-    public DeleteShareGroupOffsetsResult deleteShareGroupOffsets(String groupId, Set<TopicPartition> partitions, DeleteShareGroupOffsetsOptions options) {
-        return delegate.deleteShareGroupOffsets(groupId, partitions, options);
+    public DeleteShareGroupOffsetsResult deleteShareGroupOffsets(String groupId, Set<String> topics, DeleteShareGroupOffsetsOptions options) {
+        return delegate.deleteShareGroupOffsets(groupId, topics, options);
     }
 
     @Override

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -3846,11 +3846,11 @@ public ListShareGroupOffsetsResult listShareGroupOffsets(final Map<String, ListS
     }
 
     @Override
-    public DeleteShareGroupOffsetsResult deleteShareGroupOffsets(String groupId, Set<TopicPartition> partitions, DeleteShareGroupOffsetsOptions options) {
-        SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, ApiException>> future = DeleteShareGroupOffsetsHandler.newFuture(groupId);
-        DeleteShareGroupOffsetsHandler handler = new DeleteShareGroupOffsetsHandler(groupId, partitions, logContext);
+    public DeleteShareGroupOffsetsResult deleteShareGroupOffsets(String groupId, Set<String> topics, DeleteShareGroupOffsetsOptions options) {
+        SimpleAdminApiFuture<CoordinatorKey, Map<String, ApiException>> future = DeleteShareGroupOffsetsHandler.newFuture(groupId);
+        DeleteShareGroupOffsetsHandler handler = new DeleteShareGroupOffsetsHandler(groupId, topics, logContext);
         invokeDriver(handler, future, options.timeoutMs);
-        return new DeleteShareGroupOffsetsResult(future.get(CoordinatorKey.byGroupId(groupId)), partitions);
+        return new DeleteShareGroupOffsetsResult(future.get(CoordinatorKey.byGroupId(groupId)), topics);
     }
 
     @Override

clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteShareGroupOffsetsHandler.java

@@ -19,7 +19,6 @@
 import org.apache.kafka.clients.admin.DeleteShareGroupOffsetsOptions;
 import org.apache.kafka.clients.admin.KafkaAdminClient;
 import org.apache.kafka.common.Node;
-import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.ApiException;
 import org.apache.kafka.common.message.DeleteShareGroupOffsetsRequestData;
 import org.apache.kafka.common.protocol.Errors;
@@ -38,24 +37,23 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 /**
  * This class is the handler for {@link KafkaAdminClient#deleteShareGroupOffsets(String, Set, DeleteShareGroupOffsetsOptions)} call
  */
-public class DeleteShareGroupOffsetsHandler extends AdminApiHandler.Batched<CoordinatorKey, Map<TopicPartition, ApiException>> {
+public class DeleteShareGroupOffsetsHandler extends AdminApiHandler.Batched<CoordinatorKey, Map<String, ApiException>> {
 
     private final CoordinatorKey groupId;
 
     private final Logger log;
 
-    private final Set<TopicPartition> partitions;
+    private final Set<String> topics;
 
     private final CoordinatorStrategy lookupStrategy;
 
-    public DeleteShareGroupOffsetsHandler(String groupId, Set<TopicPartition> partitions, LogContext logContext) {
+    public DeleteShareGroupOffsetsHandler(String groupId, Set<String> topics, LogContext logContext) {
         this.groupId = CoordinatorKey.byGroupId(groupId);
-        this.partitions = partitions;
+        this.topics = topics;
         this.log = logContext.logger(DeleteShareGroupOffsetsHandler.class);
         this.lookupStrategy = new CoordinatorStrategy(FindCoordinatorRequest.CoordinatorType.GROUP, logContext);
     }
@@ -70,7 +68,7 @@ public AdminApiLookupStrategy<CoordinatorKey> lookupStrategy() {
         return lookupStrategy;
     }
 
-    public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, ApiException>> newFuture(String groupId) {
+    public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<String, ApiException>> newFuture(String groupId) {
         return AdminApiFuture.forKeys(Collections.singleton(CoordinatorKey.byGroupId(groupId)));
     }
 
@@ -85,26 +83,22 @@ private void validateKeys(Set<CoordinatorKey> groupIds) {
     DeleteShareGroupOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<CoordinatorKey> groupIds) {
         validateKeys(groupIds);
 
-        final List<DeleteShareGroupOffsetsRequestData.DeleteShareGroupOffsetsRequestTopic> topics =
+        final List<DeleteShareGroupOffsetsRequestData.DeleteShareGroupOffsetsRequestTopic> requestTopics =
             new ArrayList<>();
-        partitions.stream().collect(Collectors.groupingBy(TopicPartition::topic)).forEach((topic, topicPartitions) -> topics.add(
+        topics.forEach(topic -> requestTopics.add(
             new DeleteShareGroupOffsetsRequestData.DeleteShareGroupOffsetsRequestTopic()
                 .setTopicName(topic)
-                .setPartitions(topicPartitions.stream()
-                    .map(TopicPartition::partition)
-                    .collect(Collectors.toList())
-                )
         ));
 
         return new DeleteShareGroupOffsetsRequest.Builder(
             new DeleteShareGroupOffsetsRequestData()
                 .setGroupId(groupId.idValue)
-                .setTopics(topics)
+                .setTopics(requestTopics)
         );
     }
 
     @Override
-    public ApiResult<CoordinatorKey, Map<TopicPartition, ApiException>> handleResponse(
+    public ApiResult<CoordinatorKey, Map<String, ApiException>> handleResponse(
         Node coordinator,
         Set<CoordinatorKey> groupIds,
         AbstractResponse abstractResponse
@@ -123,23 +117,21 @@ public ApiResult<CoordinatorKey, Map<TopicPartition, ApiException>> handleRespon
 
             return new ApiResult<>(Collections.emptyMap(), groupsFailed, new ArrayList<>(groupsToUnmap));
         } else {
-            final Map<TopicPartition, ApiException> partitionResults = new HashMap<>();
-            response.data().responses().forEach(topic ->
-                topic.partitions().forEach(partition -> {
-                    if (partition.errorCode() != Errors.NONE.code()) {
-                        final Errors partitionError = Errors.forCode(partition.errorCode());
-                        final String partitionErrorMessage = partition.errorMessage();
-                        log.debug("DeleteShareGroupOffsets request for group id {}, topic {} and partition {} failed and returned error {}." + partitionErrorMessage,
-                            groupId.idValue, topic.topicName(), partition.partitionIndex(), partitionError);
-                    }
-                    partitionResults.put(
-                        new TopicPartition(topic.topicName(), partition.partitionIndex()),
-                        Errors.forCode(partition.errorCode()).exception(partition.errorMessage())
-                    );
-                })
-            );
-
-            return ApiResult.completed(groupId, partitionResults);
+            final Map<String, ApiException> topicResults = new HashMap<>();
+            response.data().responses().forEach(topic -> {
+                if (topic.errorCode() != Errors.NONE.code()) {
+                    final Errors topicError = Errors.forCode(topic.errorCode());
+                    final String topicErrorMessage = topic.errorMessage();
+                    log.debug("DeleteShareGroupOffsets request for group id {} and topic {} failed and returned error {}." + topicErrorMessage,
+                        groupId.idValue, topic.topicName(), topicError);
+                }
+                topicResults.put(
+                    topic.topicName(),
+                    Errors.forCode(topic.errorCode()).exception(topic.errorMessage())
+                );
+            });
+
+            return ApiResult.completed(groupId, topicResults);
         }
     }
 

clients/src/main/java/org/apache/kafka/common/requests/DeleteShareGroupOffsetsResponse.java

@@ -43,9 +43,7 @@ public Map<Errors, Integer> errorCounts() {
         Map<Errors, Integer> counts = new EnumMap<>(Errors.class);
         updateErrorCounts(counts, Errors.forCode(data.errorCode()));
         data.responses().forEach(
-            topicResult -> topicResult.partitions().forEach(
-                partitionResult -> updateErrorCounts(counts, Errors.forCode(partitionResult.errorCode()))
-            )
+            topicResult -> updateErrorCounts(counts, Errors.forCode(topicResult.errorCode()))
         );
         return counts;
     }

clients/src/main/resources/common/message/DeleteShareGroupOffsetsRequest.json

@@ -26,9 +26,7 @@
     { "name": "Topics", "type": "[]DeleteShareGroupOffsetsRequestTopic", "versions": "0+",
       "about": "The topics to delete offsets for.",  "fields": [
       { "name": "TopicName", "type": "string", "versions": "0+", "entityType": "topicName",
-        "about": "The topic name." },
-      { "name": "Partitions", "type": "[]int32", "versions": "0+",
-        "about": "The partitions." }
+        "about": "The topic name." }
     ]}
   ]
 }

clients/src/main/resources/common/message/DeleteShareGroupOffsetsResponse.json

@@ -30,6 +30,7 @@
   // - KAFKA_STORAGE_ERROR (version 0+)
   // - INVALID_REQUEST (version 0+)
   // - UNKNOWN_SERVER_ERROR (version 0+)
+  // - UNKNOWN_TOPIC_OR_PARTITION (version 0+)
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
       "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
@@ -43,14 +44,10 @@
         "about": "The topic name." },
       { "name": "TopicId", "type": "uuid", "versions": "0+",
         "about": "The unique topic ID." },
-      { "name": "Partitions", "type": "[]DeleteShareGroupOffsetsResponsePartition", "versions": "0+", "fields": [
-        { "name": "PartitionIndex", "type": "int32", "versions": "0+",
-          "about": "The partition index." },
-        { "name": "ErrorCode", "type": "int16", "versions": "0+",
-          "about": "The partition-level error code, or 0 if there was no error." },
-        { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
-          "about": "The partition-level error message, or null if there was no error." }
-      ]}
+      { "name": "ErrorCode", "type": "int16", "versions": "0+",
+        "about": "The topic-level error code, or 0 if there was no error." },
+      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
+        "about": "The topic-level error message, or null if there was no error." }
     ]}
   ]
 }