use merge trigger to control when to do merges

now with merge trigger, we can simply decide when to do merges based on it

src/main/java/org/elasticsearch/index/engine/robin/RobinEngine.java

@@ -49,7 +49,6 @@ import org.elasticsearch.index.deletionpolicy.SnapshotDeletionPolicy;
 import org.elasticsearch.index.deletionpolicy.SnapshotIndexCommit;
 import org.elasticsearch.index.engine.*;
 import org.elasticsearch.index.indexing.ShardIndexingService;
-import org.elasticsearch.index.merge.policy.EnableMergePolicy;
 import org.elasticsearch.index.merge.policy.MergePolicyProvider;
 import org.elasticsearch.index.merge.scheduler.MergeSchedulerProvider;
 import org.elasticsearch.index.search.nested.IncludeNestedDocsQuery;
@@ -990,9 +989,6 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
             if (indexWriter == null) {
                 throw new EngineClosedException(shardId, failedEngine);
             }
-            if (indexWriter.getConfig().getMergePolicy() instanceof EnableMergePolicy) {
-                ((EnableMergePolicy) indexWriter.getConfig().getMergePolicy()).enableMerge();
-            }
             indexWriter.maybeMerge();
         } catch (OutOfMemoryError e) {
             failEngine(e);
@@ -1006,9 +1002,6 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
             throw new OptimizeFailedEngineException(shardId, e);
         } finally {
             rwl.readLock().unlock();
-            if (indexWriter != null && indexWriter.getConfig().getMergePolicy() instanceof EnableMergePolicy) {
-                ((EnableMergePolicy) indexWriter.getConfig().getMergePolicy()).disableMerge();
-            }
         }
     }
 
@@ -1023,9 +1016,6 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
                 if (indexWriter == null) {
                     throw new EngineClosedException(shardId, failedEngine);
                 }
-                if (indexWriter.getConfig().getMergePolicy() instanceof EnableMergePolicy) {
-                    ((EnableMergePolicy) indexWriter.getConfig().getMergePolicy()).enableMerge();
-                }
                 if (optimize.onlyExpungeDeletes()) {
                     indexWriter.forceMergeDeletes(false);
                 } else if (optimize.maxNumSegments() <= 0) {
@@ -1046,9 +1036,6 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
                 throw new OptimizeFailedEngineException(shardId, e);
             } finally {
                 rwl.readLock().unlock();
-                if (indexWriter != null && indexWriter.getConfig().getMergePolicy() instanceof EnableMergePolicy) {
-                    ((EnableMergePolicy) indexWriter.getConfig().getMergePolicy()).disableMerge();
-                }
                 optimizeMutex.set(false);
             }
         }

src/main/java/org/elasticsearch/index/merge/policy/EnableMergePolicy.java

@@ -1,38 +0,0 @@
-/*
- * Licensed to Elastic Search 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
- * 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.index.merge.policy;
-
-/**
- * Allows to control if merge should be enabled on the current thread or not. Defaults to
- * not being enabled.
- * <p/>
- * <p>This allows us to disable merging for things like adding docs or refresh (which might block
- * if no threads are there to handle the merge) and do it on flush (for example) or on explicit API call.
- *
- *
- */
-public interface EnableMergePolicy {
-
-    boolean isMergeEnabled();
-
-    void enableMerge();
-
-    void disableMerge();
-}

src/main/java/org/elasticsearch/index/merge/policy/LogByteSizeMergePolicyProvider.java

@@ -19,7 +19,8 @@
 
 package org.elasticsearch.index.merge.policy;
 
-import org.apache.lucene.index.*;
+import org.apache.lucene.index.LogByteSizeMergePolicy;
+import org.apache.lucene.index.SegmentInfos;
 import org.elasticsearch.ElasticSearchException;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.Preconditions;
@@ -32,7 +33,6 @@ import org.elasticsearch.index.shard.AbstractIndexShardComponent;
 import org.elasticsearch.index.store.Store;
 
 import java.io.IOException;
-import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
 
@@ -174,62 +174,19 @@ public class LogByteSizeMergePolicyProvider extends AbstractIndexShardComponent
         }
     }
 
