Carry over shard exception failure to master node

Don't loose the shard exception failure when sending a shard failrue to the master node

core/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java

@@ -898,9 +898,8 @@ public abstract class TransportReplicationAction<Request extends ReplicationRequ
                                 onReplicaFailure(nodeId, exp);
                                 logger.trace("[{}] transport failure during replica request [{}] ", exp, node, replicaRequest);
                                 if (ignoreReplicaException(exp) == false) {
-                                    logger.warn("failed to perform " + actionName + " on remote replica " + node + shardIt.shardId(), exp);
-                                    shardStateAction.shardFailed(shard, indexMetaData.getIndexUUID(),
-                                            "Failed to perform [" + actionName + "] on replica, message [" + ExceptionsHelper.detailedMessage(exp) + "]");
+                                    logger.warn("{} failed to perform {} on node {}", exp, shardIt.shardId(), actionName, node);
+                                    shardStateAction.shardFailed(shard, indexMetaData.getIndexUUID(), "failed to perform " + actionName + " on replica on node " + node, exp);
                                 }
                             }
 

core/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.cluster.action.shard;
 
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
@@ -30,6 +31,7 @@ import org.elasticsearch.cluster.routing.*;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.routing.allocation.FailedRerouteAllocation;
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.inject.Inject;
@@ -77,22 +79,22 @@ public class ShardStateAction extends AbstractComponent {
         transportService.registerRequestHandler(SHARD_FAILED_ACTION_NAME, ShardRoutingEntry.class, ThreadPool.Names.SAME, new ShardFailedTransportHandler());
     }
 
