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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org