@Test
  public void testFutureOfMapCompletesExceptionallyIfAnyNonIgnoredErrors() {
    Map<String, CompletableFuture<String>> map = new HashMap<>();
    map.put("1", FutureUtil.failedFuture(new RuntimeException()));
    SamzaException samzaException = new SamzaException();
    map.put("2", FutureUtil.failedFuture(samzaException));

    Predicate<Throwable> mockPredicate = mock(Predicate.class);
    when(mockPredicate.test(any()))
        .thenReturn(true)
        .thenReturn(false);

    CompletableFuture<Map<String, String>> result = FutureUtil.toFutureOfMap(mockPredicate, map);
    assertTrue(result.isDone());
    assertTrue(result.isCompletedExceptionally());
    verify(mockPredicate, times(2)).test(any()); // verify that each failed value future is tested

    try {
      result.join();
      fail("Should have thrown an exception.");
    } catch (Exception e) {
      assertEquals(samzaException, FutureUtil.unwrapExceptions(CompletionException.class, e));
    }
  }
Test all futures in all collections complete before allOf completes.
Test completes exceptionally if any complete exceptionally.
Test works with heterogeneous value types.
Test works with heterogeneous collection types.
Test works with completion stages as well as completable futures.
samza-core/src/test/java/org/apache/samza/util/TestFutureUtil.java