code: private void flush(long numRecordsToFlush) {
      log.debug("Flushing records from producer buffer");
      Future future;
      long numRecordsFlushed = 0L;
      while (((future = futures.poll()) != null) && (numRecordsFlushed++ < numRecordsToFlush)) {
        try {
          future.get();
        } catch (Exception e) {
          log.error("Exception encountered when flushing record", e);
        }
      }
      log.debug("Flushed {} records from producer buffer", numRecordsFlushed);
    }
docs: Flush any records that may be present in the producer buffer upto a maximum of <code>numRecordsToFlush</code>.
  This method is needed since Kafka 0.8 producer does not have a flush() API. In the absence of the flush()
  implementation, records which are present in the buffer but not in-flight may not be delivered at all when close()
  is called, leading to data loss.
  @param numRecordsToFlush