Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Purge unused node info and transport addr info after fetching shard stats. #77266

Closed
wants to merge 8 commits into from
Closed

Purge unused node info and transport addr info after fetching shard stats. #77266

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
70445bd
fix shard fetch memory issue
howardhuanghua Sep 3, 2021
e7f9015
Merge branch 'master' into fix_shard_fetch_issue
howardhuanghua Sep 3, 2021
d159722
use cached result in NodeEntry
howardhuanghua Sep 6, 2021
9730b3c
Merge remote-tracking branch 'origin' into fix_shard_fetch_issue
howardhuanghua Sep 6, 2021
6f5f26a
revert base node response
howardhuanghua Sep 6, 2021
416199a
Merge remote-tracking branch 'origin/master' into fix_shard_fetch_issue
howardhuanghua Sep 15, 2021
d91f25d
remove dead code
howardhuanghua Sep 15, 2021
30e817c
Merge remote-tracking branch 'origin/master' into fix_shard_fetch_issue
howardhuanghua Sep 15, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions server/src/main/java/org/elasticsearch/action/support/nodes/BaseNodeResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
public abstract class BaseNodeResponse extends TransportResponse {

private DiscoveryNode node;
private String nodeId;

protected BaseNodeResponse(StreamInput in) throws IOException {
super(in);
Expand All @@ -32,10 +33,25 @@ protected BaseNodeResponse(DiscoveryNode node) {
this.node = node;
}

public void purgeNodeInfo() {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd rather not add to this class, nor make it any more mutable than it already is. Instead, let's copy the information we need out of the response and into a different object which is smaller and can be kept around for longer.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK, I throught to add new objects to replace NodeGatewayStartedShards and NodeStoreFilesMetadata, change their base class to not extend from BaseNodesResponse. But it seems it would have a lot of dependencies, I will try to check the dependencies.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think we need to change NodeGatewayStartedShards or NodeStoreFilesMetadata either. My understanding is that the memory is all held by this field:

elasticsearch/server/src/main/java/org/elasticsearch/gateway/AsyncShardFetch.java

Line 62 in 944b3f3

private final Map<String, NodeEntry<T>> cache = new HashMap<>();

Its type parameter T is required to be a BaseNodesResponse but really this parameter represents both the response type and the cache entry type and I don't think there's any reason for those types to be the same. Something like this kind of direction should work I think:

diff --git a/server/src/main/java/org/elasticsearch/gateway/AsyncShardFetch.java b/server/src/main/java/org/elasticsearch/gateway/AsyncShardFetch.java
index 3a7cdfee84e..10c022d44c3 100644
--- a/server/src/main/java/org/elasticsearch/gateway/AsyncShardFetch.java
+++ b/server/src/main/java/org/elasticsearch/gateway/AsyncShardFetch.java
@@ -45,7 +45,7 @@ import static java.util.Collections.emptySet;
  * and once the results are back, it makes sure to schedule a reroute to make sure those results will
  * be taken into account.
  */
-public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Releasable {
+public abstract class AsyncShardFetch<Response extends BaseNodeResponse, Cached> implements Releasable {
 
     /**
      * An action that lists the relevant shard data that needs to be fetched.
@@ -58,20 +58,20 @@ public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Rel
     protected final String type;
     protected final ShardId shardId;
     protected final String customDataPath;
-    private final Lister<BaseNodesResponse<T>, T> action;
-    private final Map<String, NodeEntry<T>> cache = new HashMap<>();
+    private final Lister<BaseNodesResponse<Response>, Response> action;
+    private final Map<String, NodeEntry<Cached>> cache = new HashMap<>();
     private final Set<String> nodesToIgnore = new HashSet<>();
     private final AtomicLong round = new AtomicLong();
     private boolean closed;
 
     @SuppressWarnings("unchecked")
     protected AsyncShardFetch(Logger logger, String type, ShardId shardId, String customDataPath,
-                              Lister<? extends BaseNodesResponse<T>, T> action) {
+                              Lister<? extends BaseNodesResponse<Response>, Response> action) {
         this.logger = logger;
         this.type = type;
         this.shardId = Objects.requireNonNull(shardId);
         this.customDataPath = Objects.requireNonNull(customDataPath);
-        this.action = (Lister<BaseNodesResponse<T>, T>) action;
+        this.action = (Lister<BaseNodesResponse<Response>, Response>) action;
     }
 
     @Override
@@ -84,7 +84,7 @@ public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Rel
      */
     public synchronized int getNumberOfInFlightFetches() {
         int count = 0;
-        for (NodeEntry<T> nodeEntry : cache.values()) {
+        for (NodeEntry<Cached> nodeEntry : cache.values()) {
             if (nodeEntry.isFetching()) {
                 count++;
             }
@@ -99,18 +99,18 @@ public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Rel
      * The ignoreNodes are nodes that are supposed to be ignored for this round, since fetching is async, we need
      * to keep them around and make sure we add them back when all the responses are fetched and returned.
      */
-    public synchronized FetchResult<T> fetchData(DiscoveryNodes nodes, Set<String> ignoreNodes) {
+    public synchronized FetchResult<Cached> fetchData(DiscoveryNodes nodes, Set<String> ignoreNodes) {
         if (closed) {
             throw new IllegalStateException(shardId + ": can't fetch data on closed async fetch");
         }
         nodesToIgnore.addAll(ignoreNodes);
         fillShardCacheWithDataNodes(cache, nodes);
-        List<NodeEntry<T>> nodesToFetch = findNodesToFetch(cache);
+        List<NodeEntry<Cached>> nodesToFetch = findNodesToFetch(cache);
         if (nodesToFetch.isEmpty() == false) {
             // mark all node as fetching and go ahead and async fetch them
             // use a unique round id to detect stale responses in processAsyncFetch
             final long fetchingRound = round.incrementAndGet();
-            for (NodeEntry<T> nodeEntry : nodesToFetch) {
+            for (NodeEntry<Cached> nodeEntry : nodesToFetch) {
                 nodeEntry.markAsFetching(fetchingRound);
             }
             DiscoveryNode[] discoNodesToFetch = nodesToFetch.stream().map(NodeEntry::getNodeId).map(nodes::get)
@@ -123,12 +123,12 @@ public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Rel
             return new FetchResult<>(shardId, null, emptySet());
         } else {
             // nothing to fetch, yay, build the return value
-            Map<DiscoveryNode, T> fetchData = new HashMap<>();
+            Map<DiscoveryNode, Cached> fetchData = new HashMap<>();
             Set<String> failedNodes = new HashSet<>();
-            for (Iterator<Map.Entry<String, NodeEntry<T>>> it = cache.entrySet().iterator(); it.hasNext(); ) {
-                Map.Entry<String, NodeEntry<T>> entry = it.next();
+            for (Iterator<Map.Entry<String, NodeEntry<Cached>>> it = cache.entrySet().iterator(); it.hasNext(); ) {
+                Map.Entry<String, NodeEntry<Cached>> entry = it.next();
                 String nodeId = entry.getKey();
-                NodeEntry<T> nodeEntry = entry.getValue();
+                NodeEntry<Cached> nodeEntry = entry.getValue();
 
                 DiscoveryNode node = nodes.get(nodeId);
                 if (node != null) {
@@ -153,17 +153,19 @@ public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Rel
             if (failedNodes.isEmpty() == false || allIgnoreNodes.isEmpty() == false) {
                 reroute(shardId, "nodes failed [" + failedNodes.size() + "], ignored [" + allIgnoreNodes.size() + "]");
             }
-            return new FetchResult<>(shardId, fetchData, allIgnoreNodes);
+            return new FetchResult<Cached>(shardId, fetchData, allIgnoreNodes);
         }
     }
 
+    protected abstract Cached extract(Response value);
+
     /**
      * Called by the response handler of the async action to fetch data. Verifies that its still working
      * on the same cache generation, otherwise the results are discarded. It then goes and fills the relevant data for
      * the shard (response + failures), issuing a reroute at the end of it to make sure there will be another round
      * of allocations taking this new data into account.
      */
-    protected synchronized void processAsyncFetch(List<T> responses, List<FailedNodeException> failures, long fetchingRound) {
+    protected synchronized void processAsyncFetch(List<Response> responses, List<FailedNodeException> failures, long fetchingRound) {
         if (closed) {
             // we are closed, no need to process this async fetch at all
             logger.trace("{} ignoring fetched [{}] results, already closed", shardId, type);
@@ -172,8 +174,8 @@ public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Rel
         logger.trace("{} processing fetched [{}] results", shardId, type);
 
         if (responses != null) {
-            for (T response : responses) {
-                NodeEntry<T> nodeEntry = cache.get(response.getNode().getId());
+            for (Response response : responses) {
+                NodeEntry<Cached> nodeEntry = cache.get(response.getNode().getId());
                 if (nodeEntry != null) {
                     if (nodeEntry.getFetchingRound() != fetchingRound) {
                         assert nodeEntry.getFetchingRound() > fetchingRound : "node entries only replaced by newer rounds";
@@ -185,7 +187,7 @@ public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Rel
                     } else {
                         // if the entry is there, for the right fetching round and not marked as failed already, process it
                         logger.trace("{} marking {} as done for [{}], result is [{}]", shardId, nodeEntry.getNodeId(), type, response);
-                        nodeEntry.doneFetching(response);
+                        nodeEntry.doneFetching(extract(response));
                     }
                 }
             }
@@ -193,7 +195,7 @@ public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Rel
         if (failures != null) {
             for (FailedNodeException failure : failures) {
                 logger.trace("{} processing failure {} for [{}]", shardId, failure, type);
-                NodeEntry<T> nodeEntry = cache.get(failure.nodeId());
+                NodeEntry<Cached> nodeEntry = cache.get(failure.nodeId());
                 if (nodeEntry != null) {
                     if (nodeEntry.getFetchingRound() != fetchingRound) {
                         assert nodeEntry.getFetchingRound() > fetchingRound : "node entries only replaced by newer rounds";
@@ -235,12 +237,12 @@ public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Rel
      * Fills the shard fetched data with new (data) nodes and a fresh NodeEntry, and removes from
      * it nodes that are no longer part of the state.
      */
-    private void fillShardCacheWithDataNodes(Map<String, NodeEntry<T>> shardCache, DiscoveryNodes nodes) {
+    private void fillShardCacheWithDataNodes(Map<String, NodeEntry<Cached>> shardCache, DiscoveryNodes nodes) {
         // verify that all current data nodes are there
         for (ObjectObjectCursor<String, DiscoveryNode> cursor : nodes.getDataNodes()) {
             DiscoveryNode node = cursor.value;
             if (shardCache.containsKey(node.getId()) == false) {
-                shardCache.put(node.getId(), new NodeEntry<T>(node.getId()));
+                shardCache.put(node.getId(), new NodeEntry<Cached>(node.getId()));
             }
         }
         // remove nodes that are not longer part of the data nodes set
@@ -251,9 +253,9 @@ public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Rel
      * Finds all the nodes that need to be fetched. Those are nodes that have no
      * data, and are not in fetch mode.
      */
-    private List<NodeEntry<T>> findNodesToFetch(Map<String, NodeEntry<T>> shardCache) {
-        List<NodeEntry<T>> nodesToFetch = new ArrayList<>();
-        for (NodeEntry<T> nodeEntry : shardCache.values()) {
+    private List<NodeEntry<Cached>> findNodesToFetch(Map<String, NodeEntry<Cached>> shardCache) {
+        List<NodeEntry<Cached>> nodesToFetch = new ArrayList<>();
+        for (NodeEntry<Cached> nodeEntry : shardCache.values()) {
             if (nodeEntry.hasData() == false && nodeEntry.isFetching() == false) {
                 nodesToFetch.add(nodeEntry);
             }
@@ -264,8 +266,8 @@ public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Rel
     /**
      * Are there any nodes that are fetching data?
      */
-    private boolean hasAnyNodeFetching(Map<String, NodeEntry<T>> shardCache) {
-        for (NodeEntry<T> nodeEntry : shardCache.values()) {
+    private boolean hasAnyNodeFetching(Map<String, NodeEntry<Cached>> shardCache) {
+        for (NodeEntry<Cached> nodeEntry : shardCache.values()) {
             if (nodeEntry.isFetching()) {
                 return true;
             }
@@ -279,9 +281,9 @@ public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Rel
     // visible for testing
     void asyncFetch(final DiscoveryNode[] nodes, long fetchingRound) {
         logger.trace("{} fetching [{}] from {}", shardId, type, nodes);
-        action.list(shardId, customDataPath, nodes, new ActionListener<BaseNodesResponse<T>>() {
+        action.list(shardId, customDataPath, nodes, new ActionListener<BaseNodesResponse<Response>>() {
             @Override
-            public void onResponse(BaseNodesResponse<T> response) {
+            public void onResponse(BaseNodesResponse<Response> response) {
                 processAsyncFetch(response.getNodes(), response.failures(), fetchingRound);
             }
 
@@ -300,13 +302,13 @@ public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Rel
      * The result of a fetch operation. Make sure to first check {@link #hasData()} before
      * fetching the actual data.
      */
-    public static class FetchResult<T extends BaseNodeResponse> {
+    public static class FetchResult<U> {
 
         private final ShardId shardId;
-        private final Map<DiscoveryNode, T> data;
+        private final Map<DiscoveryNode, U> data;
         private final Set<String> ignoreNodes;
 
-        public FetchResult(ShardId shardId, Map<DiscoveryNode, T> data, Set<String> ignoreNodes) {
+        public FetchResult(ShardId shardId, Map<DiscoveryNode, U> data, Set<String> ignoreNodes) {
             this.shardId = shardId;
             this.data = data;
             this.ignoreNodes = ignoreNodes;
@@ -324,7 +326,7 @@ public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Rel
          * Returns the actual data, note, make sure to check {@link #hasData()} first and
          * only use this when there is an actual data.
          */
-        public Map<DiscoveryNode, T> getData() {
+        public Map<DiscoveryNode, U> getData() {
             assert data != null : "getData should only be called if there is data to be fetched, please check hasData first";
             return this.data;
         }

// set nodeId and purge DiscoveryNode and TransportAddr structures
this.nodeId = this.getNode().getId();
this.node = null;
this.remoteAddress(null);
}

public String getNodeId() {
if (this.nodeId == null && node != null) {
return node.getId();
}
return this.nodeId;
}

/**
* The node this information relates to.
*/
public DiscoveryNode getNode() {
assert node != null;
return node;
}

Expand Down
6 changes: 5 additions & 1 deletion server/src/main/java/org/elasticsearch/gateway/AsyncShardFetch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -370,11 +370,15 @@ void markAsFetching(long fetchingRound) {
this.fetchingRound = fetchingRound;
}

@SuppressWarnings("unchecked")
void doneFetching(T value) {
assert fetching : "setting value but not in fetching mode";
assert failure == null : "setting value when failure already set";
assert value instanceof BaseNodeResponse : "incorrect value type";
BaseNodeResponse response = (BaseNodeResponse)value;
response.purgeNodeInfo();
this.value = (T) response;
this.valueSet = true;
this.value = value;
this.fetching = false;
}

Expand Down
41 changes: 22 additions & 19 deletions server/src/main/java/org/elasticsearch/gateway/PrimaryShardAllocator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
import org.apache.logging.log4j.message.ParameterizedMessage;
import org.elasticsearch.cluster.metadata.IndexMetadata;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.cluster.routing.RecoverySource;
import org.elasticsearch.cluster.routing.RoutingNode;
import org.elasticsearch.cluster.routing.RoutingNodes;
Expand Down Expand Up @@ -120,7 +121,7 @@ public AllocateUnassignedDecision makeAllocationDecision(final ShardRouting unas
logger.debug("[{}][{}]: not allocating, number_of_allocated_shards_found [{}]",
unassignedShard.index(), unassignedShard.id(), nodeShardsResult.allocationsFound);
return AllocateUnassignedDecision.no(AllocationStatus.NO_VALID_SHARD_COPY,
explain ? buildNodeDecisions(null, shardState, inSyncAllocationIds) : null);
explain ? buildNodeDecisions(null, shardState, inSyncAllocationIds, allocation.nodes()) : null);
}
}

Expand All @@ -132,9 +133,9 @@ public AllocateUnassignedDecision makeAllocationDecision(final ShardRouting unas
boolean throttled = false;
if (nodesToAllocate.yesNodeShards.isEmpty() == false) {
DecidedNode decidedNode = nodesToAllocate.yesNodeShards.get(0);
node = allocation.nodes().get(decidedNode.nodeShardState.getNodeId());
logger.debug("[{}][{}]: allocating [{}] to [{}] on primary allocation",
unassignedShard.index(), unassignedShard.id(), unassignedShard, decidedNode.nodeShardState.getNode());
node = decidedNode.nodeShardState.getNode();
unassignedShard.index(), unassignedShard.id(), unassignedShard, node);
allocationId = decidedNode.nodeShardState.allocationId();
} else if (nodesToAllocate.throttleNodeShards.isEmpty() && nodesToAllocate.noNodeShards.isEmpty() == false) {
// The deciders returned a NO decision for all nodes with shard copies, so we check if primary shard
Expand All @@ -143,9 +144,9 @@ public AllocateUnassignedDecision makeAllocationDecision(final ShardRouting unas
if (nodesToAllocate.yesNodeShards.isEmpty() == false) {
final DecidedNode decidedNode = nodesToAllocate.yesNodeShards.get(0);
final NodeGatewayStartedShards nodeShardState = decidedNode.nodeShardState;
node = allocation.nodes().get(nodeShardState.getNodeId());
logger.debug("[{}][{}]: allocating [{}] to [{}] on forced primary allocation",
unassignedShard.index(), unassignedShard.id(), unassignedShard, nodeShardState.getNode());
node = nodeShardState.getNode();
unassignedShard.index(), unassignedShard.id(), unassignedShard, node);
allocationId = nodeShardState.allocationId();
} else if (nodesToAllocate.throttleNodeShards.isEmpty() == false) {
logger.debug("[{}][{}]: throttling allocation [{}] to [{}] on forced primary allocation",
Expand All @@ -165,7 +166,7 @@ public AllocateUnassignedDecision makeAllocationDecision(final ShardRouting unas

List<NodeAllocationResult> nodeResults = null;
if (explain) {
nodeResults = buildNodeDecisions(nodesToAllocate, shardState, inSyncAllocationIds);
nodeResults = buildNodeDecisions(nodesToAllocate, shardState, inSyncAllocationIds, allocation.nodes());
}
if (allocation.hasPendingAsyncFetch()) {
return AllocateUnassignedDecision.no(AllocationStatus.FETCHING_SHARD_DATA, nodeResults);
Expand All @@ -183,21 +184,22 @@ public AllocateUnassignedDecision makeAllocationDecision(final ShardRouting unas
*/
private static List<NodeAllocationResult> buildNodeDecisions(NodesToAllocate nodesToAllocate,
FetchResult<NodeGatewayStartedShards> fetchedShardData,
Set<String> inSyncAllocationIds) {
Set<String> inSyncAllocationIds,
final DiscoveryNodes discoveryNodes) {
List<NodeAllocationResult> nodeResults = new ArrayList<>();
Collection<NodeGatewayStartedShards> ineligibleShards;
if (nodesToAllocate != null) {
final Set<DiscoveryNode> discoNodes = new HashSet<>();
final Set<String> discoNodes = new HashSet<>();
nodeResults.addAll(Stream.of(nodesToAllocate.yesNodeShards, nodesToAllocate.throttleNodeShards, nodesToAllocate.noNodeShards)
.flatMap(Collection::stream)
.map(dnode -> {
discoNodes.add(dnode.nodeShardState.getNode());
return new NodeAllocationResult(dnode.nodeShardState.getNode(),
discoNodes.add(dnode.nodeShardState.getNodeId());
return new NodeAllocationResult(discoveryNodes.get(dnode.nodeShardState.getNodeId()),
shardStoreInfo(dnode.nodeShardState, inSyncAllocationIds),
dnode.decision);
}).collect(Collectors.toList()));
ineligibleShards = fetchedShardData.getData().values().stream().filter(shardData ->
discoNodes.contains(shardData.getNode()) == false
discoNodes.contains(shardData.getNodeId()) == false
).collect(Collectors.toList());
} else {
// there were no shard copies that were eligible for being assigned the allocation,
Expand All @@ -206,7 +208,8 @@ private static List<NodeAllocationResult> buildNodeDecisions(NodesToAllocate nod
}

nodeResults.addAll(ineligibleShards.stream().map(shardData ->
new NodeAllocationResult(shardData.getNode(), shardStoreInfo(shardData, inSyncAllocationIds), null)
new NodeAllocationResult(discoveryNodes.get(shardData.getNodeId()),
shardStoreInfo(shardData, inSyncAllocationIds), null)
).collect(Collectors.toList()));

return nodeResults;
Expand Down Expand Up @@ -235,28 +238,28 @@ protected static NodeShardsResult buildNodeShardsResult(ShardRouting shard, bool
List<NodeGatewayStartedShards> nodeShardStates = new ArrayList<>();
int numberOfAllocationsFound = 0;
for (NodeGatewayStartedShards nodeShardState : shardState.getData().values()) {
DiscoveryNode node = nodeShardState.getNode();
String nodeId = nodeShardState.getNodeId();
String allocationId = nodeShardState.allocationId();

if (ignoreNodes.contains(node.getId())) {
if (ignoreNodes.contains(nodeId)) {
continue;
}

if (nodeShardState.storeException() == null) {
if (allocationId == null) {
logger.trace("[{}] on node [{}] has no shard state information", shard, nodeShardState.getNode());
logger.trace("[{}] on node [{}] has no shard state information", shard, nodeShardState.getNodeId());
} else {
logger.trace("[{}] on node [{}] has allocation id [{}]", shard, nodeShardState.getNode(), allocationId);
logger.trace("[{}] on node [{}] has allocation id [{}]", shard, nodeShardState.getNodeId(), allocationId);
}
} else {
final String finalAllocationId = allocationId;
if (nodeShardState.storeException() instanceof ShardLockObtainFailedException) {
logger.trace(() -> new ParameterizedMessage("[{}] on node [{}] has allocation id [{}] but the store can not be " +
"opened as it's locked, treating as valid shard", shard, nodeShardState.getNode(), finalAllocationId),
"opened as it's locked, treating as valid shard", shard, nodeShardState.getNodeId(), finalAllocationId),
nodeShardState.storeException());
} else {
logger.trace(() -> new ParameterizedMessage("[{}] on node [{}] has allocation id [{}] but the store can not be " +
"opened, treating as no allocation id", shard, nodeShardState.getNode(), finalAllocationId),
"opened, treating as no allocation id", shard, nodeShardState.getNodeId(), finalAllocationId),
nodeShardState.storeException());
allocationId = null;
}
Expand Down Expand Up @@ -305,7 +308,7 @@ private static NodesToAllocate buildNodesToAllocate(RoutingAllocation allocation
List<DecidedNode> throttledNodeShards = new ArrayList<>();
List<DecidedNode> noNodeShards = new ArrayList<>();
for (NodeGatewayStartedShards nodeShardState : nodeShardStates) {
RoutingNode node = allocation.routingNodes().node(nodeShardState.getNode().getId());
RoutingNode node = allocation.routingNodes().node(nodeShardState.getNodeId());
if (node == null) {
continue;
}
Expand Down