code: private void flush(KafkaTransactionState transaction) throws FlinkKafka011Exception {
  		if (transaction.producer != null) {
  			transaction.producer.flush();
  		}
  		long pendingRecordsCount = pendingRecords.get();
  		if (pendingRecordsCount != 0) {
  			throw new IllegalStateException("Pending record count must be zero at this point: " + pendingRecordsCount);
  		}
  
  		// if the flushed requests has errors, we should propagate it also and fail the checkpoint
  		checkErroneous();
  	}
docs: Flush pending records.
  @param transaction