cadonna commented on code in PR #16241:
URL: https://github.com/apache/kafka/pull/16241#discussion_r1633243629
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -514,10 +514,11 @@ private void fetchOffsetsWithRetries(final
OffsetFetchRequestState fetchRequest,
// Retry the same fetch request while it fails with RetriableException
and the retry timeout hasn't expired.
currentResult.whenComplete((res, error) -> {
- boolean inflightRemoved =
pendingRequests.inflightOffsetFetches.remove(fetchRequest);
- if (!inflightRemoved) {
- log.warn("A duplicated, inflight, request was identified, but
unable to find it in the " +
- "outbound buffer:" + fetchRequest);
+ if (!fetchRequest.isExpired()) {
+ boolean inflightRemoved =
pendingRequests.inflightOffsetFetches.remove(fetchRequest);
+ if (!inflightRemoved) {
+ log.warn("The response for the offset fetch request for
partitions {} was not found in the inflight buffer",
fetchRequest.requestedPartitions);
+ }
Review Comment:
When is the request removed from the `inflightOffsetFetches` when it is
expired?
##########
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:
Does this line not remove all elements of `inflightOffsetFetches` if the
request in `request` is expired? Should this rather be:
```java
inflightOffsetFetches.removeIf(r -> r.isExpired());
```
?
##########
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:
What do you try to remove here?
--
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]
