make MergeTask a Runnable rather than an AbstractRunnable

Author: albertzaharovits

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

@@ -61,7 +61,7 @@ private ThreadPoolMergeQueue(ThreadPool threadPool) {
 
     void submitMergeTask(MergeTask mergeTask) {
         enqueueMergeTask(mergeTask);
-        executeNextMergeTask();
+        executeSmallestMergeTask();
     }
 
     void enqueueMergeTask(MergeTask mergeTask) {
@@ -72,25 +72,31 @@ void enqueueMergeTask(MergeTask mergeTask) {
         }
     }
 
-    private void executeNextMergeTask() {
-        executorService.execute(() -> {
-            // one such task always executes a SINGLE merge task; this is important for merge queue statistics
-            while (true) {
-                MergeTask smallestMergeTask;
-                try {
-                    // will block if there are backlogged merges until they're enqueued again
-                    smallestMergeTask = queuedMergeTasks.take();
-                } catch (InterruptedException e) {
-                    Thread.currentThread().interrupt();
-                    return;
-                }
-                if (smallestMergeTask.runNowOrBacklog()) {
-                    runMergeTask(smallestMergeTask);
-                    // one runnable one merge task
-                    return;
+    private void executeSmallestMergeTask() {
+        final AtomicReference<MergeTask> smallestMergeTask = new AtomicReference<>();
+        try {
+            executorService.execute(() -> {
+                // one such task always executes a SINGLE merge task; this is important for merge queue statistics
+                while (true) {
+                    try {
+                        // will block if there are backlogged merges until they're enqueued again
+                        smallestMergeTask.set(queuedMergeTasks.take());
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                        return;
+                    }
+                    if (smallestMergeTask.get().runNowOrBacklog()) {
+                        runMergeTask(smallestMergeTask.get());
+                        // one runnable one merge task
+                        return;
+                    }
                 }
+            });
+        } catch (Exception e) {
+            if (smallestMergeTask.get() != null) {
+                smallestMergeTask.get().onRejection(e);
             }
-        });
+        }
     }
 
     private void runMergeTask(MergeTask mergeTask) {

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

@@ -21,7 +21,6 @@
 import org.apache.lucene.store.RateLimitedIndexOutput;
 import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.settings.Setting;
-import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.MergeSchedulerConfig;
@@ -228,7 +227,7 @@ public IndexOutput createOutput(String name, IOContext context) throws IOExcepti
         };
     }
 
-    final class MergeTask extends AbstractRunnable implements Comparable<MergeTask> {
+    final class MergeTask implements Runnable, Comparable<MergeTask> {
         private final String name;
         private final AtomicLong mergeStartTimeNS;
         private final MergeSource mergeSource;
@@ -271,7 +270,7 @@ public boolean isRunning() {
         }
 
         @Override
-        public void doRun() throws Exception {
+        public void run() {
             if (mergeStartTimeNS.compareAndSet(0L, System.nanoTime()) == false) {
                 throw new IllegalStateException("Cannot run the same merge task multiple times");
             }
@@ -306,57 +305,41 @@ public void doRun() throws Exception {
                     abortOnGoingMerge();
                     handleMergeException(t);
                 }
-            }
-        }
-
-        @Override
-        public void onAfter() {
-            if (isRunning() == false) {
-                throw new IllegalStateException("onAfter must only be invoked after doRun");
-            }
-            try {
-                if (verbose()) {
-                    message(String.format(Locale.ROOT, "merge task %s end", this));
-                }
-                afterMerge(onGoingMerge);
             } finally {
-                long tookMS = TimeValue.nsecToMSec(System.nanoTime() - mergeStartTimeNS.get());
                 try {
-                    mergeTracking.mergeFinished(onGoingMerge.getMerge(), onGoingMerge, tookMS);
+                    if (verbose()) {
+                        message(String.format(Locale.ROOT, "merge task %s end", this));
+                    }
+                    afterMerge(onGoingMerge);
                 } finally {
-                    mergeDone(this);
-                    // kick-off next merge, if any
-                    MergePolicy.OneMerge nextMerge = null;
+                    long tookMS = TimeValue.nsecToMSec(System.nanoTime() - mergeStartTimeNS.get());
                     try {
-                        nextMerge = mergeSource.getNextMerge();
-                    } catch (IllegalStateException e) {
-                        if (verbose()) {
-                            message("merge task poll failed, likely that index writer is failed");
+                        mergeTracking.mergeFinished(onGoingMerge.getMerge(), onGoingMerge, tookMS);
+                    } finally {
+                        mergeDone(this);
+                        // kick-off next merge, if any
+                        MergePolicy.OneMerge nextMerge = null;
+                        try {
+                            nextMerge = mergeSource.getNextMerge();
+                        } catch (IllegalStateException e) {
+                            if (verbose()) {
+                                message("merge task poll failed, likely that index writer is failed");
+                            }
+                            // ignore exception, we expect the IW failure to be logged elsewhere
+                        }
+                        if (nextMerge != null) {
+                            submitNewMergeTask(mergeSource, nextMerge, MergeTrigger.MERGE_FINISHED);
                         }
-                        // ignore exception, we expect the IW failure to be logged elsewhere
-                    }
-                    if (nextMerge != null) {
-                        submitNewMergeTask(mergeSource, nextMerge, MergeTrigger.MERGE_FINISHED);
                     }
                 }
             }
         }
 
-        @Override
-        public void onFailure(Exception e) {
-            // no-op, should not be called
-        }
-
-        @Override
         public void onRejection(Exception e) {
-            if (isRunning()) {
-                throw new IllegalStateException("A running merge cannot be rejected for running");
-            }
             if (verbose()) {
                 message(String.format(Locale.ROOT, "merge task [%s] rejected by thread pool, aborting", onGoingMerge.getId()));
             }
             abortOnGoingMerge();
-            mergeDone(this);
         }
 
         private void abortOnGoingMerge() {