Use explicit query executor to start and fail execution

Author: JunhyungSong

core/trino-main/src/main/java/io/trino/dispatcher/LocalDispatchQuery.java

@@ -126,8 +126,8 @@ private void waitForMinimumWorkers()
             }
             ListenableFuture<Void> minimumWorkerFuture = clusterSizeMonitor.waitForMinimumWorkers(executionMinCount, getRequiredWorkersMaxWait(session));
             // when worker requirement is met, start the execution
-            addSuccessCallback(minimumWorkerFuture, () -> startExecution(queryExecution));
-            addExceptionCallback(minimumWorkerFuture, throwable -> queryExecutor.execute(() -> stateMachine.transitionToFailed(throwable)));
+            addSuccessCallback(minimumWorkerFuture, () -> startExecution(queryExecution), queryExecutor);
+            addExceptionCallback(minimumWorkerFuture, throwable -> stateMachine.transitionToFailed(throwable), queryExecutor);
 
             // cancel minimumWorkerFuture if query fails for some reason or is cancelled by user
             stateMachine.addStateChangeListener(state -> {
@@ -140,25 +140,23 @@ private void waitForMinimumWorkers()
 
     private void startExecution(QueryExecution queryExecution)
     {
-        queryExecutor.execute(() -> {
-            if (stateMachine.transitionToDispatching()) {
-                try {
-                    querySubmitter.accept(queryExecution);
-                    if (notificationSentOrGuaranteed.compareAndSet(false, true)) {
-                        queryExecution.addFinalQueryInfoListener(queryMonitor::queryCompletedEvent);
-                    }
-                }
-                catch (Throwable t) {
-                    // this should never happen but be safe
-                    stateMachine.transitionToFailed(t);
-                    log.error(t, "query submitter threw exception");
-                    throw t;
-                }
-                finally {
-                    submitted.set(null);
+        if (stateMachine.transitionToDispatching()) {
+            try {
+                querySubmitter.accept(queryExecution);
+                if (notificationSentOrGuaranteed.compareAndSet(false, true)) {
+                    queryExecution.addFinalQueryInfoListener(queryMonitor::queryCompletedEvent);
                 }
             }
-        });
+            catch (Throwable t) {
+                // this should never happen but be safe
+                stateMachine.transitionToFailed(t);
+                log.error(t, "query submitter threw exception");
+                throw t;
+            }
+            finally {
+                submitted.set(null);
+            }
+        }
     }
 
     @Override