Use correct recovery priorities

benchmarks/src/main/java/org/elasticsearch/benchmark/routing/allocation/ShardsAvailabilityHealthIndicatorBenchmark.java

@@ -144,7 +144,8 @@ public void setUp() throws Exception {
                     true,
                     RecoverySource.ExistingStoreRecoverySource.INSTANCE,
                     decidersNoUnassignedInfo,
-                    ShardRouting.Role.DEFAULT
+                    ShardRouting.Role.DEFAULT,
+                    ShardRouting.RecoveryPriority.UNASSIGNED_EXISTING
                 );
                 shardBuilder.addShard(shardRouting);
                 if (shardIdNumber < numReplicas) {
@@ -154,7 +155,8 @@ public void setUp() throws Exception {
                             false,
                             RecoverySource.EmptyStoreRecoverySource.INSTANCE,
                             decidersNoUnassignedInfo,
-                            ShardRouting.Role.DEFAULT
+                            ShardRouting.Role.DEFAULT,
+                            ShardRouting.RecoveryPriority.UNASSIGNED_EXISTING
                         )
                     );
                 }

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamAutoshardingIT.java

@@ -520,7 +520,8 @@ private static ShardStats getShardStats(IndexMetadata indexMeta, int shardIndex,
             true,
             RecoverySource.EmptyStoreRecoverySource.INSTANCE,
             new UnassignedInfo(UnassignedInfo.Reason.INDEX_CREATED, null),
-            ShardRouting.Role.DEFAULT
+            ShardRouting.Role.DEFAULT,
+            ShardRouting.RecoveryPriority.UNASSIGNED_NEW
         );
         shardRouting = shardRouting.initialize(assignedShardNodeId, null, ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE);
         shardRouting = shardRouting.moveToStarted(ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE);

modules/ip-location/src/test/java/org/elasticsearch/ingest/geoip/DatabaseNodeServiceTests.java

@@ -1017,7 +1017,8 @@ private static ClusterState createClusterState(
             true,
             RecoverySource.ExistingStoreRecoverySource.INSTANCE,
             new UnassignedInfo(UnassignedInfo.Reason.INDEX_CREATED, ""),
-            ShardRouting.Role.DEFAULT
+            ShardRouting.Role.DEFAULT,
+            ShardRouting.RecoveryPriority.UNASSIGNED_NEW
         );
         String nodeId = ESTestCase.randomAlphaOfLength(8);
         shardRouting = shardRouting.initialize(nodeId, null, shardRouting.getExpectedShardSize());

server/src/internalClusterTest/java/org/elasticsearch/cluster/routing/allocation/decider/WriteLoadConstraintDeciderIT.java

@@ -927,7 +927,8 @@ private static ShardStats createShardStats(
             true,
             RecoverySource.EmptyStoreRecoverySource.INSTANCE,
             new UnassignedInfo(UnassignedInfo.Reason.INDEX_CREATED, null),
-            ShardRouting.Role.DEFAULT
+            ShardRouting.Role.DEFAULT,
+            ShardRouting.RecoveryPriority.UNASSIGNED_NEW
         );
         shardRouting = shardRouting.initialize(assignedShardNodeId, null, ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE);
         shardRouting = shardRouting.moveToStarted(ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE);

server/src/main/java/module-info.java

@@ -57,6 +57,7 @@
     requires org.apache.lucene.queryparser;
     requires org.apache.lucene.sandbox;
     requires org.apache.lucene.suggest;
+    requires org.jspecify;
 
     exports org.elasticsearch;
     exports org.elasticsearch.action;

server/src/main/java/org/elasticsearch/cluster/ClusterInfoSimulator.java

