From 3d9e87201e52a2595c14a8dec3313c9595151a50 Mon Sep 17 00:00:00 2001
From: Shay Banon <kimchy@gmail.com>
Date: Thu, 22 Dec 2011 20:42:37 +0200
Subject: [PATCH] Cluster metadata files destroyed when using blob store
 gateway causing data loss, closes #1564,

---
 .../gateway/shared/SharedStorageGateway.java    | 17 ++++++++++++++++-
 1 file changed, 16 insertions(+), 1 deletion(-)

diff --git a/src/main/java/org/elasticsearch/gateway/shared/SharedStorageGateway.java b/src/main/java/org/elasticsearch/gateway/shared/SharedStorageGateway.java
index 8d3ce981cde..e2f121d946e 100644
--- a/src/main/java/org/elasticsearch/gateway/shared/SharedStorageGateway.java
+++ b/src/main/java/org/elasticsearch/gateway/shared/SharedStorageGateway.java
@@ -32,6 +32,12 @@ import org.elasticsearch.gateway.Gateway;
 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.newSingleThreadExecutor;
+import static org.elasticsearch.common.util.concurrent.EsExecutors.daemonThreadFactory;
+
 /**
  *
  */
@@ -41,6 +47,8 @@ public abstract class SharedStorageGateway extends AbstractLifecycleComponent<Ga
 
     private final ThreadPool threadPool;
 
+    private ExecutorService writeStateExecutor;
+
     public SharedStorageGateway(Settings settings, ThreadPool threadPool, ClusterService clusterService) {
         super(settings);
         this.threadPool = threadPool;
@@ -50,11 +58,18 @@ public abstract class SharedStorageGateway extends AbstractLifecycleComponent<Ga
     @Override
     protected void doStart() throws ElasticSearchException {
         clusterService.add(this);
+        this.writeStateExecutor = newSingleThreadExecutor(daemonThreadFactory(settings, "gateway#writeMetaData"));
     }
 
     @Override
     protected void doStop() throws ElasticSearchException {
         clusterService.remove(this);
+        writeStateExecutor.shutdown();
+        try {
+            writeStateExecutor.awaitTermination(10, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            // ignore
+        }
     }
 
     @Override
@@ -101,7 +116,7 @@ public abstract class SharedStorageGateway extends AbstractLifecycleComponent<Ga
             if (!event.metaDataChanged()) {
                 return;
             }
-            threadPool.cached().execute(new Runnable() {
+            writeStateExecutor.execute(new Runnable() {
                 @Override
                 public void run() {
                     logger.debug("writing to gateway {} ...", this);