Generalize mechanism to signal need for a mutable shard (#122241)

Introduce beforeIndexShardMutableOperation in IndexEventListener to allow plugins to react when a mutable shard is needed. Add IndexShard#ensureMutable to let consumers signal this need and be notified when the shard is ready for mutable operations.

Relates ES-10787

Author: fcofdez

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

@@ -23,7 +23,6 @@
 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;
@@ -152,12 +151,7 @@ protected void shardOperationOnPrimary(
         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());
-        }
+        primary.ensureMutable(listener.delegateFailure((l, ignored) -> super.shardOperationOnPrimary(request, primary, l)));
     }
 
     @Override

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

@@ -137,6 +137,24 @@ public void indexShardStateChanged(
         }
     }
 
+    @Override
+    public void beforeIndexShardMutableOperation(IndexShard indexShard, ActionListener<Void> listener) {
+        iterateBeforeIndexShardMutableOperation(indexShard, listener.delegateResponse((l, e) -> {
+            logger.warn(() -> format("%s failed to invoke the listener before ensuring shard mutability", indexShard.shardId()), e);
+            l.onFailure(e);
+        }));
+    }
+
+    private void iterateBeforeIndexShardMutableOperation(IndexShard indexShard, ActionListener<Void> outerListener) {
+        callListeners(
+            indexShard,
+            listeners.stream()
+                .map(iel -> (Consumer<ActionListener<Void>>) (l) -> iel.beforeIndexShardMutableOperation(indexShard, l))
+                .iterator(),
+            outerListener
+        );
+    }
+
     @Override
     public void beforeIndexCreated(Index index, Settings indexSettings) {
         for (IndexEventListener listener : listeners) {

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

@@ -86,6 +86,25 @@ default void indexShardStateChanged(
         @Nullable String reason
     ) {}
 
+    /**
+     * Invoked before a shard performs a mutable operation. Mutable operations include, but are not limited to:
+     * <ul>
+     *     <li>Indexing operations</li>
+     *     <li>Force merges</li>
+     * </ul>
+     *
+     * This method ensures that the shard is ready to accept mutating operations. This is particularly useful in cases
+     * where the shard initializes its internal {@link org.elasticsearch.index.engine.Engine} lazily, which may take some time.
+     * The provided listener should be notified once the shard is prepared to proceed with the operation.
+     * This can be called from a transport thread and therefore the function should be lightweight and not block the thread.
+     *
+     * @param indexShard the shard where the mutable operation will be performed
+     * @param listener   the listener to be notified when the shard is ready to proceed
+     */
+    default void beforeIndexShardMutableOperation(IndexShard indexShard, ActionListener<Void> listener) {
+        listener.onResponse(null);
+    }
+
     /**
      * Called before the index gets created. Note that this is also called
      * when the index is created on data nodes

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

@@ -451,10 +451,6 @@ public BulkOperationListener getBulkOperationListener() {
         return this.bulkOperationListener;
     }
 
-    public IndexingOperationListener getIndexingOperationListener() {
-        return this.indexingOperationListeners;
-    }
-
     public ShardIndexWarmerService warmerService() {
         return this.shardWarmerService;
     }
@@ -4507,4 +4503,19 @@ public void waitForPrimaryTermAndGeneration(long primaryTerm, long segmentGenera
         );
     }
 
+    /**
+     * Ensures that the shard is ready to perform mutable operations.
+     * This method is particularly useful when the shard initializes its internal
+     * {@link org.elasticsearch.index.engine.Engine} lazily, as it may take some time before becoming mutable.
+     *
+     * The provided listener will be notified once the shard is ready for mutating operations.
+     *
+     * @param listener the listener to be notified when the shard is mutable
+     */
+    public void ensureMutable(ActionListener<Void> listener) {
+        indexEventListener.beforeIndexShardMutableOperation(this, listener.delegateFailure((l, unused) -> {
+            // TODO ES-10826: Acquire ref to engine and retry if it's immutable again?
+            l.onResponse(null);
+        }));
+    }
 }

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

@@ -9,11 +9,9 @@
 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.
@@ -64,18 +62,6 @@ 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.
      */
@@ -163,16 +149,5 @@ 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);
-                }
-            }
-        }
     }
 }