executing shard recovery in project context

Author: ywangd

server/src/main/java/org/elasticsearch/indices/IndicesService.java

@@ -75,6 +75,7 @@
 import org.elasticsearch.core.AbstractRefCounted;
 import org.elasticsearch.core.CheckedConsumer;
 import org.elasticsearch.core.CheckedFunction;
+import org.elasticsearch.core.CheckedRunnable;
 import org.elasticsearch.core.IOUtils;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Releasable;
@@ -950,6 +951,7 @@ public synchronized void verifyIndexMetadata(IndexMetadata metadata, IndexMetada
 
     @Override
     public void createShard(
+        final ProjectId projectId,
         final ShardRouting shardRouting,
         final PeerRecoveryTargetService recoveryTargetService,
         final PeerRecoveryTargetService.RecoveryListener recoveryListener,
@@ -968,7 +970,7 @@ public void createShard(
         RecoveryState recoveryState = indexService.createRecoveryState(shardRouting, targetNode, sourceNode);
         IndexShard indexShard = indexService.createShard(shardRouting, globalCheckpointSyncer, retentionLeaseSyncer);
         indexShard.addShardFailureCallback(onShardFailure);
-        indexShard.startRecovery(
+        final CheckedRunnable<RuntimeException> recoveryRunnable = () -> indexShard.startRecovery(
             recoveryState,
             recoveryTargetService,
             postRecoveryMerger.maybeMergeAfterRecovery(indexService.getMetadata(), shardRouting, recoveryListener),
@@ -989,6 +991,7 @@ public void createShard(
             this,
             clusterStateVersion
         );
+        projectResolver.executeOnProject(projectId, recoveryRunnable);
     }
 
     @Override

server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java

@@ -26,6 +26,7 @@
 import org.elasticsearch.cluster.ClusterStateApplier;
 import org.elasticsearch.cluster.action.shard.ShardStateAction;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
@@ -781,6 +782,7 @@ private void createShardWhenLockAvailable(
         try {
             logger.debug("{} creating shard with primary term [{}], iteration [{}]", shardRouting.shardId(), primaryTerm, iteration);
             indicesService.createShard(
+                originalState.metadata().projectFor(shardRouting.index()).id(),
                 shardRouting,
                 recoveryTargetService,
                 new RecoveryListener(shardRouting, primaryTerm),
@@ -1330,6 +1332,7 @@ void removeIndex(
         /**
          * Creates a shard for the specified shard routing and starts recovery.
          *
+         * @param projectId              the project for the shard
          * @param shardRouting           the shard routing
          * @param recoveryTargetService  recovery service for the target
          * @param recoveryListener       a callback when recovery changes state (finishes or fails)
@@ -1343,6 +1346,7 @@ void removeIndex(
          * @throws IOException if an I/O exception occurs when creating the shard
          */
         void createShard(
+            ProjectId projectId,
             ShardRouting shardRouting,
             PeerRecoveryTargetService recoveryTargetService,
             PeerRecoveryTargetService.RecoveryListener recoveryListener,

test/framework/src/main/java/org/elasticsearch/indices/cluster/AbstractIndicesClusterStateServiceTestCase.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
@@ -242,6 +243,7 @@ public MockIndexService indexService(Index index) {
 
         @Override
         public void createShard(
+            final ProjectId projectId,
             final ShardRouting shardRouting,
             final PeerRecoveryTargetService recoveryTargetService,
             final PeerRecoveryTargetService.RecoveryListener recoveryListener,