@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());
}