@Test
    public void testInitProducerIdWithMaxInFlightOne() {
        final long producerId = 123456L;
        createMockClientWithMaxFlightOneMetadataPending();

        // Initialize transaction manager. InitProducerId will be queued up until metadata response
        // is processed and FindCoordinator can be sent to `leastLoadedNode`.
        TransactionManager transactionManager = new TransactionManager(new LogContext(), "testInitProducerIdWithPendingMetadataRequest",
                60000, 100L, new ApiVersions());
        setupWithTransactionState(transactionManager, false, null, false);
        ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(producerId, (short) 0);
        transactionManager.initializeTransactions();
        sender.runOnce();

        // Process metadata response, prepare FindCoordinator and InitProducerId responses.
        // Verify producerId after the sender is run to process responses.
        MetadataResponse metadataUpdate = RequestTestUtils.metadataUpdateWith(1, Collections.emptyMap());
        client.respond(metadataUpdate);
        prepareFindCoordinatorResponse(Errors.NONE, "testInitProducerIdWithPendingMetadataRequest");
        prepareInitProducerResponse(Errors.NONE, producerIdAndEpoch.producerId, producerIdAndEpoch.epoch);
        waitForProducerId(transactionManager, producerIdAndEpoch);
    }
Verifies that InitProducerId of transactional producer succeeds even if metadata requests
are pending with only one bootstrap node available and maxInFlight=1, where multiple
polls are necessary to send requests.
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java