KAFKA-20340 Remove GroupConfigManager dependency from SharePartition (#21861)

`SharePartition` held both a `GroupConfigManager` and a
`ShareGroupConfigProvider` reference, but `ShareGroupConfigProvider`
already wraps `GroupConfigManager`.

This PR removes the redundant `GroupConfigManager` dependency so that
`SharePartition` only uses `ShareGroupConfigProvider` for dynamic group
configuration lookups, as suggested in

[#21627.](#21627 (comment))

Reviewers: Sean Quah <squah@confluent.io>, Chia-Ping Tsai
 <chia7712@gmail.com>

Author: majialoong

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

@@ -39,8 +39,6 @@
 import org.apache.kafka.common.record.internal.Record;
 import org.apache.kafka.common.record.internal.RecordBatch;
 import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.coordinator.group.GroupConfig;
-import org.apache.kafka.coordinator.group.GroupConfigManager;
 import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
 import org.apache.kafka.coordinator.group.modern.share.ShareGroupConfigProvider;
 import org.apache.kafka.server.share.acknowledge.ShareAcknowledgementBatch;
@@ -208,11 +206,6 @@ enum SharePartitionState {
      */
     private final int defaultMaxDeliveryCount;
 
-    /**
-     * The group config manager is used to retrieve the values for dynamic group configurations
-     */
-    private final GroupConfigManager groupConfigManager;
-
     /**
      * The provider used to retrieve share group dynamic configuration values.
      */
@@ -343,11 +336,11 @@ enum SharePartitionState {
         Time time,
         Persister persister,
         ReplicaManager replicaManager,
-        GroupConfigManager groupConfigManager,
+        ShareGroupConfigProvider configProvider,
         SharePartitionListener listener
     ) {
         this(groupId, topicIdPartition, leaderEpoch, defaultMaxInFlightRecords, defaultMaxDeliveryCount, defaultRecordLockDurationMs,
-            timer, time, persister, replicaManager, groupConfigManager, SharePartitionState.EMPTY, listener,
+            timer, time, persister, replicaManager, configProvider, SharePartitionState.EMPTY, listener,
             new SharePartitionMetrics(groupId, topicIdPartition.topic(), topicIdPartition.partition()));
     }
 
@@ -364,7 +357,7 @@ enum SharePartitionState {
         Time time,
         Persister persister,
         ReplicaManager replicaManager,
-        GroupConfigManager groupConfigManager,
+        ShareGroupConfigProvider configProvider,
         SharePartitionState sharePartitionState,
         SharePartitionListener listener,
         SharePartitionMetrics sharePartitionMetrics
@@ -385,8 +378,7 @@ enum SharePartitionState {
         this.persister = persister;
         this.partitionState = sharePartitionState;
         this.replicaManager = replicaManager;
-        this.groupConfigManager = groupConfigManager;
-        this.configProvider = new ShareGroupConfigProvider(groupConfigManager);
+        this.configProvider = configProvider;
         this.fetchOffsetMetadata = new OffsetMetadata();
         this.delayedShareFetchKey = new DelayedShareFetchGroupKey(groupId, topicIdPartition);
         this.listener = listener;
@@ -3040,9 +3032,7 @@ private long startOffsetDuringInitialization(long partitionDataStartOffset) {
         if (partitionDataStartOffset != PartitionFactory.UNINITIALIZED_START_OFFSET) {
             return partitionDataStartOffset;
         }
-        ShareGroupAutoOffsetResetStrategy offsetResetStrategy = groupConfigManager.groupConfig(groupId)
-            .map(GroupConfig::shareAutoOffsetReset)
-            .orElseGet(GroupConfig::defaultShareAutoOffsetReset);
+        ShareGroupAutoOffsetResetStrategy offsetResetStrategy = configProvider.autoOffsetReset(groupId);
 
         if (offsetResetStrategy.type() == ShareGroupAutoOffsetResetStrategy.StrategyType.LATEST) {
             return offsetForLatestTimestamp(topicIdPartition, replicaManager, leaderEpoch);

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

@@ -31,7 +31,7 @@
 import org.apache.kafka.common.requests.ShareRequestMetadata;
 import org.apache.kafka.common.utils.ImplicitLinkedHashCollection;
 import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.coordinator.group.GroupConfigManager;
+import org.apache.kafka.coordinator.group.modern.share.ShareGroupConfigProvider;
 import org.apache.kafka.server.common.ShareVersion;
 import org.apache.kafka.server.partition.PartitionListener;
 import org.apache.kafka.server.share.CachedSharePartition;
@@ -104,9 +104,9 @@ public class SharePartitionManager implements AutoCloseable {
     private final ShareSessionCache cache;
 
     /**
-     * The group config manager is used to retrieve the values for dynamic group configurations
+     * The provider used to retrieve share group dynamic configuration values.
      */
-    private final GroupConfigManager groupConfigManager;
+    private final ShareGroupConfigProvider configProvider;
 
     /**
      * The default record lock duration is the time in milliseconds that a record lock is held for.
@@ -157,7 +157,7 @@ public SharePartitionManager(
         int maxInFlightRecords,
         long remoteFetchMaxWaitMs,
         Persister persister,
-        GroupConfigManager groupConfigManager,
+        ShareGroupConfigProvider configProvider,
         BrokerTopicStats brokerTopicStats
     ) {
         this(replicaManager,
@@ -169,7 +169,7 @@ public SharePartitionManager(
             maxInFlightRecords,
             remoteFetchMaxWaitMs,
             persister,
-            groupConfigManager,
+            configProvider,
             new ShareGroupMetrics(time),
             brokerTopicStats
         );
@@ -185,7 +185,7 @@ private SharePartitionManager(
         int maxInFlightRecords,
         long remoteFetchMaxWaitMs,
         Persister persister,
-        GroupConfigManager groupConfigManager,
+        ShareGroupConfigProvider configProvider,
         ShareGroupMetrics shareGroupMetrics,
         BrokerTopicStats brokerTopicStats
     ) {
@@ -200,7 +200,7 @@ private SharePartitionManager(
             maxInFlightRecords,
             remoteFetchMaxWaitMs,
             persister,
-            groupConfigManager,
+            configProvider,
             shareGroupMetrics,
             brokerTopicStats
         );
@@ -218,7 +218,7 @@ private SharePartitionManager(
             int maxInFlightRecords,
             long remoteFetchMaxWaitMs,
             Persister persister,
-            GroupConfigManager groupConfigManager,
+            ShareGroupConfigProvider configProvider,
             ShareGroupMetrics shareGroupMetrics,
             BrokerTopicStats brokerTopicStats
     ) {
@@ -232,7 +232,7 @@ private SharePartitionManager(
         this.maxInFlightRecords = maxInFlightRecords;
         this.remoteFetchMaxWaitMs = remoteFetchMaxWaitMs;
         this.persister = persister;
-        this.groupConfigManager = groupConfigManager;
+        this.configProvider = configProvider;
         this.shareGroupMetrics = shareGroupMetrics;
         this.brokerTopicStats = brokerTopicStats;
         this.cache.registerShareGroupListener(new ShareGroupListenerImpl());
@@ -718,7 +718,7 @@ private SharePartition getOrCreateSharePartition(SharePartitionKey sharePartitio
                             time,
                             persister,
                             replicaManager,
-                            groupConfigManager,
+                            configProvider,
                             listener
                     );
                 });

core/src/main/scala/kafka/server/BrokerServer.scala

@@ -36,6 +36,7 @@ import org.apache.kafka.common.{ClusterResource, TopicPartition, Uuid}
 import org.apache.kafka.coordinator.common.runtime.{CoordinatorLoaderImpl, CoordinatorRecord}
 import org.apache.kafka.coordinator.group.metrics.{GroupCoordinatorMetrics, GroupCoordinatorRuntimeMetrics}
 import org.apache.kafka.coordinator.group.{GroupConfigManager, GroupCoordinator, GroupCoordinatorRecordSerde, GroupCoordinatorService}
+import org.apache.kafka.coordinator.group.modern.share.ShareGroupConfigProvider
 import org.apache.kafka.coordinator.share.metrics.{ShareCoordinatorMetrics, ShareCoordinatorRuntimeMetrics}
 import org.apache.kafka.coordinator.share.{ShareCoordinator, ShareCoordinatorRecordSerde, ShareCoordinatorService}
 import org.apache.kafka.coordinator.transaction.ProducerIdManager
@@ -456,7 +457,7 @@ class BrokerServer(
         config.shareGroupConfig.shareGroupPartitionMaxRecordLocks,
         config.remoteLogManagerConfig.remoteFetchMaxWaitMs().toLong,
         persister,
-        groupConfigManager,
+        new ShareGroupConfigProvider(groupConfigManager),
         brokerTopicStats
       )
 

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

@@ -51,6 +51,7 @@
 import org.apache.kafka.common.utils.ImplicitLinkedHashCollection;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.coordinator.group.GroupConfigManager;
+import org.apache.kafka.coordinator.group.modern.share.ShareGroupConfigProvider;
 import org.apache.kafka.server.common.ShareVersion;
 import org.apache.kafka.server.purgatory.DelayedOperationKey;
 import org.apache.kafka.server.purgatory.DelayedOperationPurgatory;
@@ -3287,7 +3288,7 @@ public SharePartitionManager build() {
                 MAX_IN_FLIGHT_MESSAGES,
                 REMOTE_FETCH_MAX_WAIT_MS,
                 persister,
-                mock(GroupConfigManager.class),
+                new ShareGroupConfigProvider(mock(GroupConfigManager.class)),
                 shareGroupMetrics,
                 brokerTopicStats
             );