Fixes for PlainAdminIntegrationTest#testAlterShareGroupOffsets

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

@@ -2921,7 +2921,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
     val testTopicName = "test_topic"
     val testGroupId = "test_group_id"
     val testClientId = "test_client_id"
-    val fakeGroupId = "fake_group_id"
+    val nonexistentGroupId = "nonexistent_group_id"
     val fakeTopicName = "foo"
 
     val tp1 = new TopicPartition(testTopicName, 0)
@@ -2955,12 +2955,12 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
         assertFutureThrows(classOf[GroupNotEmptyException], offsetAlterResult.partitionResult(tp1))
         assertFutureThrows(classOf[GroupNotEmptyException], offsetAlterResult.partitionResult(tp2))
 
-        // Test the fake group ID
-        val fakeAlterResult = client.alterShareGroupOffsets(fakeGroupId, util.Map.of(tp1, 0, tp2, 0))
+        // Test the non-existent group ID
+        val nonexistentAlterResult = client.alterShareGroupOffsets(nonexistentGroupId, util.Map.of(tp1, 0, tp2, 0))
 
-        assertFutureThrows(classOf[GroupIdNotFoundException], fakeAlterResult.all())
-        assertFutureThrows(classOf[GroupIdNotFoundException], fakeAlterResult.partitionResult(tp1))
-        assertFutureThrows(classOf[GroupIdNotFoundException], fakeAlterResult.partitionResult(tp2))
+        assertFutureThrows(classOf[UnknownTopicOrPartitionException], nonexistentAlterResult.all())
+        assertNull(nonexistentAlterResult.partitionResult(tp1).get())
+        assertFutureThrows(classOf[UnknownTopicOrPartitionException], nonexistentAlterResult.partitionResult(tp2))
       }
 
       // Test offset alter when group is empty

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

