Abort snapshots on a node that leaves the cluster

core/src/main/java/org/elasticsearch/cluster/SnapshotsInProgress.java

@@ -210,12 +210,9 @@ public static boolean completed(ObjectContainer<ShardSnapshotStatus> shards) {
 
 
     public static class ShardSnapshotStatus {
-        private State state;
-        private String nodeId;
-        private String reason;
-
-        private ShardSnapshotStatus() {
-        }
+        private final State state;
+        private final String nodeId;
+        private final String reason;
 
         public ShardSnapshotStatus(String nodeId) {
             this(nodeId, State.INIT);
@@ -231,6 +228,12 @@ public ShardSnapshotStatus(String nodeId, State state, String reason) {
             this.reason = reason;
         }
 
+        public ShardSnapshotStatus(StreamInput in) throws IOException {
+            nodeId = in.readOptionalString();
+            state = State.fromValue(in.readByte());
+            reason = in.readOptionalString();
+        }
+
         public State state() {
             return state;
         }
@@ -243,18 +246,6 @@ public String reason() {
             return reason;
         }
 
-        public static ShardSnapshotStatus readShardSnapshotStatus(StreamInput in) throws IOException {
-            ShardSnapshotStatus shardSnapshotStatus = new ShardSnapshotStatus();
-            shardSnapshotStatus.readFrom(in);
-            return shardSnapshotStatus;
-        }
-
-        public void readFrom(StreamInput in) throws IOException {
-            nodeId = in.readOptionalString();
-            state = State.fromValue(in.readByte());
-            reason = in.readOptionalString();
-        }
-
         public void writeTo(StreamOutput out) throws IOException {
             out.writeOptionalString(nodeId);
             out.writeByte(state.value);
@@ -282,6 +273,11 @@ public int hashCode() {
             result = 31 * result + (reason != null ? reason.hashCode() : 0);
             return result;
         }
+
+        @Override
+        public String toString() {
+            return "ShardSnapshotStatus[state=" + state + ", nodeId=" + nodeId + ", reason=" + reason + "]";
+        }
     }
 
     public enum State {

core/src/main/java/org/elasticsearch/cluster/node/DiscoveryNodes.java

@@ -26,6 +26,7 @@
 import org.elasticsearch.cluster.AbstractDiffable;
 import org.elasticsearch.common.Booleans;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -399,9 +400,7 @@ public Delta delta(DiscoveryNodes other) {
     public String toString() {
         StringBuilder sb = new StringBuilder();
         sb.append("{");
-        for (DiscoveryNode node : this) {
-            sb.append(node).append(',');
-        }
+        sb.append(Strings.collectionToDelimitedString(this, ","));
         sb.append("}");
         return sb.toString();
     }

core/src/main/java/org/elasticsearch/index/IndexService.java

@@ -411,7 +411,11 @@ private void closeShard(String reason, ShardId sId, IndexShard indexShard, Store
             }
         } finally {
             try {
-                store.close();
+                if (store != null) {
+                    store.close();
+                } else {
+                    logger.trace("[{}] store not initialized prior to closing shard, nothing to close", shardId);
+                }
             } catch (Exception e) {
                 logger.warn(
                     (Supplier<?>) () -> new ParameterizedMessage(

core/src/main/java/org/elasticsearch/index/snapshots/IndexShardSnapshotStatus.java

@@ -27,7 +27,7 @@ public class IndexShardSnapshotStatus {
     /**
      * Snapshot stage
      */
-    public static enum Stage {
+    public enum Stage {
         /**
          * Snapshot hasn't started yet
          */
@@ -66,7 +66,7 @@ public static enum Stage {
 
     private long indexVersion;
 
-    private boolean aborted;
+    private volatile boolean aborted;
 
     private String failure;
 

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

@@ -69,6 +69,7 @@
 import org.elasticsearch.repositories.RepositoriesService;
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.snapshots.RestoreService;
+import org.elasticsearch.snapshots.SnapshotShardsService;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.io.IOException;
@@ -113,10 +114,11 @@ public IndicesClusterStateService(Settings settings, IndicesService indicesServi
                                       NodeMappingRefreshAction nodeMappingRefreshAction,
                                       RepositoriesService repositoriesService, RestoreService restoreService,
                                       SearchService searchService, SyncedFlushService syncedFlushService,
-                                      PeerRecoverySourceService peerRecoverySourceService) {
+                                      PeerRecoverySourceService peerRecoverySourceService, SnapshotShardsService snapshotShardsService) {
         this(settings, (AllocatedIndices<? extends Shard, ? extends AllocatedIndex<? extends Shard>>) indicesService,
             clusterService, threadPool, recoveryTargetService, shardStateAction,
-            nodeMappingRefreshAction, repositoriesService, restoreService, searchService, syncedFlushService, peerRecoverySourceService);
+            nodeMappingRefreshAction, repositoriesService, restoreService, searchService, syncedFlushService, peerRecoverySourceService,
+            snapshotShardsService);
     }
 
     // for tests
@@ -128,9 +130,10 @@ public IndicesClusterStateService(Settings settings, IndicesService indicesServi
                                NodeMappingRefreshAction nodeMappingRefreshAction,
                                RepositoriesService repositoriesService, RestoreService restoreService,
                                SearchService searchService, SyncedFlushService syncedFlushService,
-                               PeerRecoverySourceService peerRecoverySourceService) {
+                               PeerRecoverySourceService peerRecoverySourceService, SnapshotShardsService snapshotShardsService) {
         super(settings);
-        this.buildInIndexListener = Arrays.asList(peerRecoverySourceService, recoveryTargetService, searchService, syncedFlushService);
+        this.buildInIndexListener = Arrays.asList(peerRecoverySourceService, recoveryTargetService, searchService, syncedFlushService,
+                                                  snapshotShardsService);
         this.indicesService = indicesService;
         this.clusterService = clusterService;
         this.threadPool = threadPool;