@Test
    public void testDrainWithANodeThatDoesntHostAnyPartitions() {
        int batchSize = 10;
        int lingerMs = 10;
        long totalSize = 10 * 1024;
        RecordAccumulator accum = createTestRecordAccumulator(batchSize, totalSize, Compression.NONE, lingerMs);

        // Create cluster metadata, node2 doesn't host any partitions.
        PartitionMetadata part1Metadata = new PartitionMetadata(Errors.NONE, tp1, Optional.of(node1.id()), Optional.empty(), null, null, null);
        MetadataSnapshot metadataCache = new MetadataSnapshot(null, nodes, Collections.singletonList(part1Metadata), Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), null, Collections.emptyMap());

        // Drain for node2, it should return 0 batches,
        Map<Integer, List<ProducerBatch>> batches = accum.drain(metadataCache,
            new HashSet<>(Collections.singletonList(node2)), 999999 /* maxSize */, time.milliseconds());
        assertTrue(batches.get(node2.id()).isEmpty());
    }
For a batch being retried, this validates ready() and drain() whether a batch should skip-backoff(retries-immediately), or backoff, based on -
1. how long it has waited between retry attempts.
2. change in leader hosting the partition.
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java