SOLR-7956: Remove a few more interrupt causes.

git-svn-id: https://svn.apache.org/repos/asf/lucene/dev/trunk@1700185 13f79535-47bb-0310-9956-ffa450edef68
2015-08-31 02:22:09 +00:00 · 2015-08-31 02:22:09 +00:00 · 431ed53d46
parent a956d3ffe2
commit 431ed53d46
8 changed files with 73 additions and 71 deletions
--- a/solr/contrib/map-reduce/src/java/org/apache/solr/hadoop/GoLive.java
+++ b/solr/contrib/map-reduce/src/java/org/apache/solr/hadoop/GoLive.java
@ -16,16 +16,6 @@
 */
 package org.apache.solr.hadoop;

-import org.apache.hadoop.fs.FileStatus;
-import org.apache.solr.client.solrj.SolrServerException;
-import org.apache.solr.client.solrj.impl.CloudSolrClient;
-import org.apache.solr.client.solrj.impl.HttpSolrClient;
-import org.apache.solr.client.solrj.request.CoreAdminRequest;
-import org.apache.solr.common.util.ExecutorUtil;
-import org.apache.solr.hadoop.MapReduceIndexerTool.Options;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.HashSet;
@ -35,12 +25,21 @@ import java.util.concurrent.Callable;
 import java.util.concurrent.CompletionService;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorCompletionService;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;

+import org.apache.hadoop.fs.FileStatus;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.impl.CloudSolrClient;
+import org.apache.solr.client.solrj.impl.HttpSolrClient;
+import org.apache.solr.client.solrj.request.CoreAdminRequest;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.hadoop.MapReduceIndexerTool.Options;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 /**
 * The optional (parallel) GoLive phase merges the output shards of the previous
 * phase into a set of live customer facing Solr servers, typically a SolrCloud.
@ -164,7 +163,7 @@ class GoLive {
      success = true;
      return true;
    } finally {
-      shutdownNowAndAwaitTermination(executor);
+      ExecutorUtil.shutdownAndAwaitTermination(executor);
      float secs = (System.nanoTime() - start) / (float)(10^9);
      LOG.info("Live merging of index shards into Solr cluster took " + secs + " secs");
      if (success) {
@ -177,25 +176,6 @@ class GoLive {
    // if an output dir does not exist, we should fail and do no merge?
  }
  
-  private void shutdownNowAndAwaitTermination(ExecutorService pool) {
-    pool.shutdown(); // Disable new tasks from being submitted
-    pool.shutdownNow(); // Cancel currently executing tasks
-    boolean shutdown = false;
-    while (!shutdown) {
-      try {
-        // Wait a while for existing tasks to terminate
-        shutdown = pool.awaitTermination(5, TimeUnit.SECONDS);
-      } catch (InterruptedException ie) {
-        // Preserve interrupt status
-        Thread.currentThread().interrupt();
-      }
-      if (!shutdown) {
-        pool.shutdownNow(); // Cancel currently executing tasks
-      }
-    }
-  }
-  
-  
  private static final class Request {
    Exception e;
    boolean success = false;
--- a/solr/core/src/java/org/apache/solr/handler/CdcrReplicatorScheduler.java
+++ b/solr/core/src/java/org/apache/solr/handler/CdcrReplicatorScheduler.java
@ -102,6 +102,8 @@ class CdcrReplicatorScheduler {

  void shutdown() {
    if (isStarted) {
+      // interrupts are often dangerous in Lucene / Solr code, but the
+      // test for this will leak threads without
      replicatorsPool.shutdown();
      try {
        replicatorsPool.awaitTermination(60, TimeUnit.SECONDS);
--- a/solr/core/src/java/org/apache/solr/handler/CdcrUpdateLogSynchronizer.java
+++ b/solr/core/src/java/org/apache/solr/handler/CdcrUpdateLogSynchronizer.java
@ -1,5 +1,10 @@
 package org.apache.solr.handler;

+import java.io.IOException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
 /*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
@ -35,11 +40,6 @@ import org.apache.solr.util.DefaultSolrThreadFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

-import java.io.IOException;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
 /**
 * <p>
 * Synchronize periodically the update log of non-leader nodes with their leaders.
@ -101,6 +101,8 @@ class CdcrUpdateLogSynchronizer implements CdcrStateManager.CdcrStateObserver {

  void shutdown() {
    if (scheduler != null) {
+      // interrupts are often dangerous in Lucene / Solr code, but the
+      // test for this will leak threads without
      scheduler.shutdownNow();
      scheduler = null;
    }
--- a/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
+++ b/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
@ -553,8 +553,7 @@ public class IndexFetcher {
      markReplicationStop();
      dirFileFetcher = null;
      localFileFetcher = null;
-      if (fsyncService != null && !fsyncService.isShutdown()) fsyncService
-          .shutdownNow();
+      if (fsyncService != null && !fsyncService.isShutdown()) fsyncService.shutdown();
      fsyncService = null;
      stop = false;
      fsyncException = null;
--- a/solr/core/src/java/org/apache/solr/schema/ManagedIndexSchema.java
+++ b/solr/core/src/java/org/apache/solr/schema/ManagedIndexSchema.java
@ -16,6 +16,27 @@ package org.apache.solr.schema;
 * limitations under the License.
 */

