@Override
  public CompletableFuture<Void> addOperation(Operation<K, V, U> operation) {
    Preconditions.checkNotNull(operation);

    if (operation.getArgs() != null && operation.getArgs().length > 0) {
      throw BATCH_NOT_SUPPORTED_EXCEPTION;
    }

    if (operation instanceof GetOperation) {
      queries.putIfAbsent(operation.getKey(), operation);
    } else if (operation instanceof UpdateOperation) {
      throw new BatchingNotSupportedException("Batching not supported for Updates with CompactBatches");
    } else {
      putsDeletes.put(operation.getKey(), operation);
    }
    if (size() >= maxBatchSize) {
      close();
    }
    return completableFuture;
  }
When adding a GetOperation to the batch, mark the GetOperation completed immediately if there is
an PutOperation/DeleteOperation for the key, otherwise, adding the operation to a list.
When adding a Put/DeleteOperation, if there is a put/delete operation for the same key, the existing
operation will be replaced by the new one. If an update operation is received, throw a BatchingNotSupportedException
exception.
samza-core/src/main/java/org/apache/samza/table/batching/CompactBatch.java