kirktrue commented on code in PR #16031:
URL: https://github.com/apache/kafka/pull/16031#discussion_r1624792665
##########
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:
It makes sense to have `poll()` expire requests that are _in flight_. I
don't think `poll()` should expire _unsent_ requests, because we want to let
them have a chance to run at least once. I'll change the logic to have `poll()`
expire `Future`s related to requests that have been sent but that are now
expired.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