+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.io.StringWriter;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
 import org.apache.commons.io.IOUtils;
 import org.apache.lucene.analysis.Analyzer;
 import org.apache.lucene.analysis.util.CharFilterFactory;
@ -53,28 +74,6 @@ import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.data.Stat;
 import org.xml.sax.InputSource;

-import java.io.File;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.OutputStreamWriter;
-import java.io.StringWriter;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-
 /** Solr-managed schema - non-user-editable, but can be mutable via internal and external REST API requests. */
 public final class ManagedIndexSchema extends IndexSchema {

@ -269,7 +268,7 @@ public final class ManagedIndexSchema extends IndexSchema {
      Thread.currentThread().interrupt();
    } finally {
      if (!parallelExecutor.isShutdown())
-        parallelExecutor.shutdownNow();
+        parallelExecutor.shutdown();
    }

    log.info("Took {}ms for {} replicas to apply schema update version {} for collection {}",
--- a/solr/core/src/java/org/apache/solr/update/CommitTracker.java
+++ b/solr/core/src/java/org/apache/solr/update/CommitTracker.java
@ -92,7 +92,7 @@ public final class CommitTracker implements Runnable {
      pending.cancel(false);
      pending = null;
    }
-    scheduler.shutdownNow();
+    scheduler.shutdown();
  }
  
  /** schedule individual commits */
--- a/solr/core/src/java/org/apache/solr/update/processor/DocExpirationUpdateProcessorFactory.java
+++ b/solr/core/src/java/org/apache/solr/update/processor/DocExpirationUpdateProcessorFactory.java
@ -23,6 +23,7 @@ import org.apache.solr.common.SolrException;

 import static org.apache.solr.common.SolrException.ErrorCode.*;

+import org.apache.solr.common.util.ExecutorUtil;
 import org.apache.solr.common.util.NamedList;
 import org.apache.solr.common.SolrInputDocument;
 import org.apache.solr.common.cloud.Replica;
@ -266,11 +267,11 @@ public final class DocExpirationUpdateProcessorFactory

    core.addCloseHook(new CloseHook() {
      public void postClose(SolrCore core) {
-        // update handler is gone, hard terminiate anything that's left.
+        // update handler is gone, terminate anything that's left.

        if (executor.isTerminating()) {
-          log.info("Triggering hard close of DocExpiration Executor");
-          executor.shutdownNow();
+          log.info("Waiting for close of DocExpiration Executor");
+          ExecutorUtil.shutdownAndAwaitTermination(executor);
        }
      }
      public void preClose(SolrCore core) {
--- a/solr/solrj/src/java/org/apache/solr/common/util/ExecutorUtil.java
+++ b/solr/solrj/src/java/org/apache/solr/common/util/ExecutorUtil.java
@ -82,20 +82,39 @@ public class ExecutorUtil {
  // this if you know what you are doing - you probably want shutdownAndAwaitTermination.
  // Marked as Deprecated to discourage use.
  @Deprecated
-  public static void shutdownNowAndAwaitTermination(ExecutorService pool) {
-    pool.shutdown(); // Disable new tasks from being submitted
-    pool.shutdownNow(); // Cancel currently executing tasks  - NOTE: this interrupts!
+  public static void shutdownWithInterruptAndAwaitTermination(ExecutorService pool) {
+    pool.shutdownNow(); // Cancel currently executing tasks - NOTE: this interrupts!
    boolean shutdown = false;
    while (!shutdown) {
      try {
        // Wait a while for existing tasks to terminate
-        shutdown = pool.awaitTermination(1, TimeUnit.SECONDS);
+        shutdown = pool.awaitTermination(60, TimeUnit.SECONDS);
      } catch (InterruptedException ie) {
        // Preserve interrupt status
        Thread.currentThread().interrupt();
      }
-      if (!shutdown) {
+    }
+  }
+  
+  // ** This will interrupt the threads! ** Lucene and Solr do not like this because it can close channels, so only use
+  // this if you know what you are doing - you probably want shutdownAndAwaitTermination.
+  // Marked as Deprecated to discourage use.
+  @Deprecated
+  public static void shutdownAndAwaitTerminationWithInterrupt(ExecutorService pool) {
+    pool.shutdown(); // Disable new tasks from being submitted
+    boolean shutdown = false;
+    boolean interrupted = false;
+    while (!shutdown) {
+      try {
+        // Wait a while for existing tasks to terminate
+        shutdown = pool.awaitTermination(60, TimeUnit.SECONDS);
+      } catch (InterruptedException ie) {
+        // Preserve interrupt status
+        Thread.currentThread().interrupt();
+      }
+      if (!shutdown && !interrupted) {
        pool.shutdownNow(); // Cancel currently executing tasks - NOTE: this interrupts!
+        interrupted = true;
      }
    }
  }
@ -106,7 +125,7 @@ public class ExecutorUtil {
    while (!shutdown) {
      try {
        // Wait a while for existing tasks to terminate
-        shutdown = pool.awaitTermination(1, TimeUnit.SECONDS);
+        shutdown = pool.awaitTermination(60, TimeUnit.SECONDS);
      } catch (InterruptedException ie) {
        // Preserve interrupt status
        Thread.currentThread().interrupt();