RecordAccumulator { public void abortIncompleteBatches() { do { abortBatches(); } while (appendsInProgress()); abortBatches(); this.batches.clear(); } RecordAccumulator(int batchSize,
        long totalSize,
        CompressionType compression,
        long lingerMs,
        long retryBackoffMs,
        Metrics metrics,
        Time time); RecordAppendResult append(TopicPartition tp,
        long timestamp,
        byte[] key,
        byte[] value,
        Callback callback,
        long maxTimeToBlock); List<RecordBatch> abortExpiredBatches(int requestTimeout, long now); void reenqueue(RecordBatch batch, long now); ReadyCheckResult ready(Cluster cluster, long nowMs); boolean hasUnsent(); Map<Integer, List<RecordBatch>> drain(Cluster cluster,
        Set<Node> nodes,
        int maxSize,
        long now); void deallocate(RecordBatch batch); void beginFlush(); void awaitFlushCompletion(); void abortIncompleteBatches(); void mutePartition(TopicPartition tp); void unmutePartition(TopicPartition tp); void close();  }
@Test public void testAbortIncompleteBatches() throws Exception { long lingerMs = Long.MAX_VALUE; final AtomicInteger numExceptionReceivedInCallback = new AtomicInteger(0); final RecordAccumulator accum = new RecordAccumulator(4 * 1024, 64 * 1024, CompressionType.NONE, lingerMs, 100L, metrics, time); class TestCallback implements Callback { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { assertTrue(exception.getMessage().equals("Producer is closed forcefully.")); numExceptionReceivedInCallback.incrementAndGet(); } } for (int i = 0; i < 100; i++) accum.append(new TopicPartition(topic, i % 3), 0L, key, value, new TestCallback(), maxBlockTimeMs); RecordAccumulator.ReadyCheckResult result = accum.ready(cluster, time.milliseconds()); assertEquals("No nodes should be ready.", 0, result.readyNodes.size()); accum.abortIncompleteBatches(); assertEquals(numExceptionReceivedInCallback.get(), 100); assertFalse(accum.hasUnsent()); }