fix for stale metrics/suspend thread after river deletion

Author: jprante

src/main/java/org/xbib/elasticsearch/plugin/jdbc/RiverThread.java

@@ -119,10 +119,8 @@ public P get() {
             for (RiverPipeline pipeline : pipelines) {
                 pipeline.setInterrupted(true);
             }
-            if (metricsThread != null) {
-                logger.debug("interrupting metrics thread");
-                metricsThread.interrupt();
-            }
+            closeMetricThread();
+            closeSuspensionThread();
             Thread.currentThread().interrupt();
             logger.warn("interrupted");
         } catch (Throwable t) {
@@ -141,6 +139,11 @@ protected void afterPipelineExecutions() throws Exception {
         if (executor != null) {
             executor.shutdown();
         }
+        closeMetricThread();
+        closeSuspensionThread();
+    }
+
+    private void closeMetricThread() {
         if (metricsThreadPoolExecutor != null) {
             logger.debug("shutting down metrics thread scheduler");
             metricsThreadPoolExecutor.shutdownNow();
@@ -149,6 +152,9 @@ protected void afterPipelineExecutions() throws Exception {
         if (metricsThread != null) {
             metricsThread.interrupt();
         }
+    }
+
+    private void closeSuspensionThread() {
         if (suspensionThreadPoolExecutor != null) {
             logger.debug("shutting down suspension thread scheduler");
             suspensionThreadPoolExecutor.shutdownNow();