kirktrue commented on code in PR #16241:
URL: https://github.com/apache/kafka/pull/16241#discussion_r1634062217
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -1132,12 +1133,26 @@ private CompletableFuture<Map<TopicPartition,
OffsetAndMetadata>> addOffsetFetch
Optional<OffsetFetchRequestState> inflight =
inflightOffsetFetches.stream().filter(r ->
r.sameRequest(request)).findAny();
- if (dupe.isPresent() || inflight.isPresent()) {
+ if (dupe.isPresent()) {
log.debug("Duplicated unsent offset fetch request found for
partitions: {}", request.requestedPartitions);
- dupe.orElseGet(inflight::get).chainFuture(request.future);
+ dupe.get().chainFuture(request.future);
+ } else if (inflight.isPresent()) {
+ log.debug("Duplicated inflight offset fetch request found for
partitions: {}", request.requestedPartitions);
+ OffsetFetchRequestState existing = inflight.get();
+ existing.chainFuture(request.future);
+
+ if (existing.future.isDone()) {
+ boolean inflightRemoved =
inflightOffsetFetches.remove(existing);
+ if (!inflightRemoved)
+ log.warn("The offset fetch request for partitions {}
was not found in the inflight buffer", request.requestedPartitions);
+ }
} else {
log.debug("Enqueuing offset fetch request for partitions: {}",
request.requestedPartitions);
this.unsentOffsetFetches.add(request);
+
+ // The incoming offset fetch request isn't in the unsent or
inflight buffers, which means we don't
+ // need to keep track of the entry in the inflight buffer any
longer.
+ inflightOffsetFetches.removeIf(r -> request.isExpired());
Review Comment:
🤦♂️
--
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]
