private void startNewBatch() {
    batch = batchProvider.getBatch();
    batchOpenTimestamp = clock.nanoTime();
    batchMetrics.incBatchCount();
    setBatchTimer(batch);
  }
@param operation The Put/Delete/Update operation to be added to the batch.
@return A {@link CompletableFuture} to indicate whether the operation is finished.
samza-core/src/main/java/org/apache/samza/table/batching/BatchProcessor.java