@Override
    public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
        // Prune any requests which have timed out
        List<TopicMetadataRequestState> expiredRequests = inflightRequests.stream()
                .filter(TimedRequestState::isExpired)
                .collect(Collectors.toList());
        expiredRequests.forEach(TopicMetadataRequestState::expire);

        List<NetworkClientDelegate.UnsentRequest> requests = inflightRequests.stream()
            .map(req -> req.send(currentTimeMs))
            .filter(Optional::isPresent)
            .map(Optional::get)
            .collect(Collectors.toList());

        return requests.isEmpty() ? EMPTY : new NetworkClientDelegate.PollResult(0, requests);
    }
<p>
Manages the state of topic metadata requests. This manager returns a
{@link NetworkClientDelegate.PollResult} when a request is ready to
be sent. Specifically, this manager handles the following user API calls:
</p>
<ul>
<li>listTopics</li>
<li>partitionsFor</li>
</ul>
<p>
The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
prevent sending it without backing off from previous attempts.
Once a request is completed successfully or times out, its corresponding entry is removed.
</p>
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java