lianetm commented on code in PR #22747:
URL: https://github.com/apache/kafka/pull/22747#discussion_r3532096372
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -1223,9 +1225,14 @@ private void onFailure(final long currentTimeMs,
" anymore.", responseError);
future.completeExceptionally(exception);
} else if (responseError == Errors.STALE_MEMBER_EPOCH) {
- log.error("OffsetFetch failed with {} and the consumer is not
part " +
- "of the group anymore (it probably left the group, got
fenced" +
- " or failed). The request cannot be retried and will
fail.", responseError);
+ if (memberInfo.memberEpoch.isPresent()) {
+ log.debug("OffsetFetch failed with {}. The member has a
newer epoch, so the " +
+ "request will be retried with it.", responseError);
Review Comment:
will be retried...if there's time left (worth clarifying, we cannot
guarantee it will be retried at this point. The retry logic is elsewhere and
based on expiration)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -554,8 +554,10 @@ private void fetchOffsetsWithRetries(final
OffsetFetchRequestState fetchRequest,
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);
+ // Requests deduplicated onto another pending request are
never added to the
+ // buffers, so there is nothing to remove for them here.
+ log.debug("Completed request not found in the in-flight buffer
(it was " +
+ "deduplicated and chained onto an existing request): {}",
fetchRequest);
Review Comment:
dedup/chained it's certainly one of the reasons for this, but I'm not sure
it's the only one. E.g, requests that remain unsent (not added to inflight) if
we don't have a coordinator?. So maybe we can consider letting the log with
what we're sure of (what it had before, dup found but not inflight), and we
mention some of the potential reasons in a comment? wdyt?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -1238,6 +1238,66 @@ public void
testSyncOffsetFetchFailsWithStaleEpochAndRetriesWithNewEpoch() {
assertEquals(memberId, reqData.groups().get(0).memberId());
}
+ // Same as testSyncOffsetFetchFailsWithStaleEpochAndRetriesWithNewEpoch,
but with a
+ // duplicated fetch for the same partitions chained onto the in-flight
request when the
+ // STALE_MEMBER_EPOCH error is received. The retry of the chained request
must not be
+ // deduplicated against the already-completed in-flight request: chaining
onto a completed
+ // future fails it again immediately and re-triggers the retry in a tight
synchronous loop
+ // that never sends a request with the new epoch and never completes the
callers' futures
+ // (KAFKA-20765).
+ @Test
+ public void
testDuplicatedOffsetFetchFailsWithStaleEpochAndRetriesWithNewEpoch() {
+ CommitRequestManager commitRequestManager = create(false, 100);
+ Set<TopicPartition> partitions = Collections.singleton(new
TopicPartition("t1", 0));
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+
+ // Two callers fetch offsets for the same partitions; the second
request is deduplicated
+ // and chained onto the first.
+ long deadlineMs = time.milliseconds() + defaultApiTimeoutMs;
+ CompletableFuture<CommitRequestManager.OffsetFetchResult> firstResult =
+ commitRequestManager.fetchOffsets(partitions, deadlineMs);
+ CompletableFuture<CommitRequestManager.OffsetFetchResult> secondResult
=
+ commitRequestManager.fetchOffsets(partitions, deadlineMs);
+
+ // A single deduplicated request goes on the wire.
+ NetworkClientDelegate.PollResult res =
commitRequestManager.poll(time.milliseconds());
+ assertEquals(1, res.unsentRequests.size());
+
+ // Mock member has a new valid epoch, so STALE_MEMBER_EPOCH is
retriable.
+ int newEpoch = 8;
+ String memberId = "member1";
+ commitRequestManager.onMemberEpochUpdated(Optional.of(newEpoch),
memberId);
+
+ // Receive error when member already has a newer member epoch. Request
should be retried.
+ res.unsentRequests.get(0).handler().onComplete(
+ buildOffsetFetchClientResponse(res.unsentRequests.get(0),
partitions, Errors.STALE_MEMBER_EPOCH));
+
+ // The failed request should be removed from the in-flight buffer, a
retry should be
+ // enqueued, and the callers' futures should still be waiting for the
retry's outcome.
+ assertEquals(0,
commitRequestManager.pendingRequests.inflightOffsetFetches.size());
+ assertEquals(1,
commitRequestManager.pendingRequests.unsentOffsetFetches.size());
+ assertFalse(firstResult.isDone());
+ assertFalse(secondResult.isDone());
+
+ // The retried request should be sent after the backoff, with the
latest member ID and epoch.
+ time.sleep(retryBackoffMs);
Review Comment:
uhm I think there is no backoff here (by the time it fails the new epoch is
known already, so we just send it on the next poll). I expect the test should
pass without the sleep and that we can remove the comment too.
--
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]
