void abortBatches(final RuntimeException reason) {
        for (ProducerBatch batch : incomplete.copyAll()) {
            Deque<ProducerBatch> dq = getDeque(batch.topicPartition);
            synchronized (dq) {
                batch.abortRecordAppends();
                dq.remove(batch);
            }
            batch.abort(reason);
            deallocate(batch);
        }
    }
Abort all incomplete batches (whether they have been sent or not)
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java