Cluster metadata files destroyed when using blob store gateway causing data loss, closes elastic#1564.

Author: kimchy

modules/elasticsearch/src/main/java/org/elasticsearch/gateway/shared/SharedStorageGateway.java

@@ -1,8 +1,8 @@
 /*
- * Licensed to Elastic Search and Shay Banon under one
+ * Licensed to ElasticSearch and Shay Banon under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
- * regarding copyright ownership. Elastic Search licenses this
+ * regarding copyright ownership. ElasticSearch licenses this
  * file to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
@@ -32,35 +32,55 @@
 import org.elasticsearch.gateway.GatewayException;
 import org.elasticsearch.threadpool.ThreadPool;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static java.util.concurrent.Executors.*;
+import static org.elasticsearch.common.util.concurrent.EsExecutors.*;
+
 /**
- * @author kimchy (shay.banon)
+ *
  */
 public abstract class SharedStorageGateway extends AbstractLifecycleComponent<Gateway> implements Gateway, ClusterStateListener {
 
     private final ClusterService clusterService;
 
     private final ThreadPool threadPool;
 
+    private ExecutorService writeStateExecutor;
+
     public SharedStorageGateway(Settings settings, ThreadPool threadPool, ClusterService clusterService) {
         super(settings);
         this.threadPool = threadPool;
         this.clusterService = clusterService;
     }
 
-    @Override protected void doStart() throws ElasticSearchException {
+    @Override
+    protected void doStart() throws ElasticSearchException {
         clusterService.add(this);
+        this.writeStateExecutor = newSingleThreadExecutor(daemonThreadFactory(settings, "gateway#writeMetaData"));
     }
 
-    @Override protected void doStop() throws ElasticSearchException {
+    @Override
+    protected void doStop() throws ElasticSearchException {
         clusterService.remove(this);
+        writeStateExecutor.shutdown();
+        try {
+            writeStateExecutor.awaitTermination(10, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            // ignore
+        }
     }
 
-    @Override protected void doClose() throws ElasticSearchException {
+    @Override
+    protected void doClose() throws ElasticSearchException {
     }
 
-    @Override public void performStateRecovery(final GatewayStateRecoveredListener listener) throws GatewayException {
+    @Override
+    public void performStateRecovery(final GatewayStateRecoveredListener listener) throws GatewayException {
         threadPool.cached().execute(new Runnable() {
-            @Override public void run() {
+            @Override
+            public void run() {
                 logger.debug("reading state from gateway {} ...", this);
                 StopWatch stopWatch = new StopWatch().start();
                 MetaData metaData;
@@ -81,7 +101,8 @@ public SharedStorageGateway(Settings settings, ThreadPool threadPool, ClusterSer
         });
     }
 
-    @Override public void clusterChanged(final ClusterChangedEvent event) {
+    @Override
+    public void clusterChanged(final ClusterChangedEvent event) {
         if (!lifecycle.started()) {
             return;
         }
@@ -95,8 +116,9 @@ public SharedStorageGateway(Settings settings, ThreadPool threadPool, ClusterSer
             if (!event.metaDataChanged()) {
                 return;
             }
-            threadPool.cached().execute(new Runnable() {
-                @Override public void run() {
+            writeStateExecutor.execute(new Runnable() {
+                @Override
+                public void run() {
                     logger.debug("writing to gateway {} ...", this);
                     StopWatch stopWatch = new StopWatch().start();
                     try {