@@ -779,10 +779,7 @@ public CoordinatorResult<DeleteShareGroupOffsetsResponseData, CoordinatorRecord>
     }
 
     /**
-     * Make the following checks to make sure the AlterShareGroupOffsetsRequest request is valid:
-     * 1. Checks whether the provided group is empty
-     * 2. Checks the requested topics are presented in the metadataImage
-     * 3. Checks the corresponding share partitions in AlterShareGroupOffsetsRequest are existing
+     * Alters the offsets for a share group.
      *
      * @param groupId - The group ID
      * @param alterShareGroupOffsetsRequestData - The request data for AlterShareGroupOffsetsRequestData
@@ -793,20 +790,7 @@ public CoordinatorResult<Map.Entry<AlterShareGroupOffsetsResponseData, Initializ
         String groupId,
         AlterShareGroupOffsetsRequestData alterShareGroupOffsetsRequestData
     ) {
-        List<CoordinatorRecord> records = new ArrayList<>();
-        ShareGroup group = groupMetadataManager.shareGroup(groupId, Long.MAX_VALUE, true);
-        group.validateOffsetsAlterable();
-
-        Map.Entry<AlterShareGroupOffsetsResponseData, InitializeShareGroupStateParameters> response = groupMetadataManager.completeAlterShareGroupOffsets(
-            groupId,
-            alterShareGroupOffsetsRequestData,
-            records,
-            group
-        );
-        return new CoordinatorResult<>(
-            records,
-            response
-        );
+        return groupMetadataManager.alterShareGroupOffsets(groupId, alterShareGroupOffsetsRequestData.topics());
     }
 
     /**

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

@@ -25,6 +25,7 @@
 import org.apache.kafka.common.errors.FencedMemberEpochException;
 import org.apache.kafka.common.errors.GroupIdNotFoundException;
 import org.apache.kafka.common.errors.GroupMaxSizeReachedException;
+import org.apache.kafka.common.errors.GroupNotEmptyException;
 import org.apache.kafka.common.errors.IllegalGenerationException;
 import org.apache.kafka.common.errors.InconsistentGroupProtocolException;
 import org.apache.kafka.common.errors.InvalidRequestException;
@@ -1171,21 +1172,8 @@ public ShareGroup shareGroup(
         String groupId,
         long committedOffset
     ) throws GroupIdNotFoundException {
-        return shareGroup(groupId, committedOffset, false);
-    }
+        Group group = group(groupId, committedOffset);
 
-    public ShareGroup shareGroup(
-            String groupId,
-            long committedOffset,
-            boolean createIfNotExists
-    ) throws GroupIdNotFoundException {
-        Group group;
-        if (createIfNotExists) {
-            // Get or create the share group. If the group exists, check that it's empty. If it is created, it is empty.
-            group = getOrMaybeCreateShareGroup(groupId, true);
-        } else {
-            group = group(groupId, committedOffset);
-        }
         if (group.type() == SHARE) {
             return (ShareGroup) group;
         } else {
@@ -1512,6 +1500,21 @@ private void throwIfShareGroupIsFull(
         }
     }
 
+    /**
+     * Checks whether the share group is empty.
+     *
+     * @param group     The share group.
+     *
+     * @throws GroupNotEmptyException if the group is not empty.
+     */
+    private void throwIfShareGroupIsNotEmpty(
+        ShareGroup group
+    ) throws GroupNotEmptyException {
+        if (group.numMembers() > 0) {
+            throw new GroupNotEmptyException(Errors.NON_EMPTY_GROUP.message());
+        }
+    }
+
     /**
      * Validates the member epoch provided in the heartbeat request.
      *
@@ -8200,19 +8203,37 @@ public List<DeleteShareGroupStateRequestData.DeleteStateData> sharePartitionsEli
         return deleteShareGroupStateRequestTopicsData;
     }
 
-    public Map.Entry<AlterShareGroupOffsetsResponseData, InitializeShareGroupStateParameters> completeAlterShareGroupOffsets(
+    /**
+     * Handles an AlterShareGroupOffsets request.
+     *
+     * Make the following checks to make sure the AlterShareGroupOffsetsRequest request is valid:
+     * 1. Checks whether the provided group is empty
+     * 2. Checks the requested topics are presented in the metadataImage
+     * 3. Checks the corresponding share partitions in AlterShareGroupOffsetsRequest are existing
+     *
+     * @param groupId   The group id from the request.
+     * @param topics    The topic information for altering the share group's offsets from the request.
+     *
+     * @return A Result containing a pair of ShareGroupHeartbeat response and maybe InitializeShareGroupStateParameters
+     *         and a list of records to update the state machine.
+     */
+    public CoordinatorResult<Map.Entry<AlterShareGroupOffsetsResponseData, InitializeShareGroupStateParameters>, CoordinatorRecord> alterShareGroupOffsets(
         String groupId,
-        AlterShareGroupOffsetsRequestData alterShareGroupOffsetsRequest,
-        List<CoordinatorRecord> records,
-        ShareGroup group
-    ) {
+        AlterShareGroupOffsetsRequestData.AlterShareGroupOffsetsRequestTopicCollection topics
+    ) throws ApiException {
         final long currentTimeMs = time.milliseconds();
+        final List<CoordinatorRecord> records = new ArrayList<>();
+
+        // Get or create the share group. If the group exists, check that it's empty. If it is created, it is empty.
+        final ShareGroup group = getOrMaybeCreateShareGroup(groupId, true);
+        throwIfShareGroupIsNotEmpty(group);
+
         AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection alterShareGroupOffsetsResponseTopics = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection();
 
         Map<Uuid, InitMapValue> initializingTopics = new HashMap<>();
         Map<Uuid, Map<Integer, Long>> offsetByTopicPartitions = new HashMap<>();
 
-        alterShareGroupOffsetsRequest.topics().forEach(topic -> {
+        topics.forEach(topic -> {
             Optional<CoordinatorMetadataImage.TopicMetadata> topicMetadataOpt = metadataImage.topicMetadata(topic.topicName());
             if (topicMetadataOpt.isPresent()) {
                 var topicMetadata = topicMetadataOpt.get();
@@ -8266,10 +8287,13 @@ public Map.Entry<AlterShareGroupOffsetsResponseData, InitializeShareGroupStatePa
         });
 
         addInitializingTopicsRecords(groupId, records, initializingTopics);
-        return Map.entry(
-            new AlterShareGroupOffsetsResponseData()
-                .setResponses(alterShareGroupOffsetsResponseTopics),
-            buildInitializeShareGroupState(groupId, group.groupEpoch(), offsetByTopicPartitions)
+        return new CoordinatorResult<>(
+            records,
+            Map.entry(
+                new AlterShareGroupOffsetsResponseData()
+                    .setResponses(alterShareGroupOffsetsResponseTopics),
+                buildInitializeShareGroupState(groupId, group.groupEpoch(), offsetByTopicPartitions)
+            )
         );
     }
 
@@ -8332,7 +8356,7 @@ public CoordinatorResult<Void, CoordinatorRecord> maybeCleanupShareGroupState(
         return new CoordinatorResult<>(records);
     }
 
-    /*
+    /**
      * Returns a list of {@link DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic} corresponding to the
      * topics for which persister delete share group state request was successful
      * @param groupId                    group ID of the share group

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

@@ -244,10 +244,6 @@ public void validateDeleteGroup() throws ApiException {
         validateEmptyGroup();
     }
 
-    public void validateOffsetsAlterable() throws ApiException {
-        validateEmptyGroup();
-    }
-
     public void validateEmptyGroup() {
         if (state() != ShareGroupState.EMPTY) {
             throw Errors.NON_EMPTY_GROUP.exception();