-    public static class EnableMergeLogByteSizeMergePolicy extends CustomLogByteSizeMergePolicy implements EnableMergePolicy {
+    public static class EnableMergeLogByteSizeMergePolicy extends CustomLogByteSizeMergePolicy {
-
-        private final ThreadLocal<Boolean> enableMerge = new ThreadLocal<Boolean>() {
-            @Override
-            protected Boolean initialValue() {
-                return Boolean.FALSE;
-            }
-        };
 
         public EnableMergeLogByteSizeMergePolicy(LogByteSizeMergePolicyProvider provider) {
             super(provider);
         }
 
-        @Override
-        public void enableMerge() {
-            enableMerge.set(Boolean.TRUE);
-        }
-
-        @Override
-        public void disableMerge() {
-            enableMerge.set(Boolean.FALSE);
-        }
-
-        @Override
-        public boolean isMergeEnabled() {
-            return enableMerge.get() == Boolean.TRUE;
-        }
-
-        @Override
-        public void close() {
-            enableMerge.remove();
-            super.close();
-        }
-
         @Override
         public MergeSpecification findMerges(MergeTrigger trigger, SegmentInfos infos) throws IOException {
-            if (enableMerge.get() == Boolean.FALSE) {
+            // we don't enable merges while indexing documents, we do them in the background
+            if (trigger == MergeTrigger.SEGMENT_FLUSH) {
                 return null;
             }
             return super.findMerges(trigger, infos);
         }
-
-        @Override
-        public MergeSpecification findForcedMerges(SegmentInfos infos, int maxSegmentCount, Map<SegmentInfoPerCommit, Boolean> segmentsToMerge) throws IOException {
-            if (enableMerge.get() == Boolean.FALSE) {
-                return null;
-            }
-            return super.findForcedMerges(infos, maxSegmentCount, segmentsToMerge);
-        }
-
-        @Override
-        public MergeSpecification findForcedDeletesMerges(SegmentInfos infos) throws CorruptIndexException, IOException {
-            if (enableMerge.get() == Boolean.FALSE) {
-                return null;
-            }
-            return super.findForcedDeletesMerges(infos);
-        }
     }
 }

src/main/java/org/elasticsearch/index/merge/policy/LogDocMergePolicyProvider.java

@@ -19,7 +19,8 @@
 
 package org.elasticsearch.index.merge.policy;
 
-import org.apache.lucene.index.*;
+import org.apache.lucene.index.LogDocMergePolicy;
+import org.apache.lucene.index.SegmentInfos;
 import org.elasticsearch.ElasticSearchException;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.Preconditions;
@@ -30,7 +31,6 @@ import org.elasticsearch.index.shard.AbstractIndexShardComponent;
 import org.elasticsearch.index.store.Store;
 
 import java.io.IOException;
-import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
 
@@ -158,62 +158,19 @@ public class LogDocMergePolicyProvider extends AbstractIndexShardComponent imple
         }
     }
 
-    public static class EnableMergeLogDocMergePolicy extends CustomLogDocMergePolicy implements EnableMergePolicy {
+    public static class EnableMergeLogDocMergePolicy extends CustomLogDocMergePolicy {
-
-        private final ThreadLocal<Boolean> enableMerge = new ThreadLocal<Boolean>() {
-            @Override
-            protected Boolean initialValue() {
-                return Boolean.FALSE;
-            }
-        };
 
         public EnableMergeLogDocMergePolicy(LogDocMergePolicyProvider provider) {
             super(provider);
         }
 
-        @Override
-        public void enableMerge() {
-            enableMerge.set(Boolean.TRUE);
-        }
-
-        @Override
-        public void disableMerge() {
-            enableMerge.set(Boolean.FALSE);
-        }
-
-        @Override
-        public boolean isMergeEnabled() {
-            return enableMerge.get() == Boolean.TRUE;
-        }
-
-        @Override
-        public void close() {
-            enableMerge.remove();
-            super.close();
-        }
-
         @Override
         public MergeSpecification findMerges(MergeTrigger trigger, SegmentInfos infos) throws IOException {
-            if (enableMerge.get() == Boolean.FALSE) {
+            // we don't enable merges while indexing documents, we do them in the background
+            if (trigger == MergeTrigger.SEGMENT_FLUSH) {
                 return null;
             }
             return super.findMerges(trigger, infos);
         }
-
-        @Override
-        public MergeSpecification findForcedMerges(SegmentInfos infos, int maxSegmentCount, Map<SegmentInfoPerCommit, Boolean> segmentsToMerge) throws IOException {
-            if (enableMerge.get() == Boolean.FALSE) {
-                return null;
-            }
-            return super.findForcedMerges(infos, maxSegmentCount, segmentsToMerge);
-        }
-
-        @Override
-        public MergeSpecification findForcedDeletesMerges(SegmentInfos infos) throws CorruptIndexException, IOException {
-            if (enableMerge.get() == Boolean.FALSE) {
-                return null;
-            }
-            return super.findForcedDeletesMerges(infos);
-        }
     }
 }

src/main/java/org/elasticsearch/index/merge/policy/TieredMergePolicyProvider.java

@@ -19,7 +19,9 @@
 
 package org.elasticsearch.index.merge.policy;
 
-import org.apache.lucene.index.*;
+import org.apache.lucene.index.MergePolicy;
+import org.apache.lucene.index.SegmentInfos;
+import org.apache.lucene.index.TieredMergePolicy;
 import org.elasticsearch.ElasticSearchException;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.inject.Inject;
@@ -31,7 +33,6 @@ import org.elasticsearch.index.shard.AbstractIndexShardComponent;
 import org.elasticsearch.index.store.Store;
 
 import java.io.IOException;
-import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
 
@@ -219,62 +220,19 @@ public class TieredMergePolicyProvider extends AbstractIndexShardComponent imple
         }
     }
 
