[SPARK-25183][SQL] Spark HiveServer2 to use Spark ShutdownHookManager

## What changes were proposed in this pull request?

Switch `org.apache.hive.service.server.HiveServer2` to register its shutdown callback with Spark's `ShutdownHookManager`, rather than direct with the Java Runtime callback.

This avoids race conditions in shutdown where the filesystem is shutdown before the flush/write/rename of the event log is completed, particularly on object stores where the write and rename can be slow.

## How was this patch tested?

There's no explicit unit for test this, which is consistent with every other shutdown hook in the codebase.

* There's an implicit test when the scalatest process is halted.
* More manual/integration testing is needed.

HADOOP-15679 has added the ability to explicitly execute the hadoop shutdown hook sequence which spark uses; that could be stabilized for testing if desired, after which all the spark hooks could be tested. Until then: external system tests only.

Author: Steve Loughran <[email protected]>

Closes apache#22186 from steveloughran/BUG/SPARK-25183-shutdown.

Author: steveloughran

sql/hive-thriftserver/src/main/java/org/apache/hive/service/server/HiveServer2.java

@@ -20,6 +20,9 @@
 
 import java.util.Properties;
 
+import scala.runtime.AbstractFunction0;
+import scala.runtime.BoxedUnit;
+
 import org.apache.commons.cli.GnuParser;
 import org.apache.commons.cli.HelpFormatter;
 import org.apache.commons.cli.Option;
@@ -39,6 +42,8 @@
 import org.apache.hive.service.cli.thrift.ThriftCLIService;
 import org.apache.hive.service.cli.thrift.ThriftHttpCLIService;
 
+import org.apache.spark.util.ShutdownHookManager;
+
 /**
  * HiveServer2.
  *
@@ -67,13 +72,23 @@ public synchronized void init(HiveConf hiveConf) {
     super.init(hiveConf);
 
     // Add a shutdown hook for catching SIGTERM & SIGINT
-    final HiveServer2 hiveServer2 = this;
-    Runtime.getRuntime().addShutdownHook(new Thread() {
-      @Override
-      public void run() {
-        hiveServer2.stop();
-      }
-    });
+    // this must be higher than the Hadoop Filesystem priority of 10,
+    // which the default priority is.
+    // The signature of the callback must match that of a scala () -> Unit
+    // function
+    ShutdownHookManager.addShutdownHook(
+        new AbstractFunction0<BoxedUnit>() {
+          public BoxedUnit apply() {
+            try {
+              LOG.info("Hive Server Shutdown hook invoked");
+              stop();
+            } catch (Throwable e) {
+              LOG.warn("Ignoring Exception while stopping Hive Server from shutdown hook",
+                  e);
+            }
+            return BoxedUnit.UNIT;
+          }
+        });
   }
 
   public static boolean isHTTPTransportMode(HiveConf hiveConf) {
@@ -95,7 +110,6 @@ public synchronized void start() {
   @Override
   public synchronized void stop() {
     LOG.info("Shutting down HiveServer2");
-    HiveConf hiveConf = this.getHiveConf();
     super.stop();
   }