unknowntpo commented on code in PR #21584:
URL: https://github.com/apache/kafka/pull/21584#discussion_r3912015282
##########
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:
Your right, I should respect the flag, I've changed the flag to the counter
to represent multiple inflight auto commits.
##########
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:
fixed.
--
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]
