private CompletableFuture<Void> addOperation(Operation<K, V, U> operation) {
    if (batch == null) {
      startNewBatch();
    }
    final CompletableFuture<Void> res = batch.addOperation(operation);
    if (batch.isClosed()) {
      processBatch(true);
    }
    return res;
  }
@param batchMetrics Batch metrics.
@param batchHandler Defines how each batch will be processed.
@param batchProvider The batch provider to create a batch instance.
@param clock A clock used to get the timestamp.
@param scheduledExecutorService A scheduled executor service to set timers for the managed batches.
samza-core/src/main/java/org/apache/samza/table/batching/BatchProcessor.java