@Test
    public void testCanRetryWithoutIdempotence() throws Exception {
        // do a successful retry
        Future<RecordMetadata> future = appendToAccumulator(tp0, 0L, "key", "value");
        sender.runOnce(); // connect
        sender.runOnce(); // send produce request
        String id = client.requests().peek().destination();
        Node node = new Node(Integer.parseInt(id), "localhost", 0);
        assertEquals(1, client.inFlightRequestCount());
        assertTrue(client.hasInFlightRequests());
        assertEquals(1, sender.inFlightBatches(tp0).size());
        assertTrue(client.isReady(node, time.milliseconds()), "Client ready status should be true");
        assertFalse(future.isDone());

        client.respond(body -> {
            ProduceRequest request = (ProduceRequest) body;
            assertFalse(RequestTestUtils.hasIdempotentRecords(request));
            return true;
        }, produceResponse(tp0, -1L, Errors.TOPIC_AUTHORIZATION_FAILED, 0));
        sender.runOnce();
        assertTrue(future.isDone());
        assertInstanceOf(TopicAuthorizationException.class, assertThrows(Exception.class, future::get).getCause());
    }
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