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

        // Initialize transaction manager. InitProducerId will be queued up until metadata response
        // is processed.
        TransactionManager transactionManager = createTransactionManager();
        setupWithTransactionState(transactionManager, false, null, false);
        ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(producerId, (short) 0);

        // Process metadata and InitProducerId responses.
        // Verify producerId after the sender is run to process responses.
        MetadataResponse metadataUpdate = RequestTestUtils.metadataUpdateWith(1, Collections.emptyMap());
        client.respond(metadataUpdate);
        sender.runOnce();
        sender.runOnce();
        client.respond(initProducerIdResponse(producerIdAndEpoch.producerId, producerIdAndEpoch.epoch, Errors.NONE));
        waitForProducerId(transactionManager, producerIdAndEpoch);
    }
Verifies that InitProducerId of idempotent 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