MINOR: Move ShareAcquireMode to consumer.internal package (#20973) (#21015)

This PR moves `ShareAcquireMode` to the `consumer.internal` package and
addresses comments from
#20246 (review)

Reviewers: Apoorv Mittal <apoorvmittal10@gmail.com>, Andrew Schofield
<aschofield@confluent.io>, Chirag Wadhwa <cwadhwa@confluent.io>,
Chia-Ping Tsai <chia7712@gmail.com>

Author: JimmyWang6

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

@@ -21,6 +21,7 @@
 import org.apache.kafka.clients.MetadataRecoveryStrategy;
 import org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy;
 import org.apache.kafka.clients.consumer.internals.ShareAcknowledgementMode;
+import org.apache.kafka.clients.consumer.internals.ShareAcquireMode;
 import org.apache.kafka.common.IsolationLevel;
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;

clients/src/main/java/org/apache/kafka/clients/consumer/ShareAcquireMode.java renamed to clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareAcquireMode.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.kafka.clients.consumer;
+package org.apache.kafka.clients.consumer.internals;
 
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigException;
@@ -29,7 +29,7 @@ public enum ShareAcquireMode {
 
     public final String name;
 
-    public final byte id;
+    final byte id;
 
     ShareAcquireMode(final String name, final byte id) {
         this.name = name;
@@ -40,6 +40,9 @@ public enum ShareAcquireMode {
      * Case-insensitive acquire mode lookup by string name.
      */
     public static ShareAcquireMode of(final String name) {
+        if (name == null) {
+            throw new IllegalArgumentException("ShareAcquireMode is null");
+        }
         try {
             return ShareAcquireMode.valueOf(name.toUpperCase(Locale.ROOT));
         } catch (IllegalArgumentException e) {
@@ -65,7 +68,7 @@ public static ShareAcquireMode forId(byte id) {
 
     @Override
     public String toString() {
-        return "ShareAcquireMode(" + name + " (" + id + "))";
+        return name;
     }
 
     public static class Validator implements ConfigDef.Validator {

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java

@@ -18,7 +18,6 @@
 
 import org.apache.kafka.clients.ClientResponse;
 import org.apache.kafka.clients.Metadata;
-import org.apache.kafka.clients.consumer.ShareAcquireMode;
 import org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult;
 import org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest;
 import org.apache.kafka.clients.consumer.internals.events.ShareAcknowledgementEvent;

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetchConfig.java

@@ -17,7 +17,6 @@
 package org.apache.kafka.clients.consumer.internals;
 
 import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.ShareAcquireMode;
 import org.apache.kafka.common.IsolationLevel;
 
 import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.configuredIsolationLevel;

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

@@ -16,7 +16,7 @@
  */
 package org.apache.kafka.common.requests;
 
-import org.apache.kafka.clients.consumer.ShareAcquireMode;
+import org.apache.kafka.clients.consumer.internals.ShareAcquireMode;
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;

clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareAcquireModeTest.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals;
+
+import org.apache.kafka.common.config.ConfigException;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ShareAcquireModeTest {
+
+    @Test
+    public void testFromString() {
+        assertEquals(ShareAcquireMode.BATCH_OPTIMIZED, ShareAcquireMode.of("batch_optimized"));
+        assertEquals(ShareAcquireMode.BATCH_OPTIMIZED, ShareAcquireMode.of("BATCH_OPTIMIZED"));
+        assertEquals(ShareAcquireMode.RECORD_LIMIT, ShareAcquireMode.of("record_limit"));
+        assertEquals(ShareAcquireMode.RECORD_LIMIT, ShareAcquireMode.of("RECORD_LIMIT"));
+        assertThrows(IllegalArgumentException.class, () -> ShareAcquireMode.of("invalid_mode"));
+        assertThrows(IllegalArgumentException.class, () -> ShareAcquireMode.of(""));
+        assertThrows(IllegalArgumentException.class, () -> ShareAcquireMode.of(null));
+    }
+
+    @Test
+    public void testValidator() {
+        ShareAcquireMode.Validator validator = new ShareAcquireMode.Validator();
+        assertDoesNotThrow(() -> validator.ensureValid("test", "batch_optimized"));
+        assertDoesNotThrow(() -> validator.ensureValid("test", "BATCH_OPTIMIZED"));
+        assertDoesNotThrow(() -> validator.ensureValid("test", "record_limit"));
+        assertDoesNotThrow(() -> validator.ensureValid("test", "RECORD_LIMIT"));
+        assertThrows(ConfigException.class, () -> validator.ensureValid("test", "invalid_mode"));
+        assertThrows(ConfigException.class, () -> validator.ensureValid("test", ""));
+        assertThrows(ConfigException.class, () -> validator.ensureValid("test", null));
+    }
+
+    @Test
+    public void testValidatorToString() {
+        ShareAcquireMode.Validator validator = new ShareAcquireMode.Validator();
+        assertEquals("[batch_optimized, record_limit]", validator.toString());
+    }
+}

clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java

@@ -24,7 +24,6 @@
 import org.apache.kafka.clients.consumer.AcknowledgeType;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ShareAcquireMode;
 import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
 import org.apache.kafka.clients.consumer.internals.events.ShareAcknowledgementEvent;
 import org.apache.kafka.clients.consumer.internals.events.ShareAcknowledgementEventHandler;

clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandlerTest.java

@@ -18,7 +18,6 @@
 
 import org.apache.kafka.clients.consumer.AcknowledgeType;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.ShareAcquireMode;
 import org.apache.kafka.common.IsolationLevel;
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.TopicPartition;

core/src/main/java/kafka/server/share/SharePartition.java

@@ -20,7 +20,7 @@
 import kafka.server.share.SharePartitionManager.SharePartitionListener;
 
 import org.apache.kafka.clients.consumer.AcknowledgeType;
-import org.apache.kafka.clients.consumer.ShareAcquireMode;
+import org.apache.kafka.clients.consumer.internals.ShareAcquireMode;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.Uuid;
@@ -1836,13 +1836,13 @@ private List<AcquiredRecords> createBatches(
                         sharePartitionMetrics);
                     int delayMs = recordLockDurationMsOrDefault(groupConfigManager, groupId, defaultRecordLockDurationMs);
                     long lastOffset = acquiredRecords.firstOffset() + maxFetchRecords - 1;
-                    acquiredRecords.setLastOffset(lastOffset);
                     inFlightBatch.maybeInitializeOffsetStateUpdate(lastOffset, delayMs);
                     updateFindNextFetchOffset(true);
 
                     cachedState.put(acquiredRecords.firstOffset(), inFlightBatch);
                     sharePartitionMetrics.recordInFlightBatchMessageCount(
                         acquiredRecords.lastOffset() - acquiredRecords.firstOffset() + 1);
+                    acquiredRecords.setLastOffset(lastOffset);
                     return List.of(acquiredRecords);
                 }
             }

core/src/test/java/kafka/server/share/DelayedShareFetchTest.java

@@ -21,7 +21,7 @@
 import kafka.server.ReplicaManager;
 import kafka.server.ReplicaQuota;
 
-import org.apache.kafka.clients.consumer.ShareAcquireMode;
+import org.apache.kafka.clients.consumer.internals.ShareAcquireMode;
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;