private CompletableFuture<?> handleBatchGet(Collection<Operation<K, V, U>> operations) {
    Preconditions.checkNotNull(operations);
    final List<K> gets = getOperationKeys(operations);
    if (gets.isEmpty()) {
      return CompletableFuture.completedFuture(Collections.EMPTY_MAP);
    }

    final Object[] args = getOperationArgs(operations);
    final CompletableFuture<Map<K, V>> getsFuture = args == null ?
        table.getAllAsync(gets) : table.getAllAsync(gets, args);

    getsFuture.whenComplete((map, throwable) -> {
      operations.forEach(operation -> {
        GetOperation<K, V, U> getOperation = (GetOperation<K, V, U>) operation;
        if (throwable != null) {
          getOperation.completeExceptionally(throwable);
        } else {
          getOperation.complete(map.get(operation.getKey()));
        }
      });
    });
    return getsFuture;
  }
Define how batch get should be done.

@param operations The batch get operations.
@return A CompletableFuture to represent the status of the batch operation.
samza-core/src/main/java/org/apache/samza/table/batching/TableBatchHandler.java