@Override
        protected void checkDisconnects(final long currentTimeMs) {
            // any disconnects affecting requests that have already been transmitted will be handled
            // by NetworkClient, so we just need to check whether connections for any of the unsent
            // requests have been disconnected; if they have, then we complete the corresponding future
            // and set the disconnect flag in the ClientResponse
            for (Node node : unsentRequestNodes()) {
                if (client.connectionFailed(node)) {
                    // Remove entry before invoking request callback to avoid callbacks handling
                    // coordinator failures traversing the unsent list again.
                    for (UnsentRequest unsentRequest : removeUnsentRequestByNode(node)) {
                        // TODO: this should likely emulate what's done in ConsumerNetworkClient
                        log.error("checkDisconnects - please update! unsentRequest: {}", unsentRequest);
                    }
                }
            }
        }
Assert that the {@link Fetcher#collectFetch() latest fetch} does not contain any
{@link Fetch#records() user-visible records}, did not
{@link Fetch#positionAdvanced() advance the consumer's position},
and is {@link Fetch#isEmpty() empty}.
@param reason the reason to include for assertion methods such as {@link org.junit.jupiter.api.Assertions#assertTrue(boolean, String)}
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java