public static <K, V> CompletableFuture<Map<K, V>> toFutureOfMap(
      Predicate<Throwable> ignoreError, Map<K, CompletableFuture<V>> keyToValueFutures) {
    CompletableFuture<Void> allEntriesFuture =
        CompletableFuture.allOf(keyToValueFutures.values().toArray(new CompletableFuture[]{}));

    return allEntriesFuture.handle((aVoid, t) -> {
      Map<K, V> successfulResults = new HashMap<>();
      for (Map.Entry<K, CompletableFuture<V>> entry : keyToValueFutures.entrySet()) {
        K key = entry.getKey();
        try {
          V value = entry.getValue().join();
          successfulResults.put(key, value);
        } catch (Throwable th) {
          if (ignoreError.test(th)) {
            // else ignore and continue
            LOG.warn("Ignoring value future completion error for key: {}", key, th);
          } else {
            throw th;
          }
        }
      }
      return successfulResults;
    });
  }
Helper method to convert: {@code Map<K, CompletableFuture<V>>}
to:                       {@code CompletableFuture<Map<K, V>>}

Returns a future that completes with successful map entries, skipping any entries with ignored errors,
when all value futures complete.
Returned future completes exceptionally if any of the futures complete with a non-ignored error.
samza-core/src/main/java/org/apache/samza/util/FutureUtil.java