@Override
  public CompletableFuture<Void> putAsync(K key, V value, Object... args) {
    try {
      return batchProcessor.processPutDeleteOrUpdateOperations(new PutOperation<>(key, value, args));
    } catch (BatchingNotSupportedException e) {
      return table.putAsync(key, value, args);
    } catch (Exception e) {
      throw new SamzaException(e);
    }
  }
@param tableId The id of the table.
@param table The target table that serves the batch operations.
@param batchProvider Batch provider to create a batch instance.
@param batchTimerExecutorService Executor service for batch timer.
samza-core/src/main/java/org/apache/samza/table/batching/AsyncBatchingTable.java