KAFKA-17509: Introduce a delayed action queue to complete purgatory a…

…ctions outside purgatory. (apache#17177)

Add purgatory actions to DelayedActionQueue when partition locks are released after fetch in forceComplete. 

Reviewers: David Arthur <[email protected]>, Apoorv Mittal <[email protected]>, Jun Rao <[email protected]>

core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java

@@ -21,6 +21,7 @@
 import kafka.log.remote.RemoteLogManager;
 import kafka.server.AddPartitionsToTxnManager;
 import kafka.server.AlterPartitionManager;
+import kafka.server.DelayedActionQueue;
 import kafka.server.DelayedDeleteRecords;
 import kafka.server.DelayedElectLeader;
 import kafka.server.DelayedFetch;
@@ -216,6 +217,7 @@ public ReplicaManager build() {
                              OptionConverters.toScala(threadNamePrefix),
                              () -> brokerEpoch,
                              OptionConverters.toScala(addPartitionsToTxnManager),
-                             directoryEventHandler);
+                             directoryEventHandler,
+                             new DelayedActionQueue());
     }
 }

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

@@ -16,7 +16,9 @@
  */
 package kafka.server.share;
 
+import kafka.server.ActionQueue;
 import kafka.server.DelayedOperation;
+import kafka.server.DelayedOperationPurgatory;
 import kafka.server.LogReadResult;
 import kafka.server.QuotaFactory;
 import kafka.server.ReplicaManager;
@@ -54,17 +56,23 @@ public class DelayedShareFetch extends DelayedOperation {
     private final ShareFetchData shareFetchData;
     private final ReplicaManager replicaManager;
     private final Map<SharePartitionKey, SharePartition> partitionCacheMap;
+    private final ActionQueue delayedActionQueue;
+    private final DelayedOperationPurgatory<DelayedShareFetch> delayedShareFetchPurgatory;
 
     private Map<TopicIdPartition, FetchRequest.PartitionData> topicPartitionDataFromTryComplete;
 
     DelayedShareFetch(
             ShareFetchData shareFetchData,
             ReplicaManager replicaManager,
-            Map<SharePartitionKey, SharePartition> partitionCacheMap) {
+            Map<SharePartitionKey, SharePartition> partitionCacheMap,
+            ActionQueue delayedActionQueue,
+            DelayedOperationPurgatory<DelayedShareFetch> delayedShareFetchPurgatory) {
         super(shareFetchData.fetchParams().maxWaitMs, Option.empty());
         this.shareFetchData = shareFetchData;
         this.replicaManager = replicaManager;
         this.partitionCacheMap = partitionCacheMap;
+        this.delayedActionQueue = delayedActionQueue;
+        this.delayedShareFetchPurgatory = delayedShareFetchPurgatory;
         this.topicPartitionDataFromTryComplete = new LinkedHashMap<>();
     }
 
@@ -131,6 +139,16 @@ public void onComplete() {
                     }
                     // Releasing the lock to move ahead with the next request in queue.
                     releasePartitionLocks(shareFetchData.groupId(), topicPartitionData.keySet());
+                    // If we have a fetch request completed for a topic-partition, we release the locks for that partition,
+                    // then we should check if there is a pending share fetch request for the topic-partition and complete it.
+                    // We add the action to delayed actions queue to avoid an infinite call stack, which could happen if
+                    // we directly call delayedShareFetchPurgatory.checkAndComplete
+                    delayedActionQueue.add(() -> {
+                        result.keySet().forEach(topicIdPartition ->
+                            delayedShareFetchPurgatory.checkAndComplete(
+                                new DelayedShareFetchKey(shareFetchData.groupId(), topicIdPartition)));
+                        return BoxedUnit.UNIT;
+                    });
                 });
 
         } catch (Exception e) {

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

@@ -16,6 +16,7 @@
  */
 package kafka.server.share;
 
+import kafka.server.ActionQueue;
 import kafka.server.DelayedOperationPurgatory;
 import kafka.server.ReplicaManager;
 
@@ -147,6 +148,11 @@ public class SharePartitionManager implements AutoCloseable {
      */
     private final DelayedOperationPurgatory<DelayedShareFetch> delayedShareFetchPurgatory;
 
+    /**
+     * The delayed actions queue is used to complete any pending delayed share fetch actions.
+     */
+    private final ActionQueue delayedActionsQueue;
+
     public SharePartitionManager(
         ReplicaManager replicaManager,
         Time time,
@@ -156,6 +162,7 @@ public SharePartitionManager(
         int maxInFlightMessages,
         int shareFetchPurgatoryPurgeIntervalRequests,
         Persister persister,
+        ActionQueue delayedActionsQueue,
         Metrics metrics
     ) {
         this(replicaManager,
@@ -167,6 +174,7 @@ public SharePartitionManager(
             maxInFlightMessages,
             shareFetchPurgatoryPurgeIntervalRequests,
             persister,
+            delayedActionsQueue,
             metrics
         );
     }
@@ -181,6 +189,7 @@ private SharePartitionManager(
         int maxInFlightMessages,
         int shareFetchPurgatoryPurgeIntervalRequests,
         Persister persister,
+        ActionQueue delayedActionsQueue,
         Metrics metrics
     ) {
         this.replicaManager = replicaManager;
@@ -197,6 +206,7 @@ private SharePartitionManager(
         this.persister = persister;
         this.shareGroupMetrics = new ShareGroupMetrics(Objects.requireNonNull(metrics), time);
         this.delayedShareFetchPurgatory = new DelayedOperationPurgatory<>("ShareFetch", this.timer, this.replicaManager.localBrokerId(), shareFetchPurgatoryPurgeIntervalRequests, true, true);
+        this.delayedActionsQueue = delayedActionsQueue;
     }
 
     // Visible for testing.
@@ -212,7 +222,8 @@ private SharePartitionManager(
             int maxInFlightMessages,
             Persister persister,
             Metrics metrics,
-            DelayedOperationPurgatory<DelayedShareFetch> delayedShareFetchPurgatory
+            DelayedOperationPurgatory<DelayedShareFetch> delayedShareFetchPurgatory,
+            ActionQueue delayedActionsQueue
     ) {
         this.replicaManager = replicaManager;
         this.time = time;
@@ -227,6 +238,7 @@ private SharePartitionManager(
         this.persister = persister;
         this.shareGroupMetrics = new ShareGroupMetrics(Objects.requireNonNull(metrics), time);
         this.delayedShareFetchPurgatory = delayedShareFetchPurgatory;
+        this.delayedActionsQueue = delayedActionsQueue;
     }
 
     /**
@@ -600,7 +612,7 @@ void maybeProcessFetchQueue() {
                     new DelayedShareFetchKey(shareFetchData.groupId(), topicIdPartition)));
 
             // Add the share fetch to the delayed share fetch purgatory to process the fetch request.
-            addDelayedShareFetch(new DelayedShareFetch(shareFetchData, replicaManager, partitionCacheMap),
+            addDelayedShareFetch(new DelayedShareFetch(shareFetchData, replicaManager, partitionCacheMap, delayedActionsQueue, delayedShareFetchPurgatory),
                 delayedShareFetchWatchKeys);
 
             // Release the lock so that other threads can process the queue.

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

@@ -320,6 +320,11 @@ class BrokerServer(
           lifecycleManager.propagateDirectoryFailure(directoryId, config.logDirFailureTimeoutMs)
       }
 
+      /**
+       * TODO: move this action queue to handle thread so we can simplify concurrency handling
+       */
+      val defaultActionQueue = new DelayedActionQueue
+
       this._replicaManager = new ReplicaManager(
         config = config,
         metrics = metrics,
@@ -338,7 +343,8 @@ class BrokerServer(
         delayedRemoteFetchPurgatoryParam = None,
         brokerEpochSupplier = () => lifecycleManager.brokerEpoch,
         addPartitionsToTxnManager = Some(addPartitionsToTxnManager),
-        directoryEventHandler = directoryEventHandler
+        directoryEventHandler = directoryEventHandler,
+        defaultActionQueue = defaultActionQueue
       )
 
       /* start token manager */
@@ -423,6 +429,7 @@ class BrokerServer(
         config.shareGroupConfig.shareGroupPartitionMaxRecordLocks,
         config.shareGroupConfig.shareFetchPurgatoryPurgeIntervalRequests,
         persister,
+        defaultActionQueue,
         new Metrics()
       )
 

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

@@ -290,7 +290,8 @@ class ReplicaManager(val config: KafkaConfig,
                      threadNamePrefix: Option[String] = None,
                      val brokerEpochSupplier: () => Long = () => -1,
                      addPartitionsToTxnManager: Option[AddPartitionsToTxnManager] = None,
-                     val directoryEventHandler: DirectoryEventHandler = DirectoryEventHandler.NOOP
+                     val directoryEventHandler: DirectoryEventHandler = DirectoryEventHandler.NOOP,
+                     val defaultActionQueue: ActionQueue = new DelayedActionQueue
                      ) extends Logging {
   private val metricsGroup = new KafkaMetricsGroup(this.getClass)
 
@@ -741,11 +742,6 @@ class ReplicaManager(val config: KafkaConfig,
     localLog(topicPartition).map(_.parentDir)
   }
 
-  /**
-   * TODO: move this action queue to handle thread so we can simplify concurrency handling
-   */
-  private val defaultActionQueue = new DelayedActionQueue
-
   def tryCompleteActions(): Unit = defaultActionQueue.tryCompleteActions()
 
   /**