lianetm commented on code in PR #16031: URL: https://github.com/apache/kafka/pull/16031#discussion_r1624689065
########## clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java: ########## @@ -61,30 +62,28 @@ */ public class TopicMetadataRequestManager implements RequestManager { + private final Time time; private final boolean allowAutoTopicCreation; private final List<TopicMetadataRequestState> inflightRequests; + private final int requestTimeoutMs; private final long retryBackoffMs; private final long retryBackoffMaxMs; private final Logger log; private final LogContext logContext; - public TopicMetadataRequestManager(final LogContext context, final ConsumerConfig config) { + public TopicMetadataRequestManager(final LogContext context, final Time time, final ConsumerConfig config) { logContext = context; log = logContext.logger(getClass()); + this.time = time; inflightRequests = new LinkedList<>(); + requestTimeoutMs = config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG); retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG); retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG); allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG); } @Override public NetworkClientDelegate.PollResult poll(final long currentTimeMs) { - // Prune any requests which have timed out - List<TopicMetadataRequestState> expiredRequests = inflightRequests.stream() - .filter(req -> req.isExpired(currentTimeMs)) - .collect(Collectors.toList()); - expiredRequests.forEach(TopicMetadataRequestState::expire); - Review Comment: I totally agree that we shouldn't prune expired request on poll before returning the requests (because it does not give requests with timeout 0 their single time to run that we wanted). But not pruning at all on poll goes a bit farther than what I was expecting. Seems conceptually not right that the manager will keep requests that it has sent out, and it knows are expired. It then relies on a timeout response to expire the request, and with this the outcome is the same, but timing is very different: the manager will wait for the full request_timeout before throwing a `TimeoutException` for a request that it knew it was expired since the moment it poll for it the first time right? (all that time it will keep the request on its queue as in-flight, skipping it on every poll just because it hasn't received a response). I expect that tests that throw Timeout on poll would fail with this, and they would need to wait for a network timeout response that was not needed before, so that the manager actually throws TimeoutException, that's what I'm referring to. Managers already apply expiration logic to requests: 1. on poll (before returning the requests) and on response (before this PR) 2. on response only (this PR) 3. on poll (after returning the requests) and on responses -> would this be a sensible compromise, so that we end up with managers that apply the expiration logic consistently wherever they can? It's closer to what we had before this PR, and also fixes the fire-and-forget behaviour we were missing. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org