@Timeout(5)
    @Test
    public void testSenderShouldCloseWhenTransactionManagerInErrorState() {
        metrics.close();
        Map<String, String> clientTags = Collections.singletonMap("client-id", "clientA");
        metrics = new Metrics(new MetricConfig().tags(clientTags));
        TransactionManager transactionManager = mock(TransactionManager.class);
        SenderMetricsRegistry metricsRegistry = new SenderMetricsRegistry(metrics);
        Sender sender = new Sender(logContext, client, metadata, this.accumulator, false, MAX_REQUEST_SIZE, ACKS_ALL,
                1, metricsRegistry, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, transactionManager, apiVersions);
        when(transactionManager.hasOngoingTransaction()).thenReturn(true);
        when(transactionManager.beginAbort()).thenThrow(new IllegalStateException());
        sender.initiateClose();

        // The sender should directly get closed.
        sender.run();
        verify(transactionManager, times(1)).close();
    }
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