Not propagating TimeoutException from Retry2::awaitClose (elastic#92773)

Logging a message rather than propagating a TimeoutException from Retry2::awaitClose

Author: masseyke

server/src/main/java/org/elasticsearch/action/bulk/BulkProcessor2.java

@@ -25,7 +25,6 @@
 
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.BiConsumer;
 
@@ -237,7 +236,7 @@ public static Builder builder(
      * @param unit    The time unit of the {@code timeout} argument
      * @throws InterruptedException If the current thread is interrupted
      */
-    public void awaitClose(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
+    public void awaitClose(long timeout, TimeUnit unit) throws InterruptedException {
         synchronized (mutex) {
             if (closed) {
                 return;

server/src/main/java/org/elasticsearch/action/bulk/Retry2.java

@@ -119,14 +119,18 @@ private void retry(
      * @param timeout
      * @param unit
      */
-    void awaitClose(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
+    void awaitClose(long timeout, TimeUnit unit) throws InterruptedException {
         isClosing = true;
         /*
          * This removes the party that was placed in the phaser at initialization so that the phaser will terminate once all in-flight
          * requests have been completed (i.e. this makes it possible that the number of parties can become 0).
          */
         inFlightRequestsPhaser.arriveAndDeregister();
-        inFlightRequestsPhaser.awaitAdvanceInterruptibly(0, timeout, unit);
+        try {
+            inFlightRequestsPhaser.awaitAdvanceInterruptibly(0, timeout, unit);
+        } catch (TimeoutException e) {
+            logger.debug("Timed out waiting for all requests to complete during awaitClose");
+        }
     }
 
     /**

server/src/test/java/org/elasticsearch/action/bulk/Retry2Tests.java

@@ -14,6 +14,7 @@
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.client.NoOpClient;
@@ -26,11 +27,12 @@
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
 
@@ -176,7 +178,7 @@ public void testAwaitClose() throws Exception {
             retry.awaitClose(1, TimeUnit.SECONDS);
         }
         /*
-         * Most requests are complete but one request is hung so awaitClose ought to throw a TimeoutException
+         * Most requests are complete but one request is hung so awaitClose ought to wait the full timeout period
          */
         {
             Retry2 retry = new Retry2(CALLS_TO_FAIL);
@@ -200,7 +202,14 @@ public void testAwaitClose() throws Exception {
             retry.consumeRequestWithRetries((bulkRequest1, listener1) -> {
                 // never calls onResponse or onFailure
             }, bulkRequest, listener);
-            expectThrows(TimeoutException.class, () -> retry.awaitClose(200, TimeUnit.MILLISECONDS));
+            long waitTimeMillis = randomLongBetween(20, 200);
+            // Make sure that awaitClose completes without exception, and takes at least waitTimeMillis
+            long startTimeNanos = System.nanoTime();
+            retry.awaitClose(waitTimeMillis, TimeUnit.MILLISECONDS);
+            long stopTimeNanos = System.nanoTime();
+            long runtimeMillis = TimeValue.timeValueNanos((stopTimeNanos - startTimeNanos)).millis();
+            assertThat(runtimeMillis, greaterThanOrEqualTo(waitTimeMillis));
+            assertThat(runtimeMillis, lessThanOrEqualTo(2 * waitTimeMillis));
         }
     }
 

x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/history/ILMHistoryStore.java

@@ -33,7 +33,6 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import java.util.stream.Collectors;
 
 import static java.util.stream.Collectors.joining;
@@ -196,8 +195,6 @@ public void close() {
         } catch (InterruptedException e) {
             logger.warn("failed to shut down ILM history bulk processor after 10 seconds", e);
             Thread.currentThread().interrupt();
-        } catch (TimeoutException e) {
-            logger.warn("failed to shut down ILM history bulk processor after 10 seconds", e);
         }
     }