-    public static class EnableMergeTieredMergePolicyProvider extends CustomTieredMergePolicyProvider implements EnableMergePolicy {
+    public static class EnableMergeTieredMergePolicyProvider extends CustomTieredMergePolicyProvider {
-
-        private final ThreadLocal<Boolean> enableMerge = new ThreadLocal<Boolean>() {
-            @Override
-            protected Boolean initialValue() {
-                return Boolean.FALSE;
-            }
-        };
 
         public EnableMergeTieredMergePolicyProvider(TieredMergePolicyProvider provider) {
             super(provider);
         }
 
-        @Override
-        public void enableMerge() {
-            enableMerge.set(Boolean.TRUE);
-        }
-
-        @Override
-        public void disableMerge() {
-            enableMerge.set(Boolean.FALSE);
-        }
-
-        @Override
-        public boolean isMergeEnabled() {
-            return enableMerge.get() == Boolean.TRUE;
-        }
-
-        @Override
-        public void close() {
-            enableMerge.remove();
-            super.close();
-        }
-
         @Override
         public MergePolicy.MergeSpecification findMerges(MergeTrigger trigger, SegmentInfos infos) throws IOException {
-            if (enableMerge.get() == Boolean.FALSE) {
+            // we don't enable merges while indexing documents, we do them in the background
+            if (trigger == MergeTrigger.SEGMENT_FLUSH) {
                 return null;
             }
             return super.findMerges(trigger, infos);
         }
-
-        @Override
-        public MergeSpecification findForcedMerges(SegmentInfos infos, int maxSegmentCount, Map<SegmentInfoPerCommit, Boolean> segmentsToMerge) throws IOException {
-            if (enableMerge.get() == Boolean.FALSE) {
-                return null;
-            }
-            return super.findForcedMerges(infos, maxSegmentCount, segmentsToMerge);
-        }
-
-        @Override
-        public MergeSpecification findForcedDeletesMerges(SegmentInfos infos) throws CorruptIndexException, IOException {
-            if (enableMerge.get() == Boolean.FALSE) {
-                return null;
-            }
-            return super.findForcedDeletesMerges(infos);
-        }
     }
 }

src/main/java/org/elasticsearch/index/merge/scheduler/ConcurrentMergeSchedulerProvider.java

@@ -20,13 +20,11 @@
 package org.elasticsearch.index.merge.scheduler;
 
 import org.apache.lucene.index.*;
-import org.apache.lucene.store.AlreadyClosedException;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.index.merge.MergeStats;
-import org.elasticsearch.index.merge.policy.EnableMergePolicy;
 import org.elasticsearch.index.settings.IndexSettings;
 import org.elasticsearch.index.shard.AbstractIndexShardComponent;
 import org.elasticsearch.index.shard.ShardId;
@@ -88,19 +86,6 @@ public class ConcurrentMergeSchedulerProvider extends AbstractIndexShardComponen
 
         @Override
         public void merge(IndexWriter writer) throws CorruptIndexException, IOException {
-            try {
-                // if merge is not enabled, don't do any merging...
-                if (writer.getConfig().getMergePolicy() instanceof EnableMergePolicy) {
-                    if (!((EnableMergePolicy) writer.getConfig().getMergePolicy()).isMergeEnabled()) {
-                        return;
-                    }
-                }
-            } catch (AlreadyClosedException e) {
-                // called writer#getMergePolicy can cause an AlreadyClosed failure, so ignore it
-                // since we are doing it on close, return here and don't do the actual merge
-                // since we do it outside of a lock in the RobinEngine
-                return;
-            }
             try {
                 super.merge(writer);
             } catch (IOException e) {

src/main/java/org/elasticsearch/index/merge/scheduler/SerialMergeSchedulerProvider.java

@@ -23,12 +23,10 @@ import org.apache.lucene.index.CorruptIndexException;
 import org.apache.lucene.index.IndexWriter;
 import org.apache.lucene.index.MergeScheduler;
 import org.apache.lucene.index.TrackingSerialMergeScheduler;
-import org.apache.lucene.store.AlreadyClosedException;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.merge.MergeStats;
-import org.elasticsearch.index.merge.policy.EnableMergePolicy;
 import org.elasticsearch.index.settings.IndexSettings;
 import org.elasticsearch.index.shard.AbstractIndexShardComponent;
 import org.elasticsearch.index.shard.ShardId;
@@ -78,19 +76,6 @@ public class SerialMergeSchedulerProvider extends AbstractIndexShardComponent im
 
         @Override
         public void merge(IndexWriter writer) throws CorruptIndexException, IOException {
-            try {
-                // if merge is not enabled, don't do any merging...
-                if (writer.getConfig().getMergePolicy() instanceof EnableMergePolicy) {
-                    if (!((EnableMergePolicy) writer.getConfig().getMergePolicy()).isMergeEnabled()) {
-                        return;
-                    }
-                }
-            } catch (AlreadyClosedException e) {
-                // called writer#getMergePolicy can cause an AlreadyClosed failure, so ignore it
-                // since we are doing it on close, return here and don't do the actual merge
-                // since we do it outside of a lock in the RobinEngine
-                return;
-            }
             try {
                 super.merge(writer);
             } catch (IOException e) {