@Override
  public synchronized void flush(String source) {
    long incompleteSends = pendingFutures.stream().filter(x -> !x.isDone()).count();
    LOG.info("Trying to flush pending {} sends.", incompleteSends);
    checkForSendCallbackErrors("Received exception on message send.");
    CompletableFuture<Void> future =
        CompletableFuture.allOf(pendingFutures.toArray(new CompletableFuture[pendingFutures.size()]));

    try {
      // Block until all the pending sends are complete or timeout.
      future.get(DEFAULT_FLUSH_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
    } catch (InterruptedException | ExecutionException | TimeoutException e) {
      incompleteSends = pendingFutures.stream().filter(x -> !x.isDone()).count();
      String msg = String.format("Flush failed with error. Total pending sends %d", incompleteSends);
      LOG.error(msg, e);
      throw new SamzaException(msg, e);
    }

    pendingFutures.clear();

    checkForSendCallbackErrors("Sending one or more of the messages failed during flush.");
  }
Default implementation of the flush that just waits for all the pendingFutures to be complete.
SystemProducer should override this, if the underlying system provides flush semantics.
@param source String representing the source of the message.
samza-azure/src/main/java/org/apache/samza/system/eventhub/producer/AsyncSystemProducer.java