kirktrue commented on code in PR #16031:
URL: https://github.com/apache/kafka/pull/16031#discussion_r1624992826
##########
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've updated the logic in `CommitRequestManager` and
`TopicMetadataRequestManager` to proactively prune expired `RequestState`s that
have had at least one request.
--
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]
