[utils] improve test coverage on FutureUtils (#969)

Author: gyang94

fluss-common/src/test/java/com/alibaba/fluss/utils/concurrent/FutureUtilsTest.java

@@ -23,18 +23,23 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
+import java.util.concurrent.FutureTask;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static com.alibaba.fluss.testutils.common.FlussAssertions.assertThatFuture;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
 
 /** Tests for the utility methods in {@link FutureUtils}. */
 class FutureUtilsTest {
@@ -153,6 +158,11 @@ void testComposeAfterwardsBothExceptional() {
 
     @Test
     void testCompleteAll() {
+        FutureUtils.ConjunctFuture<Void> emptyConjunctFuture =
+                FutureUtils.completeAll(Collections.emptyList());
+        assertThatFuture(emptyConjunctFuture).isDone();
+        assertThatFuture(emptyConjunctFuture).eventuallySucceeds().isNull();
+
         final CompletableFuture<String> inputFuture1 = new CompletableFuture<>();
         final CompletableFuture<Integer> inputFuture2 = new CompletableFuture<>();
 
@@ -389,4 +399,151 @@ void testOrTimeout() {
                 .withCauseInstanceOf(TimeoutException.class)
                 .withMessageContaining(expectedErrorMessage);
     }
+
+    @Test
+    void testCompletedVoidFuture() {
+        final CompletableFuture<Void> future = FutureUtils.completedVoidFuture();
+        assertThatFuture(future).eventuallySucceeds().isNull();
+    }
+
+    @Test
+    void testCompleteFromCallable() {
+        final CompletableFuture<String> successFuture = new CompletableFuture<>();
+        FutureUtils.completeFromCallable(successFuture, () -> "Fluss");
+        assertThatFuture(successFuture).eventuallySucceeds().isEqualTo("Fluss");
+
+        final CompletableFuture<String> failureFuture = new CompletableFuture<>();
+        FutureUtils.completeFromCallable(
+                failureFuture,
+                () -> {
+                    throw new RuntimeException("mock runtime exception");
+                });
+        assertThatFuture(failureFuture)
+                .eventuallyFailsWith(ExecutionException.class)
+                .withCauseInstanceOf(RuntimeException.class)
+                .withMessageContaining("mock runtime exception");
+    }
+
+    @Test
+    void testRunIfNotDoneAndGet() throws Exception {
+        FutureUtils.runIfNotDoneAndGet(null);
+        List<String> list = new ArrayList<>();
+        Runnable task = () -> list.add("Fluss");
+        FutureTask<String> futureTask = new FutureTask<>(task, "Hello Fluss");
+        FutureUtils.runIfNotDoneAndGet(futureTask);
+        assertThat(futureTask.get()).isEqualTo("Hello Fluss");
+        assertThat(list.contains("Fluss")).isTrue();
+    }
+
+    @Test
+    void combineAll() throws Exception {
+        FutureUtils.ConjunctFuture<Collection<String>> emptyConjunctFuture =
+                FutureUtils.combineAll(Collections.emptyList());
+        assertThatFuture(emptyConjunctFuture).isDone();
+        assertThatFuture(emptyConjunctFuture).eventuallySucceeds().asList().isEmpty();
+
+        CompletableFuture<String> future1 = new CompletableFuture<String>();
+        CompletableFuture<String> future2 = new CompletableFuture<String>();
+        FutureUtils.ConjunctFuture<Collection<String>> conjunctFuture =
+                FutureUtils.combineAll(Arrays.asList(future1, future2));
+
+        assertThatFuture(conjunctFuture).isNotDone();
+        assertThat(conjunctFuture.getNumFuturesTotal()).isEqualTo(2);
+        assertThat(conjunctFuture.getNumFuturesCompleted()).isZero();
+
+        future1.complete("Hello");
+        future2.complete("World");
+        assertThatFuture(conjunctFuture).eventuallySucceeds();
+
+        assertThatFuture(conjunctFuture).isDone();
+        assertThat(conjunctFuture.getNumFuturesTotal()).isEqualTo(2);
+        assertThat(conjunctFuture.getNumFuturesCompleted()).isEqualTo(2);
+        assertThat(conjunctFuture.get()).containsExactly("Hello", "World");
+
+        CompletableFuture<String> successFuture = new CompletableFuture<String>();
+        CompletableFuture<String> failureFuture = new CompletableFuture<String>();
+        FutureUtils.ConjunctFuture<Collection<String>> failureConjunctFuture =
+                FutureUtils.combineAll(Arrays.asList(successFuture, failureFuture));
+
+        failureFuture.completeExceptionally(new RuntimeException("mock runtime exception"));
+
+        assertThatFuture(failureConjunctFuture).isDone();
+        assertThatFuture(failureConjunctFuture)
+                .eventuallyFailsWith(ExecutionException.class)
+                .withCauseInstanceOf(RuntimeException.class)
+                .withMessageContaining("mock runtime exception");
+    }
+
+    @Test
+    void testWaitForAll() {
+        FutureUtils.ConjunctFuture<Void> emptyConjunctFuture =
+                FutureUtils.waitForAll(Collections.emptyList());
+        assertThatFuture(emptyConjunctFuture).isDone();
+        assertThatFuture(emptyConjunctFuture).eventuallySucceeds().isNull();
+
+        List<String> successRes = new ArrayList<>();
+        CompletableFuture<String> future1 = new CompletableFuture<String>();
+        CompletableFuture<String> future2 = new CompletableFuture<String>();
+        FutureUtils.ConjunctFuture<Void> conjunctFuture =
+                FutureUtils.waitForAll(
+                        Arrays.asList(future1, future2),
+                        (val, throwable) -> {
+                            if (throwable == null) {
+                                successRes.add(val);
+                            } else {
+                                throw new RuntimeException(throwable);
+                            }
+                        });
+
+        assertThatFuture(conjunctFuture).isNotDone();
+        assertThat(conjunctFuture.getNumFuturesTotal()).isEqualTo(2);
+        assertThat(conjunctFuture.getNumFuturesCompleted()).isZero();
+
+        future1.complete("Hello");
+        future2.complete("World");
+        assertThatFuture(conjunctFuture).eventuallySucceeds();
+        assertThatFuture(conjunctFuture).isDone();
+        assertThat(conjunctFuture.getNumFuturesTotal()).isEqualTo(2);
+        assertThat(conjunctFuture.getNumFuturesCompleted()).isEqualTo(2);
+
+        assertThat(successRes.size()).isEqualTo(2);
+        assertThat(successRes.contains("Hello")).isTrue();
+        assertThat(successRes.contains("World")).isTrue();
+
+        List<String> failureRes = new ArrayList<>();
+        CompletableFuture<String> successFuture = new CompletableFuture<String>();
+        CompletableFuture<String> failureFuture = new CompletableFuture<String>();
+        FutureUtils.ConjunctFuture<Void> failureConjunctFuture =
+                FutureUtils.waitForAll(
+                        Arrays.asList(successFuture, failureFuture),
+                        (val, throwable) -> {
+                            if (throwable == null) {
+                                failureRes.add(val);
+                            } else {
+                                throw new RuntimeException(throwable);
+                            }
+                        });
+
+        failureFuture.completeExceptionally(new RuntimeException("mock runtime exception"));
+
+        assertThatFuture(failureConjunctFuture).isDone();
+        assertThat(conjunctFuture.getNumFuturesTotal()).isEqualTo(2);
+        assertThat(conjunctFuture.getNumFuturesCompleted()).isEqualTo(2);
+        assertThatFuture(failureConjunctFuture)
+                .eventuallyFailsWith(ExecutionException.class)
+                .withCauseInstanceOf(RuntimeException.class)
+                .withMessageContaining("mock runtime exception");
+        assertThat(failureRes.isEmpty()).isTrue();
+    }
+
+    @Test
+    void testCatchingAndLoggingThrowables() {
+        Runnable task =
+                () -> {
+                    throw new RuntimeException("mock runtime exception");
+                };
+
+        Runnable catchingRunnable = FutureUtils.catchingAndLoggingThrowables(task);
+        assertThatNoException().isThrownBy(catchingRunnable::run);
+    }
 }

fluss-test-coverage/pom.xml

@@ -249,9 +249,6 @@
                                         <exclude>com.alibaba.fluss.memory.*</exclude>
                                         <exclude>com.alibaba.fluss.utils.*</exclude>
                                         <exclude>com.alibaba.fluss.exception.*</exclude>
-                                        <exclude>
-                                            com.alibaba.fluss.utils.concurrent.FutureUtils.Timeout
-                                        </exclude>
                                         <exclude>com.alibaba.fluss.row.arrow.*</exclude>
                                         <exclude>
                                             com.alibaba.fluss.row.columnar.BytesColumnVector.Bytes