@Test
    public void testConnectionThrottling() {
        // Instrument the test to return a response with a 100ms throttle delay.
        awaitReady(client, node);
        short requestVersion = PRODUCE.latestVersion();
        ProduceRequest.Builder builder = new ProduceRequest.Builder(
            requestVersion,
            requestVersion,
            new ProduceRequestData()
                .setAcks((short) 1)
                .setTimeoutMs(1000));
        TestCallbackHandler handler = new TestCallbackHandler();
        ClientRequest request = client.newClientRequest(node.idString(), builder, time.milliseconds(), true,
                defaultRequestTimeoutMs, handler);
        client.send(request, time.milliseconds());
        client.poll(1, time.milliseconds());
        int throttleTime = 100;
        ProduceResponse produceResponse = new ProduceResponse(new ProduceResponseData().setThrottleTimeMs(throttleTime));
        ByteBuffer buffer = RequestTestUtils.serializeResponseWithHeader(produceResponse, requestVersion, request.correlationId());
        selector.completeReceive(new NetworkReceive(node.idString(), buffer));
        client.poll(1, time.milliseconds());

        // The connection is not ready due to throttling.
        assertFalse(client.ready(node, time.milliseconds()));
        assertEquals(100, client.throttleDelayMs(node, time.milliseconds()));

        // After 50ms, the connection is not ready yet.
        time.sleep(50);
        assertFalse(client.ready(node, time.milliseconds()));
        assertEquals(50, client.throttleDelayMs(node, time.milliseconds()));

        // After another 50ms, the throttling is done and the connection becomes ready again.
        time.sleep(50);
        assertTrue(client.ready(node, time.milliseconds()));
        assertEquals(0, client.throttleDelayMs(node, time.milliseconds()));
    }
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