| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF 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.apache.kafka.common; |
| |
| import org.apache.kafka.common.internals.KafkaFutureImpl; |
| |
| import org.junit.jupiter.api.Test; |
| import org.junit.jupiter.api.Timeout; |
| |
| import java.lang.reflect.InvocationTargetException; |
| import java.lang.reflect.Method; |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.concurrent.CancellationException; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.CompletionException; |
| import java.util.concurrent.CompletionStage; |
| import java.util.concurrent.ExecutionException; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.TimeoutException; |
| import java.util.function.Supplier; |
| |
| import static org.junit.jupiter.api.Assertions.assertEquals; |
| import static org.junit.jupiter.api.Assertions.assertFalse; |
| import static org.junit.jupiter.api.Assertions.assertInstanceOf; |
| import static org.junit.jupiter.api.Assertions.assertNotEquals; |
| import static org.junit.jupiter.api.Assertions.assertNull; |
| import static org.junit.jupiter.api.Assertions.assertThrows; |
| import static org.junit.jupiter.api.Assertions.assertTrue; |
| |
| /** |
| * A unit test for KafkaFuture. |
| */ |
| @Timeout(120) |
| public class KafkaFutureTest { |
| |
| /** Asserts that the given future is done, didn't fail and wasn't cancelled. */ |
| private void assertIsSuccessful(KafkaFuture<?> future) { |
| assertTrue(future.isDone()); |
| assertFalse(future.isCompletedExceptionally()); |
| assertFalse(future.isCancelled()); |
| } |
| |
| /** Asserts that the given future is done, failed and wasn't cancelled. */ |
| private void assertIsFailed(KafkaFuture<?> future) { |
| assertTrue(future.isDone()); |
| assertFalse(future.isCancelled()); |
| assertTrue(future.isCompletedExceptionally()); |
| } |
| |
| /** Asserts that the given future is done, didn't fail and was cancelled. */ |
| private void assertIsCancelled(KafkaFuture<?> future) { |
| assertTrue(future.isDone()); |
| assertTrue(future.isCancelled()); |
| assertTrue(future.isCompletedExceptionally()); |
| assertThrows(CancellationException.class, () -> future.getNow(null)); |
| assertThrows(CancellationException.class, () -> future.get(0, TimeUnit.MILLISECONDS)); |
| } |
| |
| private <T> void awaitAndAssertResult(KafkaFuture<T> future, |
| T expectedResult, |
| T alternativeValue) { |
| assertNotEquals(expectedResult, alternativeValue); |
| try { |
| assertEquals(expectedResult, future.get(5, TimeUnit.MINUTES)); |
| } catch (Exception e) { |
| throw new AssertionError("Unexpected exception", e); |
| } |
| try { |
| assertEquals(expectedResult, future.get()); |
| } catch (Exception e) { |
| throw new AssertionError("Unexpected exception", e); |
| } |
| try { |
| assertEquals(expectedResult, future.getNow(alternativeValue)); |
| } catch (Exception e) { |
| throw new AssertionError("Unexpected exception", e); |
| } |
| } |
| |
| private Throwable awaitAndAssertFailure(KafkaFuture<?> future, |
| Class<? extends Throwable> expectedException, |
| String expectedMessage) { |
| ExecutionException executionException = assertThrows(ExecutionException.class, () -> future.get(5, TimeUnit.MINUTES)); |
| assertEquals(expectedException, executionException.getCause().getClass()); |
| assertEquals(expectedMessage, executionException.getCause().getMessage()); |
| |
| executionException = assertThrows(ExecutionException.class, future::get); |
| assertEquals(expectedException, executionException.getCause().getClass()); |
| assertEquals(expectedMessage, executionException.getCause().getMessage()); |
| |
| executionException = assertThrows(ExecutionException.class, () -> future.getNow(null)); |
| assertEquals(expectedException, executionException.getCause().getClass()); |
| assertEquals(expectedMessage, executionException.getCause().getMessage()); |
| return executionException.getCause(); |
| } |
| |
| private void awaitAndAssertCancelled(KafkaFuture<?> future, String expectedMessage) { |
| CancellationException cancellationException = assertThrows(CancellationException.class, () -> future.get(5, TimeUnit.MINUTES)); |
| assertEquals(expectedMessage, cancellationException.getMessage()); |
| assertEquals(CancellationException.class, cancellationException.getClass()); |
| |
| cancellationException = assertThrows(CancellationException.class, future::get); |
| assertEquals(expectedMessage, cancellationException.getMessage()); |
| assertEquals(CancellationException.class, cancellationException.getClass()); |
| |
| cancellationException = assertThrows(CancellationException.class, () -> future.getNow(null)); |
| assertEquals(expectedMessage, cancellationException.getMessage()); |
| assertEquals(CancellationException.class, cancellationException.getClass()); |
| } |
| |
| private Object invokeOrThrow(final Method method, final Object obj, final Object... args) throws Throwable { |
| try { |
| return method.invoke(obj, args); |
| } catch (InvocationTargetException e) { |
| throw e.getCause(); |
| } |
| } |
| |
| @Test |
| public void testCompleteFutures() throws Exception { |
| KafkaFutureImpl<Integer> future123 = new KafkaFutureImpl<>(); |
| assertTrue(future123.complete(123)); |
| assertFalse(future123.complete(456)); |
| assertFalse(future123.cancel(true)); |
| assertEquals(Integer.valueOf(123), future123.get()); |
| assertIsSuccessful(future123); |
| |
| KafkaFuture<Integer> future456 = KafkaFuture.completedFuture(456); |
| assertFalse(future456.complete(789)); |
| assertFalse(future456.cancel(true)); |
| assertEquals(Integer.valueOf(456), future456.get()); |
| assertIsSuccessful(future456); |
| } |
| |
| @Test |
| public void testCompleteFuturesExceptionally() { |
| KafkaFutureImpl<Integer> futureFail = new KafkaFutureImpl<>(); |
| assertTrue(futureFail.completeExceptionally(new RuntimeException("We require more vespene gas"))); |
| assertIsFailed(futureFail); |
| assertFalse(futureFail.completeExceptionally(new RuntimeException("We require more minerals"))); |
| assertFalse(futureFail.cancel(true)); |
| |
| ExecutionException executionException = assertThrows(ExecutionException.class, futureFail::get); |
| assertEquals(RuntimeException.class, executionException.getCause().getClass()); |
| assertEquals("We require more vespene gas", executionException.getCause().getMessage()); |
| |
| KafkaFutureImpl<Integer> tricky1 = new KafkaFutureImpl<>(); |
| assertTrue(tricky1.completeExceptionally(new CompletionException(new CancellationException()))); |
| assertIsFailed(tricky1); |
| awaitAndAssertFailure(tricky1, CompletionException.class, "java.util.concurrent.CancellationException"); |
| } |
| |
| @Test |
| public void testCompleteFuturesViaCancellation() { |
| KafkaFutureImpl<Integer> viaCancel = new KafkaFutureImpl<>(); |
| assertTrue(viaCancel.cancel(true)); |
| assertIsCancelled(viaCancel); |
| awaitAndAssertCancelled(viaCancel, null); |
| |
| KafkaFutureImpl<Integer> viaCancellationException = new KafkaFutureImpl<>(); |
| assertTrue(viaCancellationException.completeExceptionally(new CancellationException("We require more vespene gas"))); |
| assertIsCancelled(viaCancellationException); |
| awaitAndAssertCancelled(viaCancellationException, "We require more vespene gas"); |
| } |
| |
| @Test |
| public void testToString() { |
| KafkaFutureImpl<Integer> success = new KafkaFutureImpl<>(); |
| assertEquals("KafkaFuture{value=null,exception=null,done=false}", success.toString()); |
| success.complete(12); |
| assertEquals("KafkaFuture{value=12,exception=null,done=true}", success.toString()); |
| |
| KafkaFutureImpl<Integer> failure = new KafkaFutureImpl<>(); |
| failure.completeExceptionally(new RuntimeException("foo")); |
| assertEquals("KafkaFuture{value=null,exception=java.lang.RuntimeException: foo,done=true}", failure.toString()); |
| |
| KafkaFutureImpl<Integer> tricky1 = new KafkaFutureImpl<>(); |
| tricky1.completeExceptionally(new CompletionException(new CancellationException())); |
| assertEquals("KafkaFuture{value=null,exception=java.util.concurrent.CompletionException: java.util.concurrent.CancellationException,done=true}", tricky1.toString()); |
| |
| KafkaFutureImpl<Integer> cancelled = new KafkaFutureImpl<>(); |
| cancelled.cancel(true); |
| assertEquals("KafkaFuture{value=null,exception=java.util.concurrent.CancellationException,done=true}", cancelled.toString()); |
| } |
| |
| @Test |
| public void testCompletingFutures() throws Exception { |
| final KafkaFutureImpl<String> future = new KafkaFutureImpl<>(); |
| CompleterThread<String> myThread = new CompleterThread<>(future, "You must construct additional pylons."); |
| assertIsNotCompleted(future); |
| assertEquals("I am ready", future.getNow("I am ready")); |
| myThread.start(); |
| awaitAndAssertResult(future, "You must construct additional pylons.", "I am ready"); |
| assertIsSuccessful(future); |
| myThread.join(); |
| assertNull(myThread.testException); |
| } |
| |
| @Test |
| public void testCompletingFuturesExceptionally() throws Exception { |
| final KafkaFutureImpl<String> future = new KafkaFutureImpl<>(); |
| CompleterThread<String> myThread = new CompleterThread<>(future, null, |
| new RuntimeException("Ultimate efficiency achieved.")); |
| assertIsNotCompleted(future); |
| assertEquals("I am ready", future.getNow("I am ready")); |
| myThread.start(); |
| awaitAndAssertFailure(future, RuntimeException.class, "Ultimate efficiency achieved."); |
| assertIsFailed(future); |
| myThread.join(); |
| assertNull(myThread.testException); |
| } |
| |
| @Test |
| public void testCompletingFuturesViaCancellation() throws Exception { |
| final KafkaFutureImpl<String> future = new KafkaFutureImpl<>(); |
| CompleterThread<String> myThread = new CompleterThread<>(future, null, |
| new CancellationException("Ultimate efficiency achieved.")); |
| assertIsNotCompleted(future); |
| assertEquals("I am ready", future.getNow("I am ready")); |
| myThread.start(); |
| awaitAndAssertCancelled(future, "Ultimate efficiency achieved."); |
| assertIsCancelled(future); |
| myThread.join(); |
| assertNull(myThread.testException); |
| } |
| |
| private void assertIsNotCompleted(KafkaFutureImpl<String> future) { |
| assertFalse(future.isDone()); |
| assertFalse(future.isCompletedExceptionally()); |
| assertFalse(future.isCancelled()); |
| } |
| |
| @Test |
| public void testThenApplyOnSucceededFuture() throws Exception { |
| KafkaFutureImpl<Integer> future = new KafkaFutureImpl<>(); |
| KafkaFuture<Integer> doubledFuture = future.thenApply(integer -> 2 * integer); |
| assertFalse(doubledFuture.isDone()); |
| KafkaFuture<Integer> tripledFuture = future.thenApply(integer -> 3 * integer); |
| assertFalse(tripledFuture.isDone()); |
| future.complete(21); |
| assertEquals(Integer.valueOf(21), future.getNow(-1)); |
| assertEquals(Integer.valueOf(42), doubledFuture.getNow(-1)); |
| assertEquals(Integer.valueOf(63), tripledFuture.getNow(-1)); |
| KafkaFuture<Integer> quadrupledFuture = future.thenApply(integer -> 4 * integer); |
| assertEquals(Integer.valueOf(84), quadrupledFuture.getNow(-1)); |
| } |
| |
| @Test |
| public void testThenApplyOnFailedFuture() { |
| KafkaFutureImpl<Integer> future = new KafkaFutureImpl<>(); |
| KafkaFuture<Integer> dependantFuture = future.thenApply(integer -> 2 * integer); |
| future.completeExceptionally(new RuntimeException("We require more vespene gas")); |
| assertIsFailed(future); |
| assertIsFailed(dependantFuture); |
| awaitAndAssertFailure(future, RuntimeException.class, "We require more vespene gas"); |
| awaitAndAssertFailure(dependantFuture, RuntimeException.class, "We require more vespene gas"); |
| } |
| |
| @Test |
| public void testThenApplyOnFailedFutureTricky() { |
| KafkaFutureImpl<Integer> future = new KafkaFutureImpl<>(); |
| KafkaFuture<Integer> dependantFuture = future.thenApply(integer -> 2 * integer); |
| future.completeExceptionally(new CompletionException(new RuntimeException("We require more vespene gas"))); |
| assertIsFailed(future); |
| assertIsFailed(dependantFuture); |
| awaitAndAssertFailure(future, CompletionException.class, "java.lang.RuntimeException: We require more vespene gas"); |
| awaitAndAssertFailure(dependantFuture, CompletionException.class, "java.lang.RuntimeException: We require more vespene gas"); |
| } |
| |
| @Test |
| public void testThenApplyOnFailedFutureTricky2() { |
| KafkaFutureImpl<Integer> future = new KafkaFutureImpl<>(); |
| KafkaFuture<Integer> dependantFuture = future.thenApply(integer -> 2 * integer); |
| future.completeExceptionally(new CompletionException(new CancellationException())); |
| assertIsFailed(future); |
| assertIsFailed(dependantFuture); |
| awaitAndAssertFailure(future, CompletionException.class, "java.util.concurrent.CancellationException"); |
| awaitAndAssertFailure(dependantFuture, CompletionException.class, "java.util.concurrent.CancellationException"); |
| } |
| |
| @Test |
| public void testThenApplyOnSucceededFutureAndFunctionThrows() { |
| KafkaFutureImpl<Integer> future = new KafkaFutureImpl<>(); |
| KafkaFuture<Integer> dependantFuture = future.thenApply(integer -> { |
| throw new RuntimeException("We require more vespene gas"); |
| }); |
| future.complete(21); |
| assertIsSuccessful(future); |
| assertIsFailed(dependantFuture); |
| awaitAndAssertResult(future, 21, null); |
| awaitAndAssertFailure(dependantFuture, RuntimeException.class, "We require more vespene gas"); |
| } |
| |
| @Test |
| public void testThenApplyOnSucceededFutureAndFunctionThrowsCompletionException() { |
| KafkaFutureImpl<Integer> future = new KafkaFutureImpl<>(); |
| KafkaFuture<Integer> dependantFuture = future.thenApply(integer -> { |
| throw new CompletionException(new RuntimeException("We require more vespene gas")); |
| }); |
| future.complete(21); |
| assertIsSuccessful(future); |
| assertIsFailed(dependantFuture); |
| awaitAndAssertResult(future, 21, null); |
| Throwable cause = awaitAndAssertFailure(dependantFuture, CompletionException.class, "java.lang.RuntimeException: We require more vespene gas"); |
| assertInstanceOf(RuntimeException.class, cause.getCause()); |
| assertEquals(cause.getCause().getMessage(), "We require more vespene gas"); |
| } |
| |
| @Test |
| public void testThenApplyOnFailedFutureFunctionNotCalled() { |
| KafkaFutureImpl<Integer> future = new KafkaFutureImpl<>(); |
| boolean[] ran = {false}; |
| KafkaFuture<Integer> dependantFuture = future.thenApply(integer -> { |
| // Because the top level future failed, this should never be called. |
| ran[0] = true; |
| return null; |
| }); |
| future.completeExceptionally(new RuntimeException("We require more minerals")); |
| assertIsFailed(future); |
| assertIsFailed(dependantFuture); |
| awaitAndAssertFailure(future, RuntimeException.class, "We require more minerals"); |
| awaitAndAssertFailure(dependantFuture, RuntimeException.class, "We require more minerals"); |
| assertFalse(ran[0]); |
| } |
| |
| @Test |
| public void testThenApplyOnCancelledFuture() { |
| KafkaFutureImpl<Integer> future = new KafkaFutureImpl<>(); |
| KafkaFuture<Integer> dependantFuture = future.thenApply(integer -> 2 * integer); |
| future.cancel(true); |
| assertIsCancelled(future); |
| assertIsCancelled(dependantFuture); |
| awaitAndAssertCancelled(future, null); |
| awaitAndAssertCancelled(dependantFuture, null); |
| } |
| |
| @Test |
| public void testWhenCompleteOnSucceededFuture() throws Throwable { |
| KafkaFutureImpl<Integer> future = new KafkaFutureImpl<>(); |
| Throwable[] err = new Throwable[1]; |
| boolean[] ran = {false}; |
| KafkaFuture<Integer> dependantFuture = future.whenComplete((integer, ex) -> { |
| ran[0] = true; |
| try { |
| assertEquals(Integer.valueOf(21), integer); |
| if (ex != null) { |
| throw ex; |
| } |
| } catch (Throwable e) { |
| err[0] = e; |
| } |
| }); |
| assertFalse(dependantFuture.isDone()); |
| assertTrue(future.complete(21)); |
| assertTrue(ran[0]); |
| if (err[0] != null) { |
| throw err[0]; |
| } |
| } |
| |
| @Test |
| public void testWhenCompleteOnFailedFuture() { |
| KafkaFutureImpl<Integer> future = new KafkaFutureImpl<>(); |
| Throwable[] err = new Throwable[1]; |
| boolean[] ran = {false}; |
| KafkaFuture<Integer> dependantFuture = future.whenComplete((integer, ex) -> { |
| ran[0] = true; |
| err[0] = ex; |
| if (integer != null) { |
| err[0] = new AssertionError(); |
| } |
| }); |
| assertFalse(dependantFuture.isDone()); |
| RuntimeException ex = new RuntimeException("We require more vespene gas"); |
| assertTrue(future.completeExceptionally(ex)); |
| assertTrue(ran[0]); |
| assertEquals(err[0], ex); |
| } |
| |
| @Test |
| public void testWhenCompleteOnSucceededFutureAndConsumerThrows() { |
| KafkaFutureImpl<Integer> future = new KafkaFutureImpl<>(); |
| boolean[] ran = {false}; |
| KafkaFuture<Integer> dependantFuture = future.whenComplete((integer, ex) -> { |
| ran[0] = true; |
| throw new RuntimeException("We require more minerals"); |
| }); |
| assertFalse(dependantFuture.isDone()); |
| assertTrue(future.complete(21)); |
| assertIsSuccessful(future); |
| assertTrue(ran[0]); |
| assertIsFailed(dependantFuture); |
| awaitAndAssertFailure(dependantFuture, RuntimeException.class, "We require more minerals"); |
| } |
| |
| @Test |
| public void testWhenCompleteOnFailedFutureAndConsumerThrows() { |
| KafkaFutureImpl<Integer> future = new KafkaFutureImpl<>(); |
| boolean[] ran = {false}; |
| KafkaFuture<Integer> dependantFuture = future.whenComplete((integer, ex) -> { |
| ran[0] = true; |
| throw new RuntimeException("We require more minerals"); |
| }); |
| assertFalse(dependantFuture.isDone()); |
| assertTrue(future.completeExceptionally(new RuntimeException("We require more vespene gas"))); |
| assertIsFailed(future); |
| assertTrue(ran[0]); |
| assertIsFailed(dependantFuture); |
| awaitAndAssertFailure(dependantFuture, RuntimeException.class, "We require more vespene gas"); |
| } |
| |
| @Test |
| public void testWhenCompleteOnCancelledFuture() { |
| KafkaFutureImpl<Integer> future = new KafkaFutureImpl<>(); |
| Throwable[] err = new Throwable[1]; |
| boolean[] ran = {false}; |
| KafkaFuture<Integer> dependantFuture = future.whenComplete((integer, ex) -> { |
| ran[0] = true; |
| err[0] = ex; |
| if (integer != null) { |
| err[0] = new AssertionError(); |
| } |
| }); |
| assertFalse(dependantFuture.isDone()); |
| assertTrue(future.cancel(true)); |
| assertTrue(ran[0]); |
| assertInstanceOf(CancellationException.class, err[0]); |
| } |
| |
| private static class CompleterThread<T> extends Thread { |
| |
| private final KafkaFutureImpl<T> future; |
| private final T value; |
| private final Throwable exception; |
| Throwable testException = null; |
| |
| CompleterThread(KafkaFutureImpl<T> future, T value) { |
| this.future = future; |
| this.value = value; |
| this.exception = null; |
| } |
| |
| CompleterThread(KafkaFutureImpl<T> future, T value, Exception exception) { |
| this.future = future; |
| this.value = value; |
| this.exception = exception; |
| } |
| |
| @Override |
| public void run() { |
| try { |
| try { |
| Thread.sleep(0, 200); |
| } catch (InterruptedException e) { |
| } |
| if (exception == null) { |
| future.complete(value); |
| } else { |
| future.completeExceptionally(exception); |
| } |
| } catch (Throwable testException) { |
| this.testException = testException; |
| } |
| } |
| } |
| |
| private static class WaiterThread<T> extends Thread { |
| |
| private final KafkaFutureImpl<T> future; |
| private final T expected; |
| Throwable testException = null; |
| |
| WaiterThread(KafkaFutureImpl<T> future, T expected) { |
| this.future = future; |
| this.expected = expected; |
| } |
| |
| @Override |
| public void run() { |
| try { |
| T value = future.get(); |
| assertEquals(expected, value); |
| } catch (Throwable testException) { |
| this.testException = testException; |
| } |
| } |
| } |
| |
| @Test |
| public void testAllOfFutures() throws Exception { |
| final int numThreads = 5; |
| final List<KafkaFutureImpl<Integer>> futures = new ArrayList<>(); |
| for (int i = 0; i < numThreads; i++) { |
| futures.add(new KafkaFutureImpl<>()); |
| } |
| KafkaFuture<Void> allFuture = KafkaFuture.allOf(futures.toArray(new KafkaFuture[0])); |
| final List<CompleterThread<Integer>> completerThreads = new ArrayList<>(); |
| final List<WaiterThread<Integer>> waiterThreads = new ArrayList<>(); |
| for (int i = 0; i < numThreads; i++) { |
| completerThreads.add(new CompleterThread<>(futures.get(i), i)); |
| waiterThreads.add(new WaiterThread<>(futures.get(i), i)); |
| } |
| assertFalse(allFuture.isDone()); |
| for (int i = 0; i < numThreads; i++) { |
| waiterThreads.get(i).start(); |
| } |
| for (int i = 0; i < numThreads - 1; i++) { |
| completerThreads.get(i).start(); |
| } |
| assertFalse(allFuture.isDone()); |
| completerThreads.get(numThreads - 1).start(); |
| allFuture.get(); |
| assertIsSuccessful(allFuture); |
| for (int i = 0; i < numThreads; i++) { |
| assertEquals(Integer.valueOf(i), futures.get(i).get()); |
| } |
| for (int i = 0; i < numThreads; i++) { |
| completerThreads.get(i).join(); |
| waiterThreads.get(i).join(); |
| assertNull(completerThreads.get(i).testException); |
| assertNull(waiterThreads.get(i).testException); |
| } |
| } |
| |
| @Test |
| public void testAllOfFuturesWithFailure() throws Exception { |
| final int numThreads = 5; |
| final List<KafkaFutureImpl<Integer>> futures = new ArrayList<>(); |
| for (int i = 0; i < numThreads; i++) { |
| futures.add(new KafkaFutureImpl<>()); |
| } |
| KafkaFuture<Void> allFuture = KafkaFuture.allOf(futures.toArray(new KafkaFuture[0])); |
| final List<CompleterThread<Integer>> completerThreads = new ArrayList<>(); |
| final List<WaiterThread<Integer>> waiterThreads = new ArrayList<>(); |
| int lastIndex = numThreads - 1; |
| for (int i = 0; i < lastIndex; i++) { |
| completerThreads.add(new CompleterThread<>(futures.get(i), i)); |
| waiterThreads.add(new WaiterThread<>(futures.get(i), i)); |
| } |
| completerThreads.add(new CompleterThread<>(futures.get(lastIndex), null, new RuntimeException("Last one failed"))); |
| waiterThreads.add(new WaiterThread<>(futures.get(lastIndex), lastIndex)); |
| assertFalse(allFuture.isDone()); |
| for (int i = 0; i < numThreads; i++) { |
| waiterThreads.get(i).start(); |
| } |
| for (int i = 0; i < lastIndex; i++) { |
| completerThreads.get(i).start(); |
| } |
| assertFalse(allFuture.isDone()); |
| completerThreads.get(lastIndex).start(); |
| awaitAndAssertFailure(allFuture, RuntimeException.class, "Last one failed"); |
| assertIsFailed(allFuture); |
| for (int i = 0; i < lastIndex; i++) { |
| assertEquals(Integer.valueOf(i), futures.get(i).get()); |
| } |
| assertIsFailed(futures.get(lastIndex)); |
| for (int i = 0; i < numThreads; i++) { |
| completerThreads.get(i).join(); |
| waiterThreads.get(i).join(); |
| assertNull(completerThreads.get(i).testException); |
| if (i == lastIndex) { |
| assertEquals(ExecutionException.class, waiterThreads.get(i).testException.getClass()); |
| assertEquals(RuntimeException.class, waiterThreads.get(i).testException.getCause().getClass()); |
| assertEquals("Last one failed", waiterThreads.get(i).testException.getCause().getMessage()); |
| } else { |
| assertNull(waiterThreads.get(i).testException); |
| } |
| } |
| } |
| |
| @Test |
| public void testAllOfFuturesHandlesZeroFutures() throws Exception { |
| KafkaFuture<Void> allFuture = KafkaFuture.allOf(); |
| assertTrue(allFuture.isDone()); |
| assertFalse(allFuture.isCancelled()); |
| assertFalse(allFuture.isCompletedExceptionally()); |
| allFuture.get(); |
| } |
| |
| @Test |
| public void testFutureTimeoutWithZeroWait() { |
| final KafkaFutureImpl<String> future = new KafkaFutureImpl<>(); |
| assertThrows(TimeoutException.class, () -> future.get(0, TimeUnit.MILLISECONDS)); |
| } |
| |
| @Test |
| @SuppressWarnings("unchecked") |
| public void testLeakCompletableFuture() throws Throwable { |
| final KafkaFutureImpl<String> kfut = new KafkaFutureImpl<>(); |
| CompletableFuture<String> comfut = kfut.toCompletionStage().toCompletableFuture(); |
| assertThrows(UnsupportedOperationException.class, () -> comfut.complete("")); |
| assertThrows(UnsupportedOperationException.class, () -> comfut.completeExceptionally(new RuntimeException())); |
| Method completeOnTimeout = CompletableFuture.class.getDeclaredMethod("completeOnTimeout", Object.class, Long.TYPE, TimeUnit.class); |
| assertThrows(UnsupportedOperationException.class, () -> invokeOrThrow(completeOnTimeout, comfut, "", 1L, TimeUnit.MILLISECONDS)); |
| |
| Method completeAsync = CompletableFuture.class.getDeclaredMethod("completeAsync", Supplier.class); |
| assertThrows(UnsupportedOperationException.class, () -> invokeOrThrow(completeAsync, comfut, (Supplier<String>) () -> "")); |
| |
| Method obtrudeValue = CompletableFuture.class.getDeclaredMethod("obtrudeValue", Object.class); |
| assertThrows(UnsupportedOperationException.class, () -> invokeOrThrow(obtrudeValue, comfut, "")); |
| |
| Method obtrudeException = CompletableFuture.class.getDeclaredMethod("obtrudeException", Throwable.class); |
| assertThrows(UnsupportedOperationException.class, () -> invokeOrThrow(obtrudeException, comfut, new RuntimeException())); |
| |
| // Check the CF from a minimal CompletionStage doesn't cause completion of the original KafkaFuture |
| Method minimal = CompletableFuture.class.getDeclaredMethod("minimalCompletionStage"); |
| CompletionStage<String> cs = (CompletionStage<String>) invokeOrThrow(minimal, comfut); |
| cs.toCompletableFuture().complete(""); |
| |
| assertFalse(kfut.isDone()); |
| assertFalse(comfut.isDone()); |
| } |
| |
| } |
