Remove `ElasticsearchInterruptedException` and handle interrupt state

correctly.

InterruptedExceptions should be handled by either rethrowing or
restoring the interrupt state (i.e. calling
`Thread.currentThread().interrupt()`). This is important since the
caller of the is method or subequent method calls might also be
interested in this exception. If we ignore the interrupt state the
caller might be left unaware of the exception and blocks again on
a subsequent method.

Closes #4712

src/main/java/org/elasticsearch/ElasticsearchInterruptedException.java

@@ -1,36 +0,0 @@
-/*
- * Licensed to Elasticsearch under one or more contributor
- * license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright
- * ownership. Elasticsearch licenses this file to you under
- * the Apache License, Version 2.0 (the "License"); you may
- * not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.elasticsearch;
-
-/**
- * The same as {@link InterruptedException} simply a runtime one.
- *
- *
- */
-public class ElasticsearchInterruptedException extends ElasticsearchException {
-
-    public ElasticsearchInterruptedException(String message) {
-        super(message);
-    }
-
-    public ElasticsearchInterruptedException(String message, Throwable cause) {
-        super(message, cause);
-    }
-}

src/main/java/org/elasticsearch/action/ActionFuture.java

@@ -34,9 +34,9 @@ import java.util.concurrent.TimeUnit;
 public interface ActionFuture<T> extends Future<T> {
 
     /**
-     * Similar to {@link #get()}, just wrapping the {@link InterruptedException} with
-     * {@link org.elasticsearch.ElasticsearchInterruptedException}, and throwing the actual
-     * cause of the {@link java.util.concurrent.ExecutionException}.
+     * Similar to {@link #get()}, just catching the {@link InterruptedException} with
+     * restoring the interrupted state on the thread and throwing an {@link org.elasticsearch.ElasticsearchIllegalStateException},
+     * and throwing the actual cause of the {@link java.util.concurrent.ExecutionException}.
      * <p/>
      * <p>Note, the actual cause is unwrapped to the actual failure (for example, unwrapped
      * from {@link org.elasticsearch.transport.RemoteTransportException}. The root failure is
@@ -45,9 +45,9 @@ public interface ActionFuture<T> extends Future<T> {
     T actionGet() throws ElasticsearchException;
 
     /**
-     * Similar to {@link #get(long, java.util.concurrent.TimeUnit)}, just wrapping the {@link InterruptedException} with
-     * {@link org.elasticsearch.ElasticsearchInterruptedException}, and throwing the actual
-     * cause of the {@link java.util.concurrent.ExecutionException}.
+     * Similar to {@link #get(long, java.util.concurrent.TimeUnit)}, just catching the {@link InterruptedException} with
+     * restoring the interrupted state on the thread and throwing an {@link org.elasticsearch.ElasticsearchIllegalStateException},
+     * and throwing the actual cause of the {@link java.util.concurrent.ExecutionException}.
      * <p/>
      * <p>Note, the actual cause is unwrapped to the actual failure (for example, unwrapped
      * from {@link org.elasticsearch.transport.RemoteTransportException}. The root failure is
@@ -56,9 +56,9 @@ public interface ActionFuture<T> extends Future<T> {
     T actionGet(String timeout) throws ElasticsearchException;
 
     /**
-     * Similar to {@link #get(long, java.util.concurrent.TimeUnit)}, just wrapping the {@link InterruptedException} with
-     * {@link org.elasticsearch.ElasticsearchInterruptedException}, and throwing the actual
-     * cause of the {@link java.util.concurrent.ExecutionException}.
+     * Similar to {@link #get(long, java.util.concurrent.TimeUnit)}, just catching the {@link InterruptedException} with
+     * restoring the interrupted state on the thread and throwing an {@link org.elasticsearch.ElasticsearchIllegalStateException},
+     * and throwing the actual cause of the {@link java.util.concurrent.ExecutionException}.
      * <p/>
      * <p>Note, the actual cause is unwrapped to the actual failure (for example, unwrapped
      * from {@link org.elasticsearch.transport.RemoteTransportException}. The root failure is
@@ -69,9 +69,9 @@ public interface ActionFuture<T> extends Future<T> {
     T actionGet(long timeoutMillis) throws ElasticsearchException;
 
     /**
-     * Similar to {@link #get(long, java.util.concurrent.TimeUnit)}, just wrapping the {@link InterruptedException} with
-     * {@link org.elasticsearch.ElasticsearchInterruptedException}, and throwing the actual
-     * cause of the {@link java.util.concurrent.ExecutionException}.
+     * Similar to {@link #get(long, java.util.concurrent.TimeUnit)}, just catching the {@link InterruptedException} with
+     * restoring the interrupted state on the thread and throwing an {@link org.elasticsearch.ElasticsearchIllegalStateException},
+     * and throwing the actual cause of the {@link java.util.concurrent.ExecutionException}.
      * <p/>
      * <p>Note, the actual cause is unwrapped to the actual failure (for example, unwrapped
      * from {@link org.elasticsearch.transport.RemoteTransportException}. The root failure is
@@ -80,9 +80,9 @@ public interface ActionFuture<T> extends Future<T> {
     T actionGet(long timeout, TimeUnit unit) throws ElasticsearchException;
 
     /**
-     * Similar to {@link #get(long, java.util.concurrent.TimeUnit)}, just wrapping the {@link InterruptedException} with
-     * {@link org.elasticsearch.ElasticsearchInterruptedException}, and throwing the actual
-     * cause of the {@link java.util.concurrent.ExecutionException}.
+     * Similar to {@link #get(long, java.util.concurrent.TimeUnit)}, just catching the {@link InterruptedException} with
+     * restoring the interrupted state on the thread and throwing an {@link org.elasticsearch.ElasticsearchIllegalStateException},
+     * and throwing the actual cause of the {@link java.util.concurrent.ExecutionException}.
      * <p/>
      * <p>Note, the actual cause is unwrapped to the actual failure (for example, unwrapped
      * from {@link org.elasticsearch.transport.RemoteTransportException}. The root failure is

src/main/java/org/elasticsearch/action/support/AdapterActionFuture.java

@@ -20,7 +20,7 @@
 package org.elasticsearch.action.support;
 
 import org.elasticsearch.ElasticsearchException;
-import org.elasticsearch.ElasticsearchInterruptedException;
+import org.elasticsearch.ElasticsearchIllegalStateException;
 import org.elasticsearch.ElasticsearchTimeoutException;
 import org.elasticsearch.action.ActionFuture;
 import org.elasticsearch.action.ActionListener;
@@ -44,7 +44,8 @@ public abstract class AdapterActionFuture<T, L> extends BaseFuture<T> implements
         try {
             return get();
         } catch (InterruptedException e) {
-            throw new ElasticsearchInterruptedException(e.getMessage());
+            Thread.currentThread().interrupt();
+            throw new ElasticsearchIllegalStateException("Future got interrupted", e);
         } catch (ExecutionException e) {
             throw rethrowExecutionException(e);
         }
@@ -72,7 +73,8 @@ public abstract class AdapterActionFuture<T, L> extends BaseFuture<T> implements
         } catch (TimeoutException e) {
             throw new ElasticsearchTimeoutException(e.getMessage());
         } catch (InterruptedException e) {
-            throw new ElasticsearchInterruptedException(e.getMessage());
+            Thread.currentThread().interrupt();
+            throw new ElasticsearchIllegalStateException("Future got interrupted", e);
         } catch (ExecutionException e) {
             throw rethrowExecutionException(e);
         }

src/main/java/org/elasticsearch/common/util/concurrent/EsAbortPolicy.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.common.util.concurrent;
 
 import org.elasticsearch.ElasticsearchIllegalStateException;
-import org.elasticsearch.ElasticsearchInterruptedException;
 import org.elasticsearch.common.metrics.CounterMetric;
 
 import java.util.concurrent.BlockingQueue;
@@ -44,7 +43,8 @@ public class EsAbortPolicy implements XRejectedExecutionHandler {
                 try {
                     ((SizeBlockingQueue) queue).forcePut(r);
                 } catch (InterruptedException e) {
-                    throw new ElasticsearchInterruptedException(e.getMessage(), e);
+                    Thread.currentThread().interrupt();
+                    throw new ElasticsearchIllegalStateException("forced execution, but got interrupted", e);
                 }
                 return;
             }

src/main/java/org/elasticsearch/index/service/InternalIndexService.java

@@ -24,7 +24,6 @@ import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.UnmodifiableIterator;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchIllegalStateException;
-import org.elasticsearch.ElasticsearchInterruptedException;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.inject.*;
@@ -276,7 +275,8 @@ public class InternalIndexService extends AbstractIndexComponent implements Inde
         try {
             latch.await();
         } catch (InterruptedException e) {
-            throw new ElasticsearchInterruptedException("interrupted closing index [ " + index().name() + "]", e);
+            logger.debug("Interrupted closing index [{}]", e, index().name());
+            Thread.currentThread().interrupt();
         }
     }
 

src/main/java/org/elasticsearch/transport/PlainTransportFuture.java

@@ -20,7 +20,7 @@
 package org.elasticsearch.transport;
 
 import org.elasticsearch.ElasticsearchException;
-import org.elasticsearch.ElasticsearchInterruptedException;
+import org.elasticsearch.ElasticsearchIllegalStateException;
 import org.elasticsearch.ElasticsearchTimeoutException;
 import org.elasticsearch.common.util.concurrent.BaseFuture;
 
@@ -44,7 +44,8 @@ public class PlainTransportFuture<V extends TransportResponse> extends BaseFutur
         try {
             return get();
         } catch (InterruptedException e) {
-            throw new ElasticsearchInterruptedException(e.getMessage());
+            Thread.currentThread().interrupt();
+            throw new ElasticsearchIllegalStateException("Future got interrupted", e);
         } catch (ExecutionException e) {
             if (e.getCause() instanceof ElasticsearchException) {
                 throw (ElasticsearchException) e.getCause();
@@ -61,7 +62,8 @@ public class PlainTransportFuture<V extends TransportResponse> extends BaseFutur
         } catch (TimeoutException e) {
             throw new ElasticsearchTimeoutException(e.getMessage());
         } catch (InterruptedException e) {
-            throw new ElasticsearchInterruptedException(e.getMessage());
+            Thread.currentThread().interrupt();
+            throw new ElasticsearchIllegalStateException("Future got interrupted", e);
         } catch (ExecutionException e) {
             if (e.getCause() instanceof ElasticsearchException) {
                 throw (ElasticsearchException) e.getCause();

src/main/java/org/elasticsearch/tribe/TribeService.java

@@ -22,7 +22,7 @@ package org.elasticsearch.tribe;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import org.elasticsearch.ElasticsearchException;
-import org.elasticsearch.ElasticsearchInterruptedException;
+import org.elasticsearch.ElasticsearchIllegalStateException;
 import org.elasticsearch.cluster.*;
 import org.elasticsearch.cluster.block.ClusterBlock;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
@@ -150,8 +150,11 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
 
             @Override
             public void onFailure(String source, Throwable t) {
-                logger.error("{}", t, source);
-                latch.countDown();
+                try {
+                    logger.error("{}", t, source);
+                } finally {
+                    latch.countDown();
+                }
             }
 
             @Override
@@ -162,7 +165,8 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
         try {
             latch.await();
         } catch (InterruptedException e) {
-            throw new ElasticsearchInterruptedException(e.getMessage(), e);
+            Thread.currentThread().interrupt();
+            throw new ElasticsearchIllegalStateException("Interrupted while starting [" + this.getClass().getSimpleName()+ "]", e);
         }
         for (InternalNode node : nodes) {
             try {