@Test
    public void testTelemetryRequest() {
        ClientTelemetrySender mockClientTelemetrySender = mock(ClientTelemetrySender.class);
        when(mockClientTelemetrySender.timeToNextUpdate(anyLong())).thenReturn(0L);

        NetworkClient client = new NetworkClient(metadataUpdater, null, selector, "mock", Integer.MAX_VALUE,
            reconnectBackoffMsTest, reconnectBackoffMaxMsTest, 64 * 1024, 64 * 1024,
            defaultRequestTimeoutMs, connectionSetupTimeoutMsTest, connectionSetupTimeoutMaxMsTest,
            time, true, new ApiVersions(), null, new LogContext(), new DefaultHostResolver(), mockClientTelemetrySender,
            MetadataRecoveryStrategy.NONE);

        // Send the ApiVersionsRequest
        client.ready(node, time.milliseconds());
        client.poll(0, time.milliseconds());
        assertNull(client.telemetryConnectedNode());
        assertTrue(client.hasInFlightRequests(node.idString()));
        delayedApiVersionsResponse(0, ApiKeys.API_VERSIONS.latestVersion(), TestUtils.defaultApiVersionsResponse(
            ApiMessageType.ListenerType.BROKER));
        // handle ApiVersionsResponse
        client.poll(0, time.milliseconds());
        // the ApiVersionsRequest is gone
        assertFalse(client.hasInFlightRequests(node.idString()));
        selector.clear();

        GetTelemetrySubscriptionsRequest.Builder getRequest = new GetTelemetrySubscriptionsRequest.Builder(
            new GetTelemetrySubscriptionsRequestData(), true);
        when(mockClientTelemetrySender.createRequest()).thenReturn(Optional.of(getRequest));

        GetTelemetrySubscriptionsResponse getResponse = new GetTelemetrySubscriptionsResponse(new GetTelemetrySubscriptionsResponseData());
        ByteBuffer buffer = RequestTestUtils.serializeResponseWithHeader(getResponse, ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS.latestVersion(), 1);
        selector.completeReceive(new NetworkReceive(node.idString(), buffer));

        // Initiate poll to send GetTelemetrySubscriptions request
        client.poll(0, time.milliseconds());
        assertTrue(client.isReady(node, time.milliseconds()));
        assertEquals(node, client.telemetryConnectedNode());
        verify(mockClientTelemetrySender, times(1)).handleResponse(any(GetTelemetrySubscriptionsResponse.class));
        selector.clear();

        PushTelemetryRequest.Builder pushRequest = new PushTelemetryRequest.Builder(
            new PushTelemetryRequestData(), true);
        when(mockClientTelemetrySender.createRequest()).thenReturn(Optional.of(pushRequest));

        PushTelemetryResponse pushResponse = new PushTelemetryResponse(new PushTelemetryResponseData());
        ByteBuffer pushBuffer = RequestTestUtils.serializeResponseWithHeader(pushResponse, ApiKeys.PUSH_TELEMETRY.latestVersion(), 2);
        selector.completeReceive(new NetworkReceive(node.idString(), pushBuffer));

        // Initiate poll to send PushTelemetry request
        client.poll(0, time.milliseconds());
        assertTrue(client.isReady(node, time.milliseconds()));
        assertEquals(node, client.telemetryConnectedNode());
        verify(mockClientTelemetrySender, times(1)).handleResponse(any(PushTelemetryResponse.class));
        verify(mockClientTelemetrySender, times(4)).timeToNextUpdate(anyLong());
        verify(mockClientTelemetrySender, times(2)).createRequest();
    }
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