-    public void shardFailed(final ShardRouting shardRouting, final String indexUUID, final String reason) {
+    public void shardFailed(final ShardRouting shardRouting, final String indexUUID, final String message, @Nullable final Throwable failure) {
         DiscoveryNode masterNode = clusterService.state().nodes().masterNode();
         if (masterNode == null) {
             logger.warn("can't send shard failed for {}, no master known.", shardRouting);
             return;
         }
-        innerShardFailed(shardRouting, indexUUID, reason, masterNode);
+        innerShardFailed(shardRouting, indexUUID, masterNode, message, failure);
     }
 
-    public void resendShardFailed(final ShardRouting shardRouting, final String indexUUID, final String reason, final DiscoveryNode masterNode) {
-        logger.trace("{} re-sending failed shard for {}, indexUUID [{}], reason [{}]", shardRouting.shardId(), shardRouting, indexUUID, reason);
-        innerShardFailed(shardRouting, indexUUID, reason, masterNode);
+    public void resendShardFailed(final ShardRouting shardRouting, final String indexUUID, final DiscoveryNode masterNode, final String message, @Nullable final Throwable failure) {
+        logger.trace("{} re-sending failed shard for {}, indexUUID [{}], reason [{}]", failure, shardRouting.shardId(), shardRouting, indexUUID, message);
+        innerShardFailed(shardRouting, indexUUID, masterNode, message, failure);
     }
 
-    private void innerShardFailed(final ShardRouting shardRouting, final String indexUUID, final String reason, final DiscoveryNode masterNode) {
-        ShardRoutingEntry shardRoutingEntry = new ShardRoutingEntry(shardRouting, indexUUID, reason);
+    private void innerShardFailed(final ShardRouting shardRouting, final String indexUUID, final DiscoveryNode masterNode, final String message, final Throwable failure) {
+        ShardRoutingEntry shardRoutingEntry = new ShardRoutingEntry(shardRouting, indexUUID, message, failure);
         transportService.sendRequest(masterNode,
                 SHARD_FAILED_ACTION_NAME, shardRoutingEntry, new EmptyTransportResponseHandler(ThreadPool.Names.SAME) {
                     @Override
@@ -105,20 +107,17 @@ public class ShardStateAction extends AbstractComponent {
     public void shardStarted(final ShardRouting shardRouting, String indexUUID, final String reason) {
         DiscoveryNode masterNode = clusterService.state().nodes().masterNode();
         if (masterNode == null) {
-            logger.warn("can't send shard started for {}. no master known.", shardRouting);
+            logger.warn("{} can't send shard started for {}, no master known.", shardRouting.shardId(), shardRouting);
             return;
         }
         shardStarted(shardRouting, indexUUID, reason, masterNode);
     }
 
     public void shardStarted(final ShardRouting shardRouting, String indexUUID, final String reason, final DiscoveryNode masterNode) {
-
-        ShardRoutingEntry shardRoutingEntry = new ShardRoutingEntry(shardRouting, indexUUID, reason);
-
-        logger.debug("sending shard started for {}", shardRoutingEntry);
-
+        ShardRoutingEntry shardRoutingEntry = new ShardRoutingEntry(shardRouting, indexUUID, reason, null);
+        logger.debug("{} sending shard started for {}", shardRoutingEntry.shardRouting.shardId(), shardRoutingEntry);
         transportService.sendRequest(masterNode,
-                SHARD_STARTED_ACTION_NAME, new ShardRoutingEntry(shardRouting, indexUUID, reason), new EmptyTransportResponseHandler(ThreadPool.Names.SAME) {
+                SHARD_STARTED_ACTION_NAME, new ShardRoutingEntry(shardRouting, indexUUID, reason, null), new EmptyTransportResponseHandler(ThreadPool.Names.SAME) {
                     @Override
                     public void handleException(TransportException exp) {
                         logger.warn("failed to send shard started to [{}]", exp, masterNode);
@@ -128,9 +127,9 @@ public class ShardStateAction extends AbstractComponent {
     }
 
     private void handleShardFailureOnMaster(final ShardRoutingEntry shardRoutingEntry) {
-        logger.warn("{} received shard failed for {}", shardRoutingEntry.shardRouting.shardId(), shardRoutingEntry);
+        logger.warn("{} received shard failed for {}", shardRoutingEntry.failure, shardRoutingEntry.shardRouting.shardId(), shardRoutingEntry);
         failedShardQueue.add(shardRoutingEntry);
-        clusterService.submitStateUpdateTask("shard-failed (" + shardRoutingEntry.shardRouting + "), reason [" + shardRoutingEntry.reason + "]", Priority.HIGH, new ProcessedClusterStateUpdateTask() {
+        clusterService.submitStateUpdateTask("shard-failed (" + shardRoutingEntry.shardRouting + "), message [" + shardRoutingEntry.message + "]", Priority.HIGH, new ProcessedClusterStateUpdateTask() {
 
             @Override
             public ClusterState execute(ClusterState currentState) {
@@ -151,7 +150,7 @@ public class ShardStateAction extends AbstractComponent {
 
                 List<FailedRerouteAllocation.FailedShard> shardRoutingsToBeApplied = new ArrayList<>(shardRoutingEntries.size());
                 for (ShardRoutingEntry entry : extractShardsToBeApplied(shardRoutingEntries, "failed", metaData, logger)) {
-                    shardRoutingsToBeApplied.add(new FailedRerouteAllocation.FailedShard(entry.shardRouting, entry.reason));
+                    shardRoutingsToBeApplied.add(new FailedRerouteAllocation.FailedShard(entry.shardRouting, entry.message, entry.failure));
                 }
 
                 // mark all entries as processed
@@ -214,7 +213,7 @@ public class ShardStateAction extends AbstractComponent {
         // process started events as fast as possible, to make shards available
         startedShardsQueue.add(shardRoutingEntry);
 
-        clusterService.submitStateUpdateTask("shard-started (" + shardRoutingEntry.shardRouting + "), reason [" + shardRoutingEntry.reason + "]", Priority.URGENT,
+        clusterService.submitStateUpdateTask("shard-started (" + shardRoutingEntry.shardRouting + "), reason [" + shardRoutingEntry.message + "]", Priority.URGENT,
                 new ClusterStateUpdateTask() {
                     @Override
                     public ClusterState execute(ClusterState currentState) {
@@ -284,41 +283,43 @@ public class ShardStateAction extends AbstractComponent {
     static class ShardRoutingEntry extends TransportRequest {
 
         ShardRouting shardRouting;
-
         String indexUUID = IndexMetaData.INDEX_UUID_NA_VALUE;
-
-        String reason;
+        String message;
+        Throwable failure;
 
         volatile boolean processed; // state field, no need to serialize
 
         ShardRoutingEntry() {
         }
 
-        ShardRoutingEntry(ShardRouting shardRouting, String indexUUID, String reason) {
+        ShardRoutingEntry(ShardRouting shardRouting, String indexUUID, String message, @Nullable Throwable failure) {
             this.shardRouting = shardRouting;
-            this.reason = reason;
             this.indexUUID = indexUUID;
+            this.message = message;
+            this.failure = failure;
         }
 
         @Override
         public void readFrom(StreamInput in) throws IOException {
             super.readFrom(in);
             shardRouting = readShardRoutingEntry(in);
-            reason = in.readString();
             indexUUID = in.readString();
+            message = in.readString();
+            failure = in.readThrowable();
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             shardRouting.writeTo(out);
-            out.writeString(reason);
             out.writeString(indexUUID);
+            out.writeString(message);
+            out.writeThrowable(failure);
         }
 
         @Override
         public String toString() {
-            return "" + shardRouting + ", indexUUID [" + indexUUID + "], reason [" + reason + "]";
+            return "" + shardRouting + ", indexUUID [" + indexUUID + "], message [" + message + "], failure [" + ExceptionsHelper.detailedMessage(failure) + "]";
         }
     }
 }

core/src/main/java/org/elasticsearch/cluster/routing/UnassignedInfo.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.cluster.routing;
 
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.Nullable;
@@ -95,28 +96,37 @@ public class UnassignedInfo implements ToXContent, Writeable<UnassignedInfo> {
 
     private final Reason reason;
     private final long timestamp;
-    private final String details;
+    private final String message;
+    private final Throwable failure;
 
-    public UnassignedInfo(Reason reason, String details) {
-        this(reason, System.currentTimeMillis(), details);
+    public UnassignedInfo(Reason reason, String message) {
+        this(reason, System.currentTimeMillis(), message, null);
     }
 
-    private UnassignedInfo(Reason reason, long timestamp, String details) {
+    public UnassignedInfo(Reason reason, @Nullable String message, @Nullable Throwable failure) {
+        this(reason, System.currentTimeMillis(), message, failure);
+    }
+
+    private UnassignedInfo(Reason reason, long timestamp, String message, Throwable failure) {
         this.reason = reason;
         this.timestamp = timestamp;
-        this.details = details;
+        this.message = message;
+        this.failure = failure;
+        assert !(message == null && failure != null) : "provide a message if a failure exception is provided";
     }
 
     UnassignedInfo(StreamInput in) throws IOException {
         this.reason = Reason.values()[(int) in.readByte()];
         this.timestamp = in.readLong();
-        this.details = in.readOptionalString();
+        this.message = in.readOptionalString();
+        this.failure = in.readThrowable();
     }
 
     public void writeTo(StreamOutput out) throws IOException {
         out.writeByte((byte) reason.ordinal());
         out.writeLong(timestamp);
-        out.writeOptionalString(details);
+        out.writeOptionalString(message);
+        out.writeThrowable(failure);
     }
 
     public UnassignedInfo readFrom(StreamInput in) throws IOException {
@@ -144,8 +154,27 @@ public class UnassignedInfo implements ToXContent, Writeable<UnassignedInfo> {
      * Returns optional details explaining the reasons.
      */
     @Nullable
+    public String getMessage() {
+        return this.message;
+    }
+
+    /**
+     * Returns additional failure exception details if exists.
+     */
+    @Nullable
+    public Throwable getFailure() {
+        return failure;
+    }
+
+    /**
+     * Builds a string representation of the message and the failure if exists.
+     */
+    @Nullable
     public String getDetails() {
-        return this.details;
+        if (message == null) {
+            return null;
+        }
+        return message + (failure == null ? "" : ", failure " + ExceptionsHelper.detailedMessage(failure));
     }
 
     /**
@@ -233,6 +262,7 @@ public class UnassignedInfo implements ToXContent, Writeable<UnassignedInfo> {
         StringBuilder sb = new StringBuilder();
         sb.append("unassigned_info[[reason=").append(reason).append("]");
         sb.append(", at[").append(DATE_TIME_FORMATTER.printer().print(timestamp)).append("]");
+        String details = getDetails();
         if (details != null) {
             sb.append(", details[").append(details).append("]");
         }
@@ -245,6 +275,7 @@ public class UnassignedInfo implements ToXContent, Writeable<UnassignedInfo> {
         builder.startObject("unassigned_info");
         builder.field("reason", reason);
         builder.field("at", DATE_TIME_FORMATTER.printer().print(timestamp));
+        String details = getDetails();
         if (details != null) {
             builder.field("details", details);
         }

core/src/main/java/org/elasticsearch/cluster/routing/allocation/AllocationService.java

@@ -22,6 +22,7 @@ package org.elasticsearch.cluster.routing.allocation;
 import com.carrotsearch.hppc.cursors.ObjectCursor;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.cluster.ClusterInfoService;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
@@ -86,7 +87,7 @@ public class AllocationService extends AbstractComponent {
     }
 
     public RoutingAllocation.Result applyFailedShard(ClusterState clusterState, ShardRouting failedShard) {
-        return applyFailedShards(clusterState, ImmutableList.of(new FailedRerouteAllocation.FailedShard(failedShard, null)));
+        return applyFailedShards(clusterState, ImmutableList.of(new FailedRerouteAllocation.FailedShard(failedShard, null, null)));
     }
 
     /**
@@ -101,7 +102,7 @@ public class AllocationService extends AbstractComponent {
         FailedRerouteAllocation allocation = new FailedRerouteAllocation(allocationDeciders, routingNodes, clusterState.nodes(), failedShards, clusterInfoService.getClusterInfo());
         boolean changed = false;
         for (FailedRerouteAllocation.FailedShard failedShard : failedShards) {
-            changed |= applyFailedShard(allocation, failedShard.shard, true, new UnassignedInfo(UnassignedInfo.Reason.ALLOCATION_FAILED, failedShard.details));
+            changed |= applyFailedShard(allocation, failedShard.shard, true, new UnassignedInfo(UnassignedInfo.Reason.ALLOCATION_FAILED, failedShard.message, failedShard.failure));
         }
         if (!changed) {
             return new RoutingAllocation.Result(false, clusterState.routingTable());

core/src/main/java/org/elasticsearch/cluster/routing/allocation/FailedRerouteAllocation.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.cluster.routing.allocation;
 
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.cluster.ClusterInfo;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.routing.RoutingNodes;
@@ -39,19 +40,18 @@ public class FailedRerouteAllocation extends RoutingAllocation {
      */
     public static class FailedShard {
         public final ShardRouting shard;
-        public final String details;
+        public final String message;
+        public final Throwable failure;
 
-        public FailedShard(ShardRouting shard, String details) {
+        public FailedShard(ShardRouting shard, String message, Throwable failure) {
             this.shard = shard;
-            this.details = details;
+            this.message = message;
+            this.failure = failure;
         }
 
         @Override
         public String toString() {
-            return "FailedShard{" +
-                    "shard=" + shard +
-                    ", details='" + details + '\'' +
-                    '}';
+            return "failed shard, shard " + shard + ", message [" + message + "], failure [" + ExceptionsHelper.detailedMessage(failure) + "]";
         }
     }
 

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

@@ -63,8 +63,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentMap;
 
-import static org.elasticsearch.ExceptionsHelper.detailedMessage;
-
 /**
  *
  */
@@ -492,10 +490,8 @@ public class IndicesClusterStateService extends AbstractLifecycleComponent<Indic
             if (!indexService.hasShard(shardId) && shardRouting.started()) {
                 if (failedShards.containsKey(shardRouting.shardId())) {
                     if (nodes.masterNode() != null) {
-                        shardStateAction.resendShardFailed(shardRouting, indexMetaData.getIndexUUID(),
-                                "master " + nodes.masterNode() + " marked shard as started, but shard has previous failed. resending shard failure.",
-                                nodes.masterNode()
-                        );
+                        shardStateAction.resendShardFailed(shardRouting, indexMetaData.getIndexUUID(), nodes.masterNode(),
+                                "master " + nodes.masterNode() + " marked shard as started, but shard has previous failed. resending shard failure.", null);
                     }
                 } else {
                     // the master thinks we are started, but we don't have this shard at all, mark it as failed
@@ -636,9 +632,8 @@ public class IndicesClusterStateService extends AbstractLifecycleComponent<Indic
         if (!indexService.hasShard(shardId)) {
             if (failedShards.containsKey(shardRouting.shardId())) {
                 if (nodes.masterNode() != null) {
-                    shardStateAction.resendShardFailed(shardRouting, indexMetaData.getIndexUUID(),
-                            "master " + nodes.masterNode() + " marked shard as initializing, but shard is marked as failed, resend shard failure",
-                            nodes.masterNode());
+                    shardStateAction.resendShardFailed(shardRouting, indexMetaData.getIndexUUID(), nodes.masterNode(),
+                            "master " + nodes.masterNode() + " marked shard as initializing, but shard is marked as failed, resend shard failure", null);
                 }
                 return;
             }
@@ -817,7 +812,7 @@ public class IndicesClusterStateService extends AbstractLifecycleComponent<Indic
         try {
             logger.warn("[{}] marking and sending shard failed due to [{}]", failure, shardRouting.shardId(), message);
             failedShards.put(shardRouting.shardId(), new FailedShard(shardRouting.version()));
-            shardStateAction.shardFailed(shardRouting, indexUUID, "shard failure [" + message + "]" + (failure == null ? "" : "[" + detailedMessage(failure) + "]"));
+            shardStateAction.shardFailed(shardRouting, indexUUID, message, failure);
         } catch (Throwable e1) {
             logger.warn("[{}][{}] failed to mark shard as failed (because of [{}])", e1, shardRouting.getIndex(), shardRouting.getId(), message);
         }

core/src/test/java/org/elasticsearch/cluster/action/shard/ShardStateActionTest.java

@@ -66,16 +66,16 @@ public class ShardStateActionTest extends ElasticsearchTestCase {
         ArrayList<ShardStateAction.ShardRoutingEntry> listToFilter = new ArrayList<>();
         ArrayList<ShardStateAction.ShardRoutingEntry> expectedToBeApplied = new ArrayList<>();
 
-        listToFilter.add(new ShardStateAction.ShardRoutingEntry(initShard, indexMetaData.indexUUID() + "_suffix", "wrong_uuid"));
+        listToFilter.add(new ShardStateAction.ShardRoutingEntry(initShard, indexMetaData.indexUUID() + "_suffix", "wrong_uuid", null));
 
-        listToFilter.add(new ShardStateAction.ShardRoutingEntry(relocatingShard.buildTargetRelocatingShard(), indexMetaData.indexUUID(), "relocating_to_node"));
+        listToFilter.add(new ShardStateAction.ShardRoutingEntry(relocatingShard.buildTargetRelocatingShard(), indexMetaData.indexUUID(), "relocating_to_node", null));
         expectedToBeApplied.add(listToFilter.get(listToFilter.size() - 1));
 
-        listToFilter.add(new ShardStateAction.ShardRoutingEntry(startedShard, indexMetaData.indexUUID(), "started shard"));
+        listToFilter.add(new ShardStateAction.ShardRoutingEntry(startedShard, indexMetaData.indexUUID(), "started shard", null));
         expectedToBeApplied.add(listToFilter.get(listToFilter.size() - 1));
 
         listToFilter.add(new ShardStateAction.ShardRoutingEntry(TestShardRouting.newShardRouting(initShard.index() + "_NA", initShard.id(),
-                initShard.currentNodeId(), initShard.primary(), initShard.state(), initShard.version()), indexMetaData.indexUUID(), "wrong_uuid"));
+                initShard.currentNodeId(), initShard.primary(), initShard.state(), initShard.version()), indexMetaData.indexUUID(), "wrong_uuid", null));
 
         List<ShardStateAction.ShardRoutingEntry> toBeApplied = ShardStateAction.extractShardsToBeApplied(listToFilter, "for testing", state.metaData(), logger);
         if (toBeApplied.size() != expectedToBeApplied.size()) {

core/src/test/java/org/elasticsearch/cluster/routing/UnassignedInfoTests.java

@@ -76,6 +76,7 @@ public class UnassignedInfoTests extends ElasticsearchAllocationTestCase {
         UnassignedInfo read = new UnassignedInfo(StreamInput.wrap(out.bytes()));
         assertThat(read.getReason(), equalTo(meta.getReason()));
         assertThat(read.getTimestampInMillis(), equalTo(meta.getTimestampInMillis()));
+        assertThat(read.getMessage(), equalTo(meta.getMessage()));
         assertThat(read.getDetails(), equalTo(meta.getDetails()));
     }
 
@@ -248,12 +249,13 @@ public class UnassignedInfoTests extends ElasticsearchAllocationTestCase {
         assertThat(clusterState.routingNodes().hasUnassigned(), equalTo(false));
         // fail shard
         ShardRouting shardToFail = clusterState.routingNodes().shardsWithState(STARTED).get(0);
-        clusterState = ClusterState.builder(clusterState).routingResult(allocation.applyFailedShards(clusterState, ImmutableList.of(new FailedRerouteAllocation.FailedShard(shardToFail, "test fail")))).build();
+        clusterState = ClusterState.builder(clusterState).routingResult(allocation.applyFailedShards(clusterState, ImmutableList.of(new FailedRerouteAllocation.FailedShard(shardToFail, "test fail", null)))).build();
         // verify the reason and details
         assertThat(clusterState.routingNodes().hasUnassigned(), equalTo(true));
         assertThat(clusterState.routingNodes().shardsWithState(UNASSIGNED).size(), equalTo(1));
         assertThat(clusterState.routingNodes().shardsWithState(UNASSIGNED).get(0).unassignedInfo(), notNullValue());
         assertThat(clusterState.routingNodes().shardsWithState(UNASSIGNED).get(0).unassignedInfo().getReason(), equalTo(UnassignedInfo.Reason.ALLOCATION_FAILED));
+        assertThat(clusterState.routingNodes().shardsWithState(UNASSIGNED).get(0).unassignedInfo().getMessage(), equalTo("test fail"));
         assertThat(clusterState.routingNodes().shardsWithState(UNASSIGNED).get(0).unassignedInfo().getDetails(), equalTo("test fail"));
         assertThat(clusterState.routingNodes().shardsWithState(UNASSIGNED).get(0).unassignedInfo().getTimestampInMillis(), greaterThan(0l));
     }

core/src/test/java/org/elasticsearch/cluster/routing/allocation/FailedShardsRoutingTests.java

@@ -318,7 +318,7 @@ public class FailedShardsRoutingTests extends ElasticsearchAllocationTestCase {
             String n = "node" + Integer.toString(randomInt(numberOfReplicas));
             logger.info("failing shard on node [{}]", n);
             ShardRouting shardToFail = routingNodes.node(n).get(0);
-            failedShards.add(new FailedRerouteAllocation.FailedShard(new ShardRouting(shardToFail), null));
+            failedShards.add(new FailedRerouteAllocation.FailedShard(new ShardRouting(shardToFail), null, null));
         }
 
         routingTable = strategy.applyFailedShards(clusterState, failedShards).routingTable();