add a merge thread pool that only does async merges (optimize), so it can be controlled by itself

Author: kimchy

modules/elasticsearch/src/main/java/org/elasticsearch/index/shard/service/InternalIndexShard.java

@@ -91,11 +91,11 @@ public class InternalIndexShard extends AbstractIndexShardComponent implements I
     private volatile IndexShardState state;
 
     private final TimeValue refreshInterval;
-    private final TimeValue optimizeInterval;
+    private final TimeValue mergeInterval;
 
     private volatile ScheduledFuture refreshScheduledFuture;
 
-    private volatile ScheduledFuture optimizeScheduleFuture;
+    private volatile ScheduledFuture mergeScheduleFuture;
 
     private volatile ShardRouting shardRouting;
 
@@ -121,7 +121,7 @@ public class InternalIndexShard extends AbstractIndexShardComponent implements I
         } else {
             refreshInterval = new TimeValue(-2);
         }
-        optimizeInterval = indexSettings.getAsTime("index.merge.async_interval", TimeValue.timeValueSeconds(1));
+        mergeInterval = indexSettings.getAsTime("index.merge.async_interval", TimeValue.timeValueSeconds(1));
 
         logger.debug("state: [CREATED]");
 
@@ -436,9 +436,9 @@ public void close(String reason) {
                     refreshScheduledFuture.cancel(true);
                     refreshScheduledFuture = null;
                 }
-                if (optimizeScheduleFuture != null) {
-                    optimizeScheduleFuture.cancel(true);
-                    optimizeScheduleFuture = null;
+                if (mergeScheduleFuture != null) {
+                    mergeScheduleFuture.cancel(true);
+                    mergeScheduleFuture = null;
                 }
             }
             logger.debug("state: [{}]->[{}], reason [{}]", state, IndexShardState.CLOSED, reason);
@@ -553,9 +553,9 @@ private void startScheduledTasksIfNeeded() {
         // since we can do async merging, it will not be called explicitly when indexing (adding / deleting docs), and only when flushing
         // so, make sure we periodically call it, this need to be a small enough value so mergine will actually
         // happen and reduce the number of segments
-        if (optimizeInterval.millis() > 0) {
-            optimizeScheduleFuture = threadPool.schedule(optimizeInterval, ThreadPool.Names.MANAGEMENT, new EngineOptimizer());
-            logger.debug("scheduling optimizer / merger every {}", optimizeInterval);
+        if (mergeInterval.millis() > 0) {
+            mergeScheduleFuture = threadPool.schedule(mergeInterval, ThreadPool.Names.MERGE, new EngineMerger());
+            logger.debug("scheduling optimizer / merger every {}", mergeInterval);
         } else {
             logger.debug("scheduled optimizer / merger disabled");
         }
@@ -606,7 +606,7 @@ private class EngineRefresher implements Runnable {
         }
     }
 
-    private class EngineOptimizer implements Runnable {
+    private class EngineMerger implements Runnable {
         @Override public void run() {
             try {
                 // -1 means maybe merge
@@ -627,7 +627,7 @@ private class EngineOptimizer implements Runnable {
                 logger.warn("Failed to perform scheduled engine optimize/merge", e);
             }
             if (state != IndexShardState.CLOSED) {
-                optimizeScheduleFuture = threadPool.schedule(optimizeInterval, ThreadPool.Names.MANAGEMENT, this);
+                mergeScheduleFuture = threadPool.schedule(mergeInterval, ThreadPool.Names.MERGE, this);
             }
         }
     }

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

@@ -52,6 +52,7 @@ public static class Names {
         public static final String SEARCH = "search";
         public static final String PERCOLATE = "percolate";
         public static final String MANAGEMENT = "management";
+        public static final String MERGE = "merge";
         public static final String SNAPSHOT = "snapshot";
     }
 
@@ -74,6 +75,7 @@ public ThreadPool() {
         executors.put(Names.SEARCH, build(Names.SEARCH, "cached", groupSettings.get(Names.SEARCH), ImmutableSettings.Builder.EMPTY_SETTINGS));
         executors.put(Names.PERCOLATE, build(Names.PERCOLATE, "cached", groupSettings.get(Names.PERCOLATE), ImmutableSettings.Builder.EMPTY_SETTINGS));
         executors.put(Names.MANAGEMENT, build(Names.MANAGEMENT, "scaling", groupSettings.get(Names.MANAGEMENT), settingsBuilder().put("keep_alive", "30s").put("size", 20).build()));
+        executors.put(Names.MERGE, build(Names.MERGE, "scaling", groupSettings.get(Names.MERGE), settingsBuilder().put("keep_alive", "30s").put("size", 20).build()));
         executors.put(Names.SNAPSHOT, build(Names.SNAPSHOT, "scaling", groupSettings.get(Names.SNAPSHOT), ImmutableSettings.Builder.EMPTY_SETTINGS));
         executors.put(Names.SAME, MoreExecutors.sameThreadExecutor());
         this.executors = ImmutableMap.copyOf(executors);