default CompletableFuture<Map<K, V>> getAllAsync(Collection<K> keys) {
    Map<K, CompletableFuture<V>> getFutures =  keys.stream().collect(
        Collectors.toMap(k -> k, k -> getAsync(k)));

    return CompletableFuture.allOf(
        Iterables.toArray(getFutures.values(), CompletableFuture.class))
        .thenApply(future ->
          getFutures.entrySet()
            .stream()
            .collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue().join())));
  }
Asynchronously fetch the table {@code records} for specified {@code keys}. This method must be thread-safe.
The default implementation calls getAsync for each key and return a combined future.
@param keys keys for the table records
@return CompletableFuture for the get request
samza-api/src/main/java/org/apache/samza/table/remote/TableReadFunction.java