@Test
    public void testReconnectAfterAddressChange() {
        AddressChangeHostResolver mockHostResolver = new AddressChangeHostResolver(
                initialAddresses.toArray(new InetAddress[0]), newAddresses.toArray(new InetAddress[0]));
        AtomicInteger initialAddressConns = new AtomicInteger();
        AtomicInteger newAddressConns = new AtomicInteger();
        MockSelector selector = new MockSelector(this.time, inetSocketAddress -> {
            InetAddress inetAddress = inetSocketAddress.getAddress();
            if (initialAddresses.contains(inetAddress)) {
                initialAddressConns.incrementAndGet();
            } else if (newAddresses.contains(inetAddress)) {
                newAddressConns.incrementAndGet();
            }
            return (mockHostResolver.useNewAddresses() && newAddresses.contains(inetAddress)) ||
                   (!mockHostResolver.useNewAddresses() && initialAddresses.contains(inetAddress));
        });

        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, false, new ApiVersions(), null, new LogContext(), mockHostResolver, mockClientTelemetrySender,
                MetadataRecoveryStrategy.NONE);

        // Connect to one the initial addresses, then change the addresses and disconnect
        client.ready(node, time.milliseconds());
        time.sleep(connectionSetupTimeoutMaxMsTest);
        client.poll(0, time.milliseconds());
        assertTrue(client.isReady(node, time.milliseconds()));
        // First poll should try to update the node but couldn't because node remains in connecting state
        // i.e. connection handling is completed after telemetry update.
        assertNull(client.telemetryConnectedNode());

        client.poll(0, time.milliseconds());
        assertEquals(node, client.telemetryConnectedNode());

        mockHostResolver.changeAddresses();
        selector.serverDisconnect(node.idString());
        client.poll(0, time.milliseconds());
        assertFalse(client.isReady(node, time.milliseconds()));
        assertNull(client.telemetryConnectedNode());

        time.sleep(reconnectBackoffMaxMsTest);
        client.ready(node, time.milliseconds());
        time.sleep(connectionSetupTimeoutMaxMsTest);
        client.poll(0, time.milliseconds());
        assertTrue(client.isReady(node, time.milliseconds()));
        assertNull(client.telemetryConnectedNode());

        client.poll(0, time.milliseconds());
        assertEquals(node, client.telemetryConnectedNode());

        // We should have tried to connect to one initial address and one new address, and resolved DNS twice
        assertEquals(1, initialAddressConns.get());
        assertEquals(1, newAddressConns.get());
        assertEquals(2, mockHostResolver.resolutionCount());
        verify(mockClientTelemetrySender, times(5)).timeToNextUpdate(anyLong());
    }
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