@Test
    public void testClusterAuthorizationExceptionInInitProducerIdRequest() throws Exception {
        final long producerId = 343434L;
        TransactionManager transactionManager = createTransactionManager();
        setupWithTransactionState(transactionManager);
        // cluster authorization failed on initProducerId is retriable
        prepareAndReceiveInitProducerId(producerId, Errors.CLUSTER_AUTHORIZATION_FAILED);
        assertFalse(transactionManager.hasProducerId());
        assertTrue(transactionManager.hasError());
        assertInstanceOf(ClusterAuthorizationException.class, transactionManager.lastError());
        assertEquals(-1, transactionManager.producerIdAndEpoch().epoch);

        assertSendFailure(ClusterAuthorizationException.class);
        prepareAndReceiveInitProducerId(producerId, Errors.NONE);
        // sender retry initProducerId and succeed
        sender.runOnce();
        assertFalse(transactionManager.hasFatalError());
        assertTrue(transactionManager.hasProducerId());
        assertEquals(0, transactionManager.producerIdAndEpoch().epoch);
        // subsequent send should be successful
        assertSuccessfulSend();
    }
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