Merge pull request #16096 from ywelsch/feature/prefer-previous-primary

Prefer nodes that previously held primary shard for primary shard allocation

Author: Yannick Welsch

core/src/main/java/org/elasticsearch/gateway/PrimaryShardAllocator.java

@@ -38,6 +38,7 @@
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -175,8 +176,8 @@ public boolean allocateUnassigned(RoutingAllocation allocation) {
      */
     protected NodesAndVersions buildAllocationIdBasedNodes(ShardRouting shard, boolean matchAnyShard, Set<String> ignoreNodes,
                                                            Set<String> lastActiveAllocationIds, AsyncShardFetch.FetchResult<TransportNodesListGatewayStartedShards.NodeGatewayStartedShards> shardState) {
-        List<DiscoveryNode> matchingNodes = new ArrayList<>();
-        List<DiscoveryNode> nonMatchingNodes = new ArrayList<>();
+        LinkedList<DiscoveryNode> matchingNodes = new LinkedList<>();
+        LinkedList<DiscoveryNode> nonMatchingNodes = new LinkedList<>();
         long highestVersion = -1;
         for (TransportNodesListGatewayStartedShards.NodeGatewayStartedShards nodeShardState : shardState.getData().values()) {
             DiscoveryNode node = nodeShardState.getNode();
@@ -200,10 +201,18 @@ protected NodesAndVersions buildAllocationIdBasedNodes(ShardRouting shard, boole
 
             if (allocationId != null) {
                 if (lastActiveAllocationIds.contains(allocationId)) {
-                    matchingNodes.add(node);
+                    if (nodeShardState.primary()) {
+                        matchingNodes.addFirst(node);
+                    } else {
+                        matchingNodes.addLast(node);
+                    }
                     highestVersion = Math.max(highestVersion, nodeShardState.version());
                 } else if (matchAnyShard) {
-                    nonMatchingNodes.add(node);
+                    if (nodeShardState.primary()) {
+                        nonMatchingNodes.addFirst(node);
+                    } else {
+                        nonMatchingNodes.addLast(node);
+                    }
                     highestVersion = Math.max(highestVersion, nodeShardState.version());
                 }
             }

core/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java

@@ -139,7 +139,7 @@ protected NodeGatewayStartedShards nodeOperation(NodeRequest request) {
                     } catch (Exception exception) {
                         logger.trace("{} can't open index for shard [{}] in path [{}]", exception, shardId, shardStateMetaData, (shardPath != null) ? shardPath.resolveIndex() : "");
                         String allocationId = shardStateMetaData.allocationId != null ? shardStateMetaData.allocationId.getId() : null;
-                        return new NodeGatewayStartedShards(clusterService.localNode(), shardStateMetaData.version, allocationId, exception);
+                        return new NodeGatewayStartedShards(clusterService.localNode(), shardStateMetaData.version, allocationId, shardStateMetaData.primary, exception);
                     }
                 }
                 // old shard metadata doesn't have the actual index UUID so we need to check if the actual uuid in the metadata
@@ -150,11 +150,11 @@ protected NodeGatewayStartedShards nodeOperation(NodeRequest request) {
                 } else {
                     logger.debug("{} shard state info found: [{}]", shardId, shardStateMetaData);
                     String allocationId = shardStateMetaData.allocationId != null ? shardStateMetaData.allocationId.getId() : null;
-                    return new NodeGatewayStartedShards(clusterService.localNode(), shardStateMetaData.version, allocationId);
+                    return new NodeGatewayStartedShards(clusterService.localNode(), shardStateMetaData.version, allocationId, shardStateMetaData.primary);
                 }
             }
             logger.trace("{} no local shard info found", shardId);
-            return new NodeGatewayStartedShards(clusterService.localNode(), -1, null);
+            return new NodeGatewayStartedShards(clusterService.localNode(), -1, null, false);
         } catch (Exception e) {
             throw new ElasticsearchException("failed to load started shards", e);
         }
@@ -279,18 +279,20 @@ public static class NodeGatewayStartedShards extends BaseNodeResponse {
 
         private long version = -1;
         private String allocationId = null;
+        private boolean primary = false;
         private Throwable storeException = null;
 
         public NodeGatewayStartedShards() {
         }
-        public NodeGatewayStartedShards(DiscoveryNode node, long version, String allocationId) {
-            this(node, version, allocationId, null);
+        public NodeGatewayStartedShards(DiscoveryNode node, long version, String allocationId, boolean primary) {
+            this(node, version, allocationId, primary, null);
         }
 
-        public NodeGatewayStartedShards(DiscoveryNode node, long version, String allocationId, Throwable storeException) {
+        public NodeGatewayStartedShards(DiscoveryNode node, long version, String allocationId, boolean primary, Throwable storeException) {
             super(node);
             this.version = version;
             this.allocationId = allocationId;
+            this.primary = primary;
             this.storeException = storeException;
         }
 
@@ -302,6 +304,10 @@ public String allocationId() {
             return this.allocationId;
         }
 
+        public boolean primary() {
+            return this.primary;
+        }
+
         public Throwable storeException() {
             return this.storeException;
         }
@@ -311,6 +317,7 @@ public void readFrom(StreamInput in) throws IOException {
             super.readFrom(in);
             version = in.readLong();
             allocationId = in.readOptionalString();
+            primary = in.readBoolean();
             if (in.readBoolean()) {
                 storeException = in.readThrowable();
             }
@@ -321,6 +328,7 @@ public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             out.writeLong(version);
             out.writeOptionalString(allocationId);
+            out.writeBoolean(primary);
             if (storeException != null) {
                 out.writeBoolean(true);
                 out.writeThrowable(storeException);