@Test
    public void testDisconnectDuringUserMetadataRequest() {
        // this test ensures that the default metadata updater does not intercept a user-initiated
        // metadata request when the remote node disconnects with the request in-flight.
        awaitReady(client, node);

        MetadataRequest.Builder builder = new MetadataRequest.Builder(Collections.emptyList(), true);
        long now = time.milliseconds();
        ClientRequest request = client.newClientRequest(node.idString(), builder, now, true);
        client.send(request, now);
        client.poll(defaultRequestTimeoutMs, now);
        assertEquals(1, client.inFlightRequestCount(node.idString()));
        assertTrue(client.hasInFlightRequests(node.idString()));
        assertTrue(client.hasInFlightRequests());

        selector.close(node.idString());
        List<ClientResponse> responses = client.poll(defaultRequestTimeoutMs, time.milliseconds());
        assertEquals(1, responses.size());
        assertTrue(responses.iterator().next().wasDisconnected());
    }
This is a helper method that will execute two produce calls. The first call is expected to work and the
second produce call is intentionally made to emulate a request timeout. In the case that a timeout occurs
during a request, we want to ensure that we {@link Metadata#requestUpdate() request a metadata update} so that
on a subsequent invocation of {@link NetworkClient#poll(long, long) poll}, the metadata request will be sent.

<p/>

The {@link MetadataUpdater} has a specific method to handle
{@link NetworkClient.DefaultMetadataUpdater#handleServerDisconnect(long, String, Optional) server disconnects}
which is where we {@link Metadata#requestUpdate() request a metadata update}. This test helper method ensures
that is invoked by checking {@link Metadata#updateRequested()} after the simulated timeout.

@param requestTimeoutMs Timeout in ms
clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java