From 38d895bf1aeb6441b2981ac45355cdcdece5d34b Mon Sep 17 00:00:00 2001
From: Luca Cavanna <cavannaluca@gmail.com>
Date: Mon, 18 Nov 2013 10:39:55 +0100
Subject: [PATCH] Added catch EsRejectedExecutionException when trying to
 reschedule river start

---
 .../org/elasticsearch/river/RiversService.java  | 17 +++++++++++------
 1 file changed, 11 insertions(+), 6 deletions(-)

diff --git a/src/main/java/org/elasticsearch/river/RiversService.java b/src/main/java/org/elasticsearch/river/RiversService.java
index b32a1bff871..45f53bd9566 100644
--- a/src/main/java/org/elasticsearch/river/RiversService.java
+++ b/src/main/java/org/elasticsearch/river/RiversService.java
@@ -39,6 +39,7 @@ import org.elasticsearch.common.inject.Injectors;
 import org.elasticsearch.common.inject.ModulesBuilder;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.indices.IndexMissingException;
@@ -282,12 +283,16 @@ public class RiversService extends AbstractLifecycleComponent<RiversService> {
                         if (isShardNotAvailableException(failure)) {
                             logger.debug("failed to get _meta from [{}]/[{}], retrying...", e, routing.riverName().type(), routing.riverName().name());
                             final ActionListener<GetResponse> listener = this;
-                            threadPool.schedule(TimeValue.timeValueSeconds(5), ThreadPool.Names.SAME, new Runnable() {
-                                @Override
-                                public void run() {
-                                    client.prepareGet(riverIndexName, routing.riverName().name(), "_meta").setListenerThreaded(true).execute(listener);
-                                }
-                            });
+                            try {
+                                threadPool.schedule(TimeValue.timeValueSeconds(5), ThreadPool.Names.SAME, new Runnable() {
+                                    @Override
+                                    public void run() {
+                                        client.prepareGet(riverIndexName, routing.riverName().name(), "_meta").setListenerThreaded(true).execute(listener);
+                                    }
+                                });
+                            } catch (EsRejectedExecutionException ex) {
+                                logger.debug("Couldn't schedule river start retry, node might be shutting down", ex);
+                            }
                         } else {
                             logger.warn("failed to get _meta from [{}]/[{}]", e, routing.riverName().type(), routing.riverName().name());
                         }