[Presto-on-spark native] [localMode] Close executor service on shutdown

presto-spark-base/src/main/java/com/facebook/presto/spark/execution/NativeExecutionProcess.java

@@ -152,9 +152,12 @@ public void close()
         if (process != null && process.isAlive()) {
             process.destroy();
             try {
-                // For native process it takes 10s to initiate SHUTDOWN. The task cleanup interval is 60s. To be sure task cleanup is run at least once we just roughly double the
-                // wait time.
-                process.waitFor(120, TimeUnit.SECONDS);
+                // This 1 sec is arbitrary. Ideally, we do not need to be give any heads up
+                // to CPP process on presto-on-spark native, because the resources
+                // are reclaimed by the container manager.
+                // For localmode, we still want to provide an opportunity for
+                // graceful termination as there is no resource/container manager.
+                process.waitFor(1, TimeUnit.SECONDS);
             }
             catch (InterruptedException e) {
                 e.printStackTrace();

presto-spark-launcher/src/main/java/com/facebook/presto/spark/launcher/PrestoSparkRunner.java

@@ -166,7 +166,19 @@ private void execute(IPrestoSparkQueryExecutionFactory queryExecutionFactory, Pr
     @Override
     public void close()
     {
+        // Shutdown the driver Airlift application
         driverPrestoSparkService.close();
+
+        // If we are in localMode, the executor spawns the Executor Airlift application
+        // (which is long-running and holds onto resources) on the same JVM.
+        //
+        // On query completion, the SparkContext shutdown calls the Driver Airlift
+        // application shutdown, but it has no hook to call Executor Airlift application
+        // shutdown. So the query hangs forever.
+        //
+        // This code, prevents this hanging state by explicitly calling the
+        // Executor Airlift application shutdown.
+        DistributionBasedPrestoSparkTaskExecutorFactoryProvider.close();
     }
 
     private static IPrestoSparkServiceFactory createServiceFactory(File directory)
@@ -324,5 +336,12 @@ public static void checkEquals(String name, Object first, Object second)
                 throw new IllegalStateException(format("%s is different: %s != %s", name, first, second));
             }
         }
+
+        public static synchronized void close()
+        {
+            if (service != null) {
+                service.close();
+            }
+        }
     }
 }