Add support for relocating shards with externally acquired primary permits (#120098)

For hollow shards, we HollowIndexEngine holds primary permits and we can relocate an index shard with the permits provided by HollowIndexEngine

Author: arteam

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

@@ -779,10 +779,28 @@ public void relocated(
         final String targetAllocationId,
         final BiConsumer<ReplicationTracker.PrimaryContext, ActionListener<Void>> consumer,
         final ActionListener<Void> listener
+    ) throws IllegalIndexShardStateException, IllegalStateException {
+        relocated(targetNodeId, targetAllocationId, consumer, listener, null);
+    }
+
+    /**
+     * Provides an variant of {@link IndexShard#relocated(String, String, BiConsumer, ActionListener, Releasable)} with an option
+     * to relocate the shard under externally acquired primary permits.
+     *
+     * @param acquiredPrimaryPermits if null, waits until all the primary permits are acquired, otherwise it calls the consumer immediately
+     */
+    public void relocated(
+        final String targetNodeId,
+        final String targetAllocationId,
+        final BiConsumer<ReplicationTracker.PrimaryContext, ActionListener<Void>> consumer,
+        final ActionListener<Void> listener,
+        @Nullable final Releasable acquiredPrimaryPermits
     ) throws IllegalIndexShardStateException, IllegalStateException {
         assert shardRouting.primary() : "only primaries can be marked as relocated: " + shardRouting;
+        assert acquiredPrimaryPermits == null || indexShardOperationPermits.getActiveOperationsCount() == OPERATIONS_BLOCKED
+            : "external primary permits are provided but not held by the shard";
         try (Releasable forceRefreshes = refreshListeners.forceRefreshes()) {
-            indexShardOperationPermits.blockOperations(new ActionListener<>() {
+            ActionListener<Releasable> onAcquired = new ActionListener<>() {
                 @Override
                 public void onResponse(Releasable releasable) {
                     boolean success = false;
@@ -860,8 +878,13 @@ public void onFailure(Exception e) {
                         listener.onFailure(e);
                     }
                 }
-            }, 30L, TimeUnit.MINUTES, EsExecutors.DIRECT_EXECUTOR_SERVICE); // Wait on current thread because this execution is wrapped by
-                                                                            // CancellableThreads and we want to be able to interrupt it
+            };
+            if (acquiredPrimaryPermits == null) {
+                // Wait on current thread because this execution is wrapped by CancellableThreads and we want to be able to interrupt it
+                indexShardOperationPermits.blockOperations(onAcquired, 30L, TimeUnit.MINUTES, EsExecutors.DIRECT_EXECUTOR_SERVICE);
+            } else {
+                ActionListener.completeWith(onAcquired, () -> acquiredPrimaryPermits);
+            }
         }
     }