frankvicky commented on code in PR #21584:
URL: https://github.com/apache/kafka/pull/21584#discussion_r3861454415
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -276,15 +276,29 @@ private CompletableFuture<Map<TopicPartition,
OffsetAndMetadata>> requestAutoCom
*/
private void maybeAutoCommitAsync() {
if (autoCommitEnabled() && autoCommitState.get().shouldAutoCommit()) {
- OffsetCommitRequestState requestState = createOffsetCommitRequest(
- subscriptions.allConsumed(),
- Long.MAX_VALUE);
- CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> result =
requestAutoCommit(requestState);
- // Reset timer to the interval (even if no request was generated),
but ensure that if
- // the request completes with a retriable error, the timer is
reset to send the next
- // auto-commit after the backoff expires.
- resetAutoCommitTimer();
- maybeResetTimerWithBackoff(result);
+ doAutoCommitAsync();
+ }
+ }
+
+ private void doAutoCommitAsync() {
+ OffsetCommitRequestState requestState = createOffsetCommitRequest(
+ subscriptions.allConsumed(),
+ Long.MAX_VALUE);
+ CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> result =
requestAutoCommit(requestState);
+ // Reset timer to the interval (even if no request was generated), but
ensure that if
+ // the request completes with a retriable error, the timer is reset to
send the next
+ // auto-commit after the backoff expires.
+ resetAutoCommitTimer();
+ maybeResetTimerWithBackoff(result);
+ }
Review Comment:
`maybeAutoCommitOnAssignment()` goes through `doAutoCommitAsync()`, which
resets the interval timer upfront via `resetAutoCommitTimer()` — before we know
the outcome of the commit. If this assign-triggered commit fails with a
non-retriable error, we've delayed the next periodic auto-commit by a full
interval without having actually committed anything. This is the same concern
@lianetm raised earlier on the classic path, and it was addressed there: the
classic `maybeAutoCommitOffsetsAsync()` leaves `nextAutoCommitTimer` untouched
and only shortens it to the backoff on a retriable failure.
It also leaves the two protocols behaving differently on assign(): classic
doesn't reset the timer at all, while the async consumer resets it upfront.
Could we avoid the upfront reset in the assign path — either only reset the
timer once the commit completes successfully, or leave the timer alone entirely
to stay consistent with the classic path?
--
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]
