@Test
    public void testReadyAndDrainWhenABatchIsBeingRetried() throws InterruptedException {
        int part1LeaderEpoch = 100;
        // Create cluster metadata, partition1 being hosted by node1
        PartitionMetadata part1Metadata = new PartitionMetadata(Errors.NONE, tp1, Optional.of(node1.id()),  Optional.of(part1LeaderEpoch), null, null, null);
        MetadataSnapshot metadataCache = new MetadataSnapshot(null, nodes, Collections.singletonList(part1Metadata), Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), null, Collections.emptyMap());

        int batchSize = 10;
        int lingerMs = 10;
        int retryBackoffMs = 100;
        int retryBackoffMaxMs = 1000;
        int deliveryTimeoutMs = Integer.MAX_VALUE;
        long totalSize = 10 * 1024;
        String metricGrpName = "producer-metrics";
        final RecordAccumulator accum = new RecordAccumulator(logContext, batchSize,
            Compression.NONE, lingerMs, retryBackoffMs, retryBackoffMaxMs,
            deliveryTimeoutMs, metrics, metricGrpName, time, new ApiVersions(), null,
            new BufferPool(totalSize, batchSize, metrics, time, metricGrpName));

        // Create 1 batch(batchA) to be produced to partition1.
        long now = time.milliseconds();
        accum.append(topic, partition1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs, false, now, cluster);

        // 1st attempt(not a retry) to produce batchA, it should be ready & drained to be produced.
        {
            now += lingerMs + 1;
            RecordAccumulator.ReadyCheckResult result = accum.ready(metadataCache, now);
            assertTrue(result.readyNodes.contains(node1), "Node1 is ready");

            Map<Integer, List<ProducerBatch>> batches = accum.drain(metadataCache,
                result.readyNodes, 999999 /* maxSize */, now);
            assertTrue(batches.containsKey(node1.id()) && batches.get(node1.id()).size() == 1, "Node1 has 1 batch ready & drained");
            ProducerBatch batch = batches.get(node1.id()).get(0);
            assertEquals(OptionalInt.of(part1LeaderEpoch), batch.currentLeaderEpoch());
            assertEquals(0, batch.attemptsWhenLeaderLastChanged());
            // Re-enqueue batch for subsequent retries & test-cases
            accum.reenqueue(batch, now);
        }

        // In this retry of batchA, wait-time between retries is less than configured and no leader change, so should backoff.
        {
            now += 1;
            RecordAccumulator.ReadyCheckResult result = accum.ready(metadataCache, now);
            assertFalse(result.readyNodes.contains(node1), "Node1 is not ready");

            // Try to drain from node1, it should return no batches.
            Map<Integer, List<ProducerBatch>> batches = accum.drain(metadataCache,
                new HashSet<>(Collections.singletonList(node1)), 999999 /* maxSize */, now);
            assertTrue(batches.containsKey(node1.id()) && batches.get(node1.id()).isEmpty(),
                "No batches ready to be drained on Node1");
        }

        // In this retry of batchA, wait-time between retries is less than configured and leader has changed, so should not backoff.
        {
            now += 1;
            part1LeaderEpoch++;
            // Create cluster metadata, with new leader epoch.
            part1Metadata = new PartitionMetadata(Errors.NONE, tp1, Optional.of(node1.id()),  Optional.of(part1LeaderEpoch), null, null, null);
            metadataCache = new MetadataSnapshot(null, nodes, Collections.singletonList(part1Metadata), Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), null, Collections.emptyMap());
            RecordAccumulator.ReadyCheckResult result = accum.ready(metadataCache, now);
            assertTrue(result.readyNodes.contains(node1), "Node1 is ready");

            Map<Integer, List<ProducerBatch>> batches = accum.drain(metadataCache,
                result.readyNodes, 999999 /* maxSize */, now);
            assertTrue(batches.containsKey(node1.id()) && batches.get(node1.id()).size() == 1, "Node1 has 1 batch ready & drained");
            ProducerBatch batch = batches.get(node1.id()).get(0);
            assertEquals(OptionalInt.of(part1LeaderEpoch), batch.currentLeaderEpoch());
            assertEquals(1, batch.attemptsWhenLeaderLastChanged());

            // Re-enqueue batch for subsequent retries/test-cases.
            accum.reenqueue(batch, now);
        }

        // In this retry of batchA, wait-time between retries is more than configured and no leader change, so should not backoff.
        {
            now += 2 * retryBackoffMaxMs;
            // Create cluster metadata, with new leader epoch.
            part1Metadata = new PartitionMetadata(Errors.NONE, tp1, Optional.of(node1.id()),  Optional.of(part1LeaderEpoch), null, null, null);
            metadataCache = new MetadataSnapshot(null, nodes, Collections.singletonList(part1Metadata), Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), null, Collections.emptyMap());
            RecordAccumulator.ReadyCheckResult result = accum.ready(metadataCache, now);
            assertTrue(result.readyNodes.contains(node1), "Node1 is ready");

            Map<Integer, List<ProducerBatch>> batches = accum.drain(metadataCache,
                result.readyNodes, 999999 /* maxSize */, now);
            assertTrue(batches.containsKey(node1.id()) && batches.get(node1.id()).size() == 1, "Node1 has 1 batch ready & drained");
            ProducerBatch batch = batches.get(node1.id()).get(0);
            assertEquals(OptionalInt.of(part1LeaderEpoch), batch.currentLeaderEpoch());
            assertEquals(1, batch.attemptsWhenLeaderLastChanged());

            // Re-enqueue batch for subsequent retries/test-cases.
            accum.reenqueue(batch, now);
        }

        // In this retry of batchA, wait-time between retries is more than configured and leader has changed, so should not backoff.
        {
            now += 2 * retryBackoffMaxMs;
            part1LeaderEpoch++;
            // Create cluster metadata, with new leader epoch.
            part1Metadata = new PartitionMetadata(Errors.NONE, tp1, Optional.of(node1.id()),  Optional.of(part1LeaderEpoch), null, null, null);
            metadataCache = new MetadataSnapshot(null, nodes, Collections.singletonList(part1Metadata), Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), null, Collections.emptyMap());
            RecordAccumulator.ReadyCheckResult result = accum.ready(metadataCache, now);
            assertTrue(result.readyNodes.contains(node1), "Node1 is ready");

            Map<Integer, List<ProducerBatch>> batches = accum.drain(metadataCache,
                result.readyNodes, 999999 /* maxSize */, now);
            assertTrue(batches.containsKey(node1.id()) && batches.get(node1.id()).size() == 1, "Node1 has 1 batch ready & drained");
            ProducerBatch batch = batches.get(node1.id()).get(0);
            assertEquals(OptionalInt.of(part1LeaderEpoch), batch.currentLeaderEpoch());
            assertEquals(3, batch.attemptsWhenLeaderLastChanged());

            // Re-enqueue batch for subsequent retries/test-cases.
            accum.reenqueue(batch, now);
        }
    }
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