draft

Author: pxsalehi

server/src/main/java/org/elasticsearch/index/engine/ThreadPoolMergeExecutorService.java

@@ -14,6 +14,7 @@
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.engine.ThreadPoolMergeScheduler.MergeTask;
+import org.elasticsearch.index.merge.OnGoingMerge;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.util.Comparator;
@@ -73,6 +74,8 @@ public class ThreadPoolMergeExecutorService {
     private final int concurrentMergesFloorLimitForThrottling;
     private final int concurrentMergesCeilLimitForThrottling;
 
+    private volatile MergeEventConsumer mergeEventConsumer;
+
     public static @Nullable ThreadPoolMergeExecutorService maybeCreateThreadPoolMergeExecutorService(
         ThreadPool threadPool,
         Settings settings
@@ -278,6 +281,17 @@ public boolean usingMaxTargetIORateBytesPerSec() {
         return MAX_IO_RATE.getBytes() == targetIORateBytesPerSec.get();
     }
 
+    public void registerMergeEventConsumer(MergeEventConsumer consumer) {
+        assert this.mergeEventConsumer == null;
+        this.mergeEventConsumer = consumer;
+    }
+
+    public interface MergeEventConsumer {
+        void onMergeQueued(OnGoingMerge merge, long estimateMergeMemoryBytes);
+
+        void onMergeCompleted(OnGoingMerge merge);
+    }
+
     // exposed for tests
     Set<MergeTask> getRunningMergeTasks() {
         return runningMergeTasks;

server/src/main/java/org/elasticsearch/index/engine/ThreadPoolMergeScheduler.java

@@ -65,6 +65,7 @@ public class ThreadPoolMergeScheduler extends MergeScheduler implements Elastics
     private final AtomicLong doneMergeTaskCount = new AtomicLong();
     private final CountDownLatch closedWithNoRunningMerges = new CountDownLatch(1);
     private volatile boolean closed = false;
+    private final MergeMemoryEstimator mergeMemoryEstimator;
 
     public ThreadPoolMergeScheduler(
         ShardId shardId,
@@ -449,6 +450,14 @@ long estimatedMergeSize() {
             return onGoingMerge.getMerge().getStoreMergeInfo().estimatedMergeBytes();
         }
 
+        public long getEstimateMergeMemoryBytes() {
+            return mergeMemoryEstimator.estimateMergeMemoryBytes(onGoingMerge.getMerge());
+        }
+
+        public OnGoingMerge getOnGoingMerge() {
+            return onGoingMerge;
+        }
+
         @Override
         public String toString() {
             return name + (onGoingMerge.getMerge().isAborted() ? " (aborted)" : "");