Fix: execute can ignore timeout and block indefinitely

CHANGELOG.md

@@ -15,6 +15,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
  - `TopicUpdateTransaction.[get|set]ExpirationTime()`
  - `CustomFee.[set|get]AllCollectorsAreExempt()`
 
+### Fixed
+ - Execute with a timeout can ignore timeout and block indefinitely in CI tests
+
 ## 2.17.4
 
 ### Added

sdk/src/main/java/com/hedera/hashgraph/sdk/Client.java

@@ -20,11 +20,11 @@
 package com.hedera.hashgraph.sdk;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.gson.Gson;
 import com.google.gson.JsonElement;
 import com.google.gson.JsonParseException;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
-import java8.util.Lists;
 import java8.util.concurrent.CompletableFuture;
 import java8.util.concurrent.CompletionStage;
 import java8.util.function.BiConsumer;
@@ -48,10 +48,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ForkJoinPool;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.*;
 import java.util.function.Supplier;
 
 import static com.hedera.hashgraph.sdk.BaseNodeAddress.PORT_NODE_PLAIN;
@@ -137,7 +134,14 @@ public final class Client implements AutoCloseable {
      * @return                          the executor service
      */
     static ExecutorService createExecutor() {
-        return ForkJoinPool.commonPool();
+        var threadFactory = new ThreadFactoryBuilder()
+            .setNameFormat("hedera-sdk-%d")
+            .setDaemon(true)
+            .build();
+
+        return Executors.newFixedThreadPool(
+            Runtime.getRuntime().availableProcessors(),
+            threadFactory);
     }
 
     /**
@@ -1289,6 +1293,20 @@ public synchronized void close(Duration timeout) throws TimeoutException {
         var networkError = network.awaitClose(closeDeadline, null);
         var mirrorNetworkError = mirrorNetwork.awaitClose(closeDeadline, networkError);
 
+        // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html
+        try {
+            executor.shutdown();
+            if (!executor.awaitTermination(timeout.getSeconds() / 2, TimeUnit.SECONDS)) {
+                executor.shutdownNow();
+                if (!executor.awaitTermination(timeout.getSeconds() / 2, TimeUnit.SECONDS)) {
+                    logger.warn("Pool did not terminate");
+                }
+            }
+        } catch (InterruptedException ex) {
+            executor.shutdownNow();
+            Thread.currentThread().interrupt();
+        }
+
         if (mirrorNetworkError != null) {
             if (mirrorNetworkError instanceof TimeoutException) {
                 throw (TimeoutException) mirrorNetworkError;

sdk/src/test/java/com/hedera/hashgraph/sdk/ClientCloseTest.java

@@ -1,24 +1,15 @@
 package com.hedera.hashgraph.sdk;
 
-import org.bouncycastle.asn1.x509.Time;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.mockito.MockSettings;
+import org.threeten.bp.Duration;
 
 import java.util.Collections;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.*;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatException;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
-import static org.assertj.core.api.Assertions.in;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
 
 public class ClientCloseTest {
     @Test
@@ -70,4 +61,57 @@ void closeHandlesMirrorNetworkInterrupted() {
         assertThatExceptionOfType(RuntimeException.class).isThrownBy(client::close).withCause(interruptedException);
         assertThat(network.hasShutDownNow).isFalse();
     }
+
+    @Test
+    void closeHandlesExecutorShutdown() throws TimeoutException {
+        var executor = Client.createExecutor();
+        var network = Network.forNetwork(executor, Collections.emptyMap());
+        var mirrorNetwork = MirrorNetwork.forNetwork(executor, Collections.emptyList());
+        var client = new Client(executor, network, mirrorNetwork, null, null);
+
+        client.close();
+        assertThat(executor.isShutdown()).isTrue();
+    }
+
+    @Test
+    void closeHandlesExecutorTerminatingInTime() throws InterruptedException, TimeoutException {
+        var duration = Duration.ofSeconds(30);
+        var executor = mock(ThreadPoolExecutor.class);
+        var network = Network.forNetwork(executor, Collections.emptyMap());
+        var mirrorNetwork = MirrorNetwork.forNetwork(executor, Collections.emptyList());
+        var client = new Client(executor, network, mirrorNetwork, null, null);
+
+        doReturn(true).when(executor).awaitTermination(30 / 2, TimeUnit.SECONDS);
+
+        client.close(duration);
+        verify(executor, times(0)).shutdownNow();
+    }
+
+    @Test
+    void closeHandlesExecutorNotTerminatingInTime() throws InterruptedException, TimeoutException {
+        var duration = Duration.ofSeconds(30);
+        var executor = mock(ThreadPoolExecutor.class);
+        var network = Network.forNetwork(executor, Collections.emptyMap());
+        var mirrorNetwork = MirrorNetwork.forNetwork(executor, Collections.emptyList());
+        var client = new Client(executor, network, mirrorNetwork, null, null);
+
+        doReturn(false).when(executor).awaitTermination(30 / 2, TimeUnit.SECONDS);
+
+        client.close(duration);
+        verify(executor, times(1)).shutdownNow();
+    }
+
+    @Test
+    void closeHandlesExecutorWhenThreadIsInterrupted() throws InterruptedException, TimeoutException {
+        var duration = Duration.ofSeconds(30);
+        var executor = mock(ThreadPoolExecutor.class);
+        var network = Network.forNetwork(executor, Collections.emptyMap());
+        var mirrorNetwork = MirrorNetwork.forNetwork(executor, Collections.emptyList());
+        var client = new Client(executor, network, mirrorNetwork, null, null);
+
+        doThrow(new InterruptedException()).when(executor).awaitTermination(30 / 2, TimeUnit.SECONDS);
+
+        client.close(duration);
+        verify(executor, times(1)).shutdownNow();
+    }
 }