code: public void flush() {
      try {
        ArrayList<Batch> batches = this.incomplete.all();
        int numOutstandingRecords = 0;
        for (Batch batch: batches) {
          numOutstandingRecords += batch.getRecords().size();
        }
        LOG.debug ("Flush called on {} batches with {} records total", batches.size(), numOutstandingRecords);
        for (Batch batch: batches) {
          batch.await();
        }
      } catch (Exception e) {
        LOG.error ("Error happened while flushing batches");
      }
    }
docs: This will block until all the incomplete batches are acknowledged