class BatchWrite implements Function<InputT, CompletableFuture<Void>> {

  @Override
  public CompletableFuture<Void> apply(InputT input) {
    AtomicReference<CompletableFuture<Void>> output = new AtomicReference<>();
    EncodedT encodedInput = encodeInput(input);
    batches.compute(getBatchKey(input), (batchKey, batch) -> {
      if (batch != null) {
        batch.add(encodedInput).ifPresent(output::set);
      }
      if (output.get() == null) {
        batch = getBatch(batchKey);
        // add first input before allowing batch
        batch.add(encodedInput).ifPresent(output::set);
        // allow batch to complete by timeout now that we have attempted to add the first item
        batch.init.complete(null);
      }
      return batch;
    });

    return Optional.ofNullable(output.get())
        // this is only reachable if getBatch returns a closed batch,
        .orElseThrow(() -> new IllegalArgumentException("Empty batch rejected input"));
  }

  public  BatchWrite(long maxBytes, int maxMessages, Duration maxDelay,
      PubsubMessageToTemplatedString batchKeyTemplate, Executor executor);

}