private void testCompleteExceptionally(
        int recordCount,
        RuntimeException topLevelException,
        Function<Integer, RuntimeException> recordExceptions
    ) {
        ProducerBatch batch = new ProducerBatch(
            new TopicPartition("topic", 1),
            memoryRecordsBuilder,
            now
        );

        List<FutureRecordMetadata> futures = new ArrayList<>(recordCount);
        for (int i = 0; i < recordCount; i++) {
            futures.add(batch.tryAppend(now, null, new byte[10], Record.EMPTY_HEADERS, null, now));
        }
        assertEquals(recordCount, batch.recordCount);

        batch.completeExceptionally(topLevelException, recordExceptions);
        assertTrue(batch.isDone());

        for (int i = 0; i < futures.size(); i++) {
            FutureRecordMetadata future = futures.get(i);
            RuntimeException caughtException = TestUtils.assertFutureThrows(future, RuntimeException.class);
            RuntimeException expectedException = recordExceptions.apply(i);
            assertEquals(expectedException, caughtException);
        }
    }
This tests that leader is correctly maintained & leader-change is correctly detected across retries
of the batch. It does so by testing primarily testing methods
1. maybeUpdateLeaderEpoch
2. hasLeaderChangedForTheOngoingRetry
clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerBatchTest.java