private void closeInternal(final Duration timeout) {
        long timeoutMs = timeout.toMillis();
        log.trace("Signaling the consumer network thread to close in {}ms", timeoutMs);
        running = false;
        closeTimeout = timeout;
        wakeup();

        try {
            join();
        } catch (InterruptedException e) {
            log.error("Interrupted while waiting for consumer network thread to complete", e);
        }
    }
Starts the closing process.

<p/>

This method is called from the application thread, but our resources are owned by the network thread. As such,
we don't actually close any of those resources here, immediately, on the application thread. Instead, we just
update our internal state on the application thread. When the network thread next
{@link #run() executes its loop}, it will notice that state, cease processing any further events, and begin
{@link #cleanup() closing its resources}.

<p/>

This method will wait (i.e. block the application thread) for up to the duration of the given timeout to give
the network thread the time to close down cleanly.

@param timeout Upper bound of time to wait for the network thread to close its resources
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java