@Override
    public List<ClientResponse> poll(long timeoutMs, long now) {
        maybeAwaitWakeup();
        checkTimeoutOfPendingRequests(now);

        // We skip metadata updates if all nodes are currently blacked out
        if (metadataUpdater.isUpdateNeeded() && leastLoadedNode(now).node() != null) {
            MetadataUpdate metadataUpdate = metadataUpdates.poll();
            if (metadataUpdate != null) {
                metadataUpdater.update(time, metadataUpdate);
            } else {
                metadataUpdater.updateWithCurrentMetadata(time);
            }
        }

        List<ClientResponse> copy = new ArrayList<>();
        ClientResponse response;
        while ((response = this.responses.poll()) != null) {
            response.onComplete();
            copy.add(response);
        }

        return copy;
    }
Simulate a blocking poll in order to test wakeup behavior.

@param numBlockingWakeups The number of polls which will block until woken up
clients/src/test/java/org/apache/kafka/clients/MockClient.java