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:
`doAutoCommitAsync()` resets the interval timer upfront, so if this
assign-triggered commit fails with a non-retriable error we delay the next
periodic auto-commit by a full interval without having committed — the same
concern raised earlier on the classic path (and fixed there, where the timer is
left untouched). Should we only reset on success here, or leave the timer alone
to stay consistent with the classic path?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:
##########
@@ -1182,7 +1182,7 @@ public boolean commitOffsetsSync(Map<TopicPartition,
OffsetAndMetadata> offsets,
return false;
}
- private void maybeAutoCommitOffsetsSync(Timer timer) {
+ public void maybeAutoCommitOffsetsSync(Timer timer) {
Review Comment:
should we change it back to `private`?
There is no outside caller.
##########
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);
+ }
+
+ /**
+ * Trigger a best-effort async auto-commit when assign() is called with
new partitions.
+ * Fires once without blocking; the caller does not wait for the result.
+ */
+ public void maybeAutoCommitOnAssignment() {
+ if (autoCommitEnabled()) {
Review Comment:
`maybeAutoCommitOnAssignment()` bypasses `shouldAutoCommit()`, which also
skips the `hasInflightCommit` guard. Skipping the timer check is the point of
this PR, and skipping the inflight guard arguably makes sense too (if we
skipped the commit here, the offsets of the previously-assigned partitions
would be lost). But it breaks the flag's bookkeeping: with two overlapping
auto-commits, the older one completing sets `hasInflightCommit` back to `false`
in `autoCommitCallback()` while the newer one is still in flight, so the
interval-path guard is no longer reliable.
The practical impact is small (requests to the coordinator go over the same
connection, so they stay ordered), but could we either note in the javadoc that
bypassing the inflight guard is intentional, or replace the boolean with an
inflight counter so the guard stays correct?
--
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]