@@ -173,10 +173,12 @@ public void simulateAlreadyStartedShard(ShardRouting startedShard, @Nullable Str
         }
         final long expectedShardSize = startedShard.getExpectedShardSize();
         if (sourceNodeId != null) {
+            // Use an arbitrary recovery priority, it does not affect the simulation:
+            ShardRouting.RecoveryPriority recoveryPriority = ShardRouting.RecoveryPriority.RELOCATE_REBALANCING;
             final var relocatingShard = startedShard.moveToUnassigned(new UnassignedInfo(REINITIALIZED, "simulation"))
                 .initialize(sourceNodeId, null, expectedShardSize)
                 .moveToStarted(expectedShardSize)
-                .relocate(startedShard.currentNodeId(), expectedShardSize)
+                .relocate(startedShard.currentNodeId(), expectedShardSize, recoveryPriority)
                 .getTargetRelocatingShard();
             simulateShardStarted(relocatingShard, false);
         } else {

server/src/main/java/org/elasticsearch/cluster/routing/IndexRoutingTable.java

@@ -387,35 +387,60 @@ public Builder(ShardRoutingRoleStrategy shardRoutingRoleStrategy, Index index) {
          * Initializes a new empty index, as if it was created from an API.
          */
         public Builder initializeAsNew(IndexMetadata indexMetadata) {
-            return initializeEmpty(indexMetadata, new UnassignedInfo(UnassignedInfo.Reason.INDEX_CREATED, null), null);
+            return initializeEmpty(
+                indexMetadata,
+                new UnassignedInfo(UnassignedInfo.Reason.INDEX_CREATED, null),
+                ShardRouting.RecoveryPriority.UNASSIGNED_NEW,
+                null
+            );
         }
 
         /**
          * Initializes an existing index.
          */
         public Builder initializeAsRecovery(IndexMetadata indexMetadata) {
-            return initializeEmpty(indexMetadata, new UnassignedInfo(UnassignedInfo.Reason.CLUSTER_RECOVERED, null), null);
+            return initializeEmpty(
+                indexMetadata,
+                new UnassignedInfo(UnassignedInfo.Reason.CLUSTER_RECOVERED, null),
+                ShardRouting.RecoveryPriority.UNASSIGNED_EXISTING,
+                null
+            );
         }
 
         /**
          * Initializes a new index caused by dangling index imported.
          */
         public Builder initializeAsFromDangling(IndexMetadata indexMetadata) {
-            return initializeEmpty(indexMetadata, new UnassignedInfo(UnassignedInfo.Reason.DANGLING_INDEX_IMPORTED, null), null);
+            return initializeEmpty(
+                indexMetadata,
+                new UnassignedInfo(UnassignedInfo.Reason.DANGLING_INDEX_IMPORTED, null),
+                ShardRouting.RecoveryPriority.UNASSIGNED_EXISTING,
+                null
+            );
         }
 
         /**
          * Initializes a new empty index, as a result of opening a closed index.
          */
         public Builder initializeAsFromCloseToOpen(IndexMetadata indexMetadata, IndexRoutingTable indexRoutingTable) {
-            return initializeEmpty(indexMetadata, new UnassignedInfo(UnassignedInfo.Reason.INDEX_REOPENED, null), indexRoutingTable);
+            return initializeEmpty(
+                indexMetadata,
+                new UnassignedInfo(UnassignedInfo.Reason.INDEX_REOPENED, null),
+                ShardRouting.RecoveryPriority.UNASSIGNED_NEW,
+                indexRoutingTable
+            );
         }
 
         /**
          * Initializes a new empty index, as a result of closing an opened index.
          */
         public Builder initializeAsFromOpenToClose(IndexMetadata indexMetadata, IndexRoutingTable indexRoutingTable) {
-            return initializeEmpty(indexMetadata, new UnassignedInfo(UnassignedInfo.Reason.INDEX_CLOSED, null), indexRoutingTable);
+            return initializeEmpty(
+                indexMetadata,
+                new UnassignedInfo(UnassignedInfo.Reason.INDEX_CLOSED, null),
+                ShardRouting.RecoveryPriority.UNASSIGNED_NEW,
+                indexRoutingTable
+            );
         }
 
         /**
@@ -486,7 +511,8 @@ private Builder initializeAsRestore(
                                 primary,
                                 primary ? EmptyStoreRecoverySource.INSTANCE : PeerRecoverySource.INSTANCE,
                                 unassignedInfo,
-                                shardRoutingRoleStrategy.newRestoredRole(i)
+                                shardRoutingRoleStrategy.newRestoredRole(i),
+                                ShardRouting.RecoveryPriority.UNASSIGNED_NEW
                             )
                         );
                     } else {
@@ -496,7 +522,8 @@ private Builder initializeAsRestore(
                                 primary,
                                 primary ? recoverySource : PeerRecoverySource.INSTANCE,
                                 withLastAllocatedNodeId(unassignedInfo, previousNodes, i),
-                                shardRoutingRoleStrategy.newRestoredRole(i)
+                                shardRoutingRoleStrategy.newRestoredRole(i),
+                                asNew ? ShardRouting.RecoveryPriority.UNASSIGNED_NEW : ShardRouting.RecoveryPriority.UNASSIGNED_EXISTING
                             )
                         );
                     }
@@ -512,7 +539,8 @@ private Builder initializeAsRestore(
         private Builder initializeEmpty(
             IndexMetadata indexMetadata,
             UnassignedInfo unassignedInfo,
-            @Nullable IndexRoutingTable previousIndexRoutingTable
+            ShardRouting.RecoveryPriority recoveryPriority,
+            IndexRoutingTable previousIndexRoutingTable
         ) {
             assert indexMetadata.getIndex().equals(index);
             assert previousIndexRoutingTable == null || previousIndexRoutingTable.size() == indexMetadata.getNumberOfShards();
@@ -550,7 +578,8 @@ private Builder initializeEmpty(
                             primary,
                             primary ? primaryRecoverySource : PeerRecoverySource.INSTANCE,
                             withLastAllocatedNodeId(shardUnassignedInfo, previousNodes, i),
-                            shardRoutingRoleStrategy.newEmptyRole(i)
+                            shardRoutingRoleStrategy.newEmptyRole(i),
+                            recoveryPriority
                         )
                     );
                 }
@@ -614,7 +643,8 @@ public Builder addReplica(ShardRouting.Role role) {
                         false,
                         PeerRecoverySource.INSTANCE,
                         new UnassignedInfo(UnassignedInfo.Reason.REPLICA_ADDED, null),
-                        role
+                        role,
+                        ShardRouting.RecoveryPriority.UNASSIGNED_NEW
                     )
                 );
             }

server/src/main/java/org/elasticsearch/cluster/routing/ShardRouting.java

@@ -220,7 +220,8 @@ public static ShardRouting newUnassigned(
         boolean primary,
         RecoverySource recoverySource,
         UnassignedInfo unassignedInfo,
-        Role role
+        Role role,
+        RecoveryPriority recoveryPriority
     ) {
         return new ShardRouting(
             shardId,
@@ -229,9 +230,7 @@ public static ShardRouting newUnassigned(
             primary,
             ShardRoutingState.UNASSIGNED,
             recoverySource,
-            // TODO: Populate this with the correct priority. For now, we use the higher of the two choices for a recovery from unassigned.
-            // As long as master-side recovery throttling is in effect, the priority is not that important.
-            RecoveryPriority.UNASSIGNED_EXISTING,
+            recoveryPriority,
             unassignedInfo,
             RelocationFailureInfo.NO_FAILURES,
             null,
@@ -592,7 +591,17 @@ public ShardRouting initialize(String nodeId, @Nullable String existingAllocatio
      *
      * @param relocatingNodeId id of the node to relocate the shard
      */
+    @Deprecated
     public ShardRouting relocate(String relocatingNodeId, long expectedShardSize) {
+        return relocate(relocatingNodeId, expectedShardSize, RecoveryPriority.RELOCATION_CAN_REMAIN_NO);
+    }
+
+    /**
+     * Relocate the shard to another node.
+     *
+     * @param relocatingNodeId id of the node to relocate the shard
+     */
+    public ShardRouting relocate(String relocatingNodeId, long expectedShardSize, RecoveryPriority recoveryPriority) {
         assert state == ShardRoutingState.STARTED : "current shard has to be started in order to be relocated " + this;
         return new ShardRouting(
             shardId,
@@ -601,9 +610,7 @@ public ShardRouting relocate(String relocatingNodeId, long expectedShardSize) {
             primary,
             ShardRoutingState.RELOCATING,
             recoverySource,
-            // TODO: Populate this with the correct priority. For now, we use the highest of the three choices for a relocation.
-            // As long as master-side recovery throttling is in effect, the priority is not that important.
-            RecoveryPriority.RELOCATION_CAN_REMAIN_NO,
+            recoveryPriority,
             null,
             relocationFailureInfo,
             AllocationId.newRelocation(allocationId),

server/src/main/java/org/elasticsearch/cluster/routing/allocation/allocator/BalancedShardsAllocator.java

@@ -1667,7 +1667,7 @@ private boolean tryRelocateShard(ModelNode minNode, ModelNode maxNode, ProjectIn
                         canAllocateOrRebalance == Type.YES
                             /* only allocate on the cluster if we are not throttled */
                             ? routingNodes.relocateShard(shard, minNode.getNodeId(), shardSize, REBALANCE_REASON, allocation.changes()).v1()
-                            : shard.relocate(minNode.getNodeId(), shardSize)
+                            : shard.relocate(minNode.getNodeId(), shardSize, ShardRouting.RecoveryPriority.RELOCATE_REBALANCING)
                     );
                     return true;
                 }

server/src/main/java/org/elasticsearch/cluster/routing/allocation/decider/ThrottlingAllocationDecider.java

@@ -130,7 +130,7 @@ private void setPrimariesInitialRecoveries(int primariesInitialRecoveries) {
     public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
         if (shardRouting.primary() && shardRouting.unassigned()) {
             // Primary is unassigned, means we are going to do recovery from store, snapshot or local shards
-            assert initializingShard(shardRouting, node.nodeId()).recoverySource().getType() != RecoverySource.Type.PEER;
+            assert initializingShardRecoverySourceType(shardRouting, node.nodeId()) != RecoverySource.Type.PEER;
 
             if (allocation.isSimulating()) {
                 return allocation.decision(Decision.YES, NAME, "primary allocation is not throttled when simulating");
@@ -170,7 +170,7 @@ public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, Routing
             return allocation.decision(Decision.YES, NAME, "replica allocation is not throttled when simulating");
         } else {
             // Peer recovery
-            assert initializingShard(shardRouting, node.nodeId()).recoverySource().getType() == RecoverySource.Type.PEER;
+            assert initializingShardRecoverySourceType(shardRouting, node.nodeId()) == RecoverySource.Type.PEER;
 
             // Allocating a shard to this node will increase the incoming recoveries
             int currentInRecoveries = allocation.routingNodes().getIncomingRecoveries(node.nodeId());
@@ -231,9 +231,9 @@ public Decision canForceAllocateDuringReplace(ShardRouting shardRouting, Routing
      * - the started shard routing in case if we want to check if we can relocate to this node.
      * - the relocating shard routing if we want to relocate to this node now instead.
      *
-     * This method returns the corresponding initializing shard that would be allocated to this node.
+     * This method returns the recovery source type of the corresponding initializing shard that would be allocated to this node.
      */
-    public static ShardRouting initializingShard(ShardRouting shardRouting, String currentNodeId) {
+    public static RecoverySource.Type initializingShardRecoverySourceType(ShardRouting shardRouting, String currentNodeId) {
         final ShardRouting initializingShard;
         if (shardRouting.unassigned()) {
             initializingShard = shardRouting.initialize(currentNodeId, null, ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE);
@@ -247,14 +247,17 @@ public static ShardRouting initializingShard(ShardRouting shardRouting, String c
                 .initialize(currentNodeId, null, ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE);
         } else if (shardRouting.relocating()) {
             initializingShard = shardRouting.cancelRelocation()
-                .relocate(currentNodeId, ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE)
+                .relocate(currentNodeId, ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE, shardRouting.recoveryPriority())
                 .getTargetRelocatingShard();
         } else {
             assert shardRouting.started();
-            initializingShard = shardRouting.relocate(currentNodeId, ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE)
+            // Use an arbitrary recovery priority, it makes no difference to teh recovery source type:
+            ShardRouting.RecoveryPriority recoveryPriority = ShardRouting.RecoveryPriority.RELOCATION_CAN_REMAIN_NO;
+            initializingShard = shardRouting.relocate(currentNodeId, ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE, recoveryPriority)
                 .getTargetRelocatingShard();
         }
         assert initializingShard.initializing();
-        return initializingShard;
+        assert initializingShard.recoverySource() != null;
+        return initializingShard.recoverySource().getType();
     }
 }

server/src/test/java/org/elasticsearch/action/admin/cluster/node/stats/NodeStatsTests.java

@@ -681,7 +681,8 @@ private static ShardStats createShardStats(ShardId shardId) {
             true,
             RecoverySource.EmptyStoreRecoverySource.INSTANCE,
             new UnassignedInfo(UnassignedInfo.Reason.INDEX_CREATED, "message"),
-            ShardRouting.Role.DEFAULT
+            ShardRouting.Role.DEFAULT,
+            ShardRouting.RecoveryPriority.UNASSIGNED_NEW
         );
         Path path = createTempDir().resolve("indices")
             .resolve(shardRouting.shardId().getIndex().getUUID())

server/src/test/java/org/elasticsearch/action/admin/cluster/stats/VersionStatsTests.java

@@ -104,7 +104,8 @@ public void testCreation() {
             true,
             RecoverySource.EmptyStoreRecoverySource.INSTANCE,
             new UnassignedInfo(UnassignedInfo.Reason.INDEX_CREATED, "message"),
-            ShardRouting.Role.DEFAULT
+            ShardRouting.Role.DEFAULT,
+            ShardRouting.RecoveryPriority.UNASSIGNED_NEW
         );
         Path path = createTempDir().resolve("indices")
             .resolve(shardRouting.shardId().getIndex().getUUID())

server/src/test/java/org/elasticsearch/action/admin/indices/recovery/RecoveryResponseTests.java

@@ -50,7 +50,8 @@ public void testChunkedToXContent() {
                                         false,
                                         RecoverySource.PeerRecoverySource.INSTANCE,
                                         new UnassignedInfo(UnassignedInfo.Reason.INDEX_CREATED, null),
-                                        ShardRouting.Role.DEFAULT
+                                        ShardRouting.Role.DEFAULT,
+                                        ShardRouting.RecoveryPriority.UNASSIGNED_NEW
                                     ).initialize(sourceNode.getId(), null, randomNonNegativeLong()),
                                     sourceNode,
                                     targetNode

server/src/test/java/org/elasticsearch/action/admin/indices/rollover/TransportRolloverActionTests.java

@@ -931,7 +931,8 @@ public static IndicesStatsResponse randomIndicesStatsResponse(final IndexMetadat
                     primary,
                     primary ? RecoverySource.EmptyStoreRecoverySource.INSTANCE : RecoverySource.PeerRecoverySource.INSTANCE,
                     new UnassignedInfo(UnassignedInfo.Reason.INDEX_CREATED, null),
-                    ShardRouting.Role.DEFAULT
+                    ShardRouting.Role.DEFAULT,
+                    ShardRouting.RecoveryPriority.UNASSIGNED_NEW
                 );
                 shardRouting = shardRouting.initialize("node-0", null, ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE);
                 shardRouting = shardRouting.moveToStarted(ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE);

server/src/test/java/org/elasticsearch/action/bulk/BulkPrimaryExecutionContextTests.java

@@ -182,6 +182,13 @@ public Translog.Location getTranslogLocation() {
 
     private ShardRouting newShardRouting(ShardId shardId, ShardRouting.Role role) {
         final UnassignedInfo unassignedInfo = new UnassignedInfo(UnassignedInfo.Reason.INDEX_CREATED, "_message");
-        return ShardRouting.newUnassigned(shardId, true, RecoverySource.ExistingStoreRecoverySource.INSTANCE, unassignedInfo, role);
+        return ShardRouting.newUnassigned(
+            shardId,
+            true,
+            RecoverySource.ExistingStoreRecoverySource.INSTANCE,
+            unassignedInfo,
+            role,
+            ShardRouting.RecoveryPriority.UNASSIGNED_NEW
+        );
     }
 }