Move hollow hook from permits to bulk action (#120945)

We do not need the hook for unhollowing shards on all actions
taking primary permits, but rather only one the one that can
ingest into the engine. So moving the hook to the bulk action.

Relates ES-10654

Author: kingherc

server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexResponse;
 import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.RefCountingListener;
 import org.elasticsearch.action.support.replication.PostWriteRefresh;
 import org.elasticsearch.action.support.replication.TransportReplicationAction;
 import org.elasticsearch.action.support.replication.TransportWriteAction;
@@ -145,6 +146,20 @@ protected BulkShardResponse newResponseInstance(StreamInput in) throws IOExcepti
         return new BulkShardResponse(in);
     }
 
+    @Override
+    protected void shardOperationOnPrimary(
+        BulkShardRequest request,
+        IndexShard primary,
+        ActionListener<PrimaryResult<BulkShardRequest, BulkShardResponse>> listener
+    ) {
+        final ActionListener<Void> wrappedListener = listener.delegateFailure(
+            (l, ignored) -> super.shardOperationOnPrimary(request, primary, l)
+        );
+        try (var preBulkProceedListeners = new RefCountingListener(wrappedListener)) {
+            primary.getIndexingOperationListener().preBulkOnPrimary(primary, () -> preBulkProceedListeners.acquire());
+        }
+    }
+
     @Override
     protected void dispatchedShardOperationOnPrimary(
         BulkShardRequest request,

server/src/main/java/org/elasticsearch/index/CompositeIndexEventListener.java

@@ -28,7 +28,6 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.function.Consumer;
-import java.util.function.Supplier;
 
 import static org.elasticsearch.core.Strings.format;
 
@@ -351,15 +350,4 @@ public void afterFilesRestoredFromRepository(IndexShard indexShard) {
         }
     }
 
-    @Override
-    public void onAcquirePrimaryOperationPermit(IndexShard indexShard, Supplier<ActionListener<Void>> onPermitAcquiredListenerSupplier) {
-        for (IndexEventListener listener : listeners) {
-            try {
-                listener.onAcquirePrimaryOperationPermit(indexShard, onPermitAcquiredListenerSupplier);
-            } catch (Exception e) {
-                logger.warn(() -> "[" + indexShard.shardId() + "] failed to invoke the listener on acquiring a primary permit", e);
-                throw e;
-            }
-        }
-    }
 }

server/src/main/java/org/elasticsearch/index/shard/IndexEventListener.java

@@ -17,8 +17,6 @@
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.indices.cluster.IndicesClusterStateService.AllocatedIndices.IndexRemovalReason;
 
-import java.util.function.Supplier;
-
 /**
  * An index event listener is the primary extension point for plugins and build-in services
  * to react / listen to per-index and per-shard events. These listeners are registered per-index
@@ -192,14 +190,4 @@ default void afterIndexShardRecovery(IndexShard indexShard, ActionListener<Void>
      * @param indexShard the shard that is recovering
      */
     default void afterFilesRestoredFromRepository(IndexShard indexShard) {}
-
-    /**
-     * Called when a single primary permit is acquired for the given shard (see
-     * {@link IndexShard#acquirePrimaryOperationPermit(ActionListener, java.util.concurrent.Executor)}).
-     *
-     * @param indexShard the shard of which a primary permit is requested
-     * @param onPermitAcquiredListenerSupplier call this immediately to get a listener when the permit is acquired. The listener must be
-     *                                         completed in order for the permit to be given to the acquiring operation.
-     */
-    default void onAcquirePrimaryOperationPermit(IndexShard indexShard, Supplier<ActionListener<Void>> onPermitAcquiredListenerSupplier) {}
 }

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -34,7 +34,6 @@
 import org.elasticsearch.action.admin.indices.flush.FlushRequest;
 import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeRequest;
 import org.elasticsearch.action.support.PlainActionFuture;
-import org.elasticsearch.action.support.RefCountingListener;
 import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.action.support.replication.PendingReplicationActions;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
@@ -452,6 +451,10 @@ public BulkOperationListener getBulkOperationListener() {
         return this.bulkOperationListener;
     }
 
+    public IndexingOperationListener getIndexingOperationListener() {
+        return this.indexingOperationListeners;
+    }
+
     public ShardIndexWarmerService warmerService() {
         return this.shardWarmerService;
     }
@@ -3585,19 +3588,7 @@ public void acquirePrimaryOperationPermit(
     ) {
         verifyNotClosed();
         assert shardRouting.primary() : "acquirePrimaryOperationPermit should only be called on primary shard: " + shardRouting;
-
-        ActionListener<Releasable> onPermitAcquiredWrapped = onPermitAcquired.delegateFailureAndWrap((delegate, releasable) -> {
-            final ActionListener<Releasable> wrappedListener = indexShardOperationPermits.wrapContextPreservingActionListener(
-                delegate,
-                executorOnDelay,
-                forceExecution
-            );
-            try (var listeners = new RefCountingListener(wrappedListener.map(unused -> releasable))) {
-                indexEventListener.onAcquirePrimaryOperationPermit(this, () -> listeners.acquire());
-            }
-        });
-
-        indexShardOperationPermits.acquire(wrapPrimaryOperationPermitListener(onPermitAcquiredWrapped), executorOnDelay, forceExecution);
+        indexShardOperationPermits.acquire(wrapPrimaryOperationPermitListener(onPermitAcquired), executorOnDelay, forceExecution);
     }
 
     public boolean isPrimaryMode() {

server/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

@@ -230,7 +230,7 @@ private void innerAcquire(
         onAcquired.onResponse(releasable);
     }
 
-    public <T extends Closeable> ActionListener<T> wrapContextPreservingActionListener(
+    private <T extends Closeable> ActionListener<T> wrapContextPreservingActionListener(
         ActionListener<T> listener,
         @Nullable final Executor executorOnDelay,
         final boolean forceExecution

server/src/main/java/org/elasticsearch/index/shard/IndexingOperationListener.java

@@ -9,9 +9,11 @@
 package org.elasticsearch.index.shard;
 
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.index.engine.Engine;
 
 import java.util.List;
+import java.util.function.Supplier;
 
 /**
  * An indexing listener for indexing, delete, events.
@@ -62,6 +64,18 @@ default void postDelete(ShardId shardId, Engine.Delete delete, Engine.DeleteResu
      */
     default void postDelete(ShardId shardId, Engine.Delete delete, Exception ex) {}
 
+    /**
+     * Called when a {@link org.elasticsearch.action.bulk.TransportShardBulkAction} is about to perform index and/or delete operation(s)
+     * on a primary shard.
+     *
+     * This is called from a transport thread and therefore the function should be lightweight and not block the thread. The acquired
+     * listener(s) can be asynchronously completed on another thread at a later time.
+     *
+     * @param indexShard the shard the bulk is about to be performed on
+     * @param proceedListenerSupplier call this immediately to get a listener which must be completed so that the bulk can proceed.
+     */
+    default void preBulkOnPrimary(IndexShard indexShard, Supplier<ActionListener<Void>> proceedListenerSupplier) {}
+
     /**
      * A Composite listener that multiplexes calls to each of the listeners methods.
      */
@@ -149,5 +163,16 @@ public void postDelete(ShardId shardId, Engine.Delete delete, Exception ex) {
                 }
             }
         }
+
+        @Override
+        public void preBulkOnPrimary(IndexShard indexShard, Supplier<ActionListener<Void>> proceedListenerSupplier) {
+            for (IndexingOperationListener listener : listeners) {
+                try {
+                    listener.preBulkOnPrimary(indexShard, proceedListenerSupplier);
+                } catch (Exception e) {
+                    logger.warn(() -> "preBulkOnPrimary listener [" + listener + "] failed", e);
+                }
+            }
+        }
     }
 }