start ingest thread pool only when node.ingest is set to true

core/src/main/java/org/elasticsearch/action/ActionModule.java

@@ -203,6 +203,7 @@ import org.elasticsearch.common.inject.AbstractModule;
 import org.elasticsearch.common.inject.multibindings.MapBinder;
 import org.elasticsearch.common.inject.multibindings.Multibinder;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.ingest.IngestModule;
 
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -227,15 +228,13 @@ public class ActionModule extends AbstractModule {
             this.transportAction = transportAction;
             this.supportTransportActions = supportTransportActions;
         }
-
-
     }
 
     private final boolean ingestEnabled;
     private final boolean proxy;
 
     public ActionModule(Settings settings, boolean proxy) {
-        this.ingestEnabled = settings.getAsBoolean("node.ingest", false);
+        this.ingestEnabled = IngestModule.isIngestEnabled(settings);
         this.proxy = proxy;
     }
 

core/src/main/java/org/elasticsearch/ingest/IngestModule.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.ingest;
 
 import org.elasticsearch.common.inject.AbstractModule;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.ingest.core.Processor;
 import org.elasticsearch.ingest.core.TemplateService;
@@ -52,4 +53,8 @@ public class IngestModule extends AbstractModule {
     public void registerProcessor(String type, BiFunction<Environment, TemplateService, Processor.Factory<?>> processorFactoryProvider) {
         processorsRegistry.registerProcessor(type, processorFactoryProvider);
     }
+
+    public static boolean isIngestEnabled(Settings settings) {
+        return settings.getAsBoolean("node.ingest", false);
+    }
 }

core/src/main/java/org/elasticsearch/threadpool/ThreadPool.java

@@ -38,6 +38,7 @@ import org.elasticsearch.common.util.concurrent.XRejectedExecutionHandler;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentBuilderString;
+import org.elasticsearch.ingest.IngestModule;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -236,7 +237,9 @@ public class ThreadPool extends AbstractComponent {
         add(defaultExecutorTypeSettings, new ExecutorSettingsBuilder(Names.FORCE_MERGE).size(1));
         add(defaultExecutorTypeSettings, new ExecutorSettingsBuilder(Names.FETCH_SHARD_STARTED).size(availableProcessors * 2).keepAlive("5m"));
         add(defaultExecutorTypeSettings, new ExecutorSettingsBuilder(Names.FETCH_SHARD_STORE).size(availableProcessors * 2).keepAlive("5m"));
+        if (IngestModule.isIngestEnabled(settings)) {
             add(defaultExecutorTypeSettings, new ExecutorSettingsBuilder(Names.INGEST).size(availableProcessors).queueSize(200));
+        }
 
         this.defaultExecutorTypeSettings = unmodifiableMap(defaultExecutorTypeSettings);
 

core/src/test/java/org/elasticsearch/threadpool/ThreadPoolTests.java

@@ -0,0 +1,64 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information 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
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.threadpool;
+
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.test.ESTestCase;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.not;
+
+public class ThreadPoolTests extends ESTestCase {
+
+    public void testIngestThreadPoolNotStartedWithIngestDisabled() throws Exception {
+        Settings settings = Settings.builder().put("name", "test").put("node.ingest", false).build();
+        ThreadPool threadPool = null;
+        try {
+            threadPool = new ThreadPool(settings);
+            for (ThreadPool.Info info : threadPool.info()) {
+                assertThat(info.getName(), not(equalTo("ingest")));
+            }
+        } finally {
+            if (threadPool != null) {
+                terminate(threadPool);
+            }
+        }
+    }
+
+    public void testIngestThreadPoolStartedWithIngestEnabled() throws Exception {
+        Settings settings = Settings.builder().put("name", "test").put("node.ingest", true).build();
+        ThreadPool threadPool = null;
+        try {
+            threadPool = new ThreadPool(settings);
+            boolean ingestFound = false;
+            for (ThreadPool.Info info : threadPool.info()) {
+                if (info.getName().equals("ingest")) {
+                    ingestFound = true;
+                    break;
+                }
+            }
+            assertThat(ingestFound, equalTo(true));
+        } finally {
+            if (threadPool != null) {
+                terminate(threadPool);
+            }
+        }
+    }
+}

plugins/ingest/src/main/java/org/elasticsearch/plugin/ingest/IngestPlugin.java

@@ -19,7 +19,6 @@
 
 package org.elasticsearch.plugin.ingest;
 
-import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.ingest.IngestModule;
 import org.elasticsearch.ingest.processor.AppendProcessor;
 import org.elasticsearch.ingest.processor.ConvertProcessor;
@@ -42,12 +41,6 @@ public class IngestPlugin extends Plugin {
 
     public static final String NAME = "ingest";
 
-    private final boolean ingestEnabled;
-
-    public IngestPlugin(Settings nodeSettings) {
-        this.ingestEnabled = nodeSettings.getAsBoolean("node.ingest", false);
-    }
-
     @Override
     public String name() {
         return NAME;