@Test
    public void testRecordErrorPropagatedToApplication() throws InterruptedException {
        int recordCount = 5;

        setup();

        Map<Integer, FutureRecordMetadata> futures = new HashMap<>(recordCount);
        for (int i = 0; i < recordCount; i++) {
            futures.put(i, appendToAccumulator(tp0));
        }

        sender.runOnce();  // send request
        assertEquals(1, client.inFlightRequestCount());
        assertEquals(1, sender.inFlightBatches(tp0).size());

        OffsetAndError offsetAndError = new OffsetAndError(-1L, Errors.INVALID_RECORD, Arrays.asList(
            new BatchIndexAndErrorMessage().setBatchIndex(0).setBatchIndexErrorMessage("0"),
            new BatchIndexAndErrorMessage().setBatchIndex(2).setBatchIndexErrorMessage("2"),
            new BatchIndexAndErrorMessage().setBatchIndex(3)
        ));

        client.respond(produceResponse(Collections.singletonMap(tp0, offsetAndError)));
        sender.runOnce();

        for (Map.Entry<Integer, FutureRecordMetadata> futureEntry : futures.entrySet()) {
            FutureRecordMetadata future = futureEntry.getValue();
            assertTrue(future.isDone());

            KafkaException exception = TestUtils.assertFutureThrows(future, KafkaException.class);
            Integer index = futureEntry.getKey();
            if (index == 0 || index == 2) {
                assertInstanceOf(InvalidRecordException.class, exception);
                assertEquals(index.toString(), exception.getMessage());
            } else if (index == 3) {
                assertInstanceOf(InvalidRecordException.class, exception);
                assertEquals(Errors.INVALID_RECORD.message(), exception.getMessage());
            } else {
                assertEquals(KafkaException.class, exception.getClass());
            }
        }
    }
Tests the code path where the target node to send FindCoordinator or InitProducerId
is not ready.
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java