You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "kirktrue (via GitHub)" <gi...@apache.org> on 2025/11/17 18:09:39 UTC
[PR] KAFKA-19588: Reduce waiting for event completion in AsyncKafkaConsumer.poll() [kafka]
kirktrue opened a new pull request, #20363:
URL: https://github.com/apache/kafka/pull/20363
We create—and wait on—PollEvent in Consumer.poll() to ensure we wait for
reconciliation and/or auto-commit. However, reconciliation is relatively
rare, and auto-commit only happens every N seconds, so the remainder of
the time, we should try to avoid sending poll events.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-19588: Reduce number of events generated in AsyncKafkaConsumer.poll() [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #20363:
URL: https://github.com/apache/kafka/pull/20363#discussion_r2291474222
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -859,16 +865,19 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
}
- do {
- PollEvent event = new PollEvent(timer.currentTimeMs());
- // Make sure to let the background thread know that we are still polling.
- // This will trigger async auto-commits of consumed positions when hitting
- // the interval time or reconciling new assignments
- applicationEventHandler.add(event);
+ PollEvent event = new PollEvent(timer.currentTimeMs());
+ // Make sure to let the background thread know that we are still polling.
+ // This will trigger async auto-commits of consumed positions when hitting
+ // the interval time or reconciling new assignments
+ applicationEventHandler.add(event);
+
+ if (reconciliationInProgress.get() || autoCommitState.shouldAutoCommit()) {
Review Comment:
Agreed, and trying to push it a bit more, even with auto-commit enabled, maybe we can still optimise if we check the timer in the app thread, and then use an atomic var "timerExpired" that the background can reuse (instead of checking the timer again).
So we would optimse if auto-commit disabled, or if it's enabled but the timer hasn't expired.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-19588: Reduce number of events generated in AsyncKafkaConsumer.poll() [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #20363:
URL: https://github.com/apache/kafka/pull/20363#issuecomment-3199051616
@lianetm Could you add the `ci-approved` label so that I can see how this runs with GitHub Actions? Thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-19588: Reduce number of events generated in AsyncKafkaConsumer.poll() [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #20363:
URL: https://github.com/apache/kafka/pull/20363#discussion_r2288686114
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -859,16 +865,19 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
}
- do {
- PollEvent event = new PollEvent(timer.currentTimeMs());
- // Make sure to let the background thread know that we are still polling.
- // This will trigger async auto-commits of consumed positions when hitting
- // the interval time or reconciling new assignments
- applicationEventHandler.add(event);
+ PollEvent event = new PollEvent(timer.currentTimeMs());
+ // Make sure to let the background thread know that we are still polling.
+ // This will trigger async auto-commits of consumed positions when hitting
+ // the interval time or reconciling new assignments
+ applicationEventHandler.add(event);
+
+ if (reconciliationInProgress.get() || autoCommitState.shouldAutoCommit()) {
Review Comment:
Couldn't we end up with a race condition here if the app thread sees `autoCommitState.shouldAutoCommit()` false at this point (because interval hasn't expired just yet), but by the time the background checks the same when processing the poll event the interval expired?
In that case, I expect the background would trigger the auto-commit while the app thread moved onto updating positions for fetching (and that leads to a whole new set of race conditions that we already dealt with before). Basically, whatever change we introduce here to not wait on Poll, needs to ensure that we retrieve the positions to commit before moving on to update fetch positions, that's the main challenge with this change I expect. Thinking, but not sure yet how to address that if we don't wait on Poll. Thougts?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19588: Reduce waiting for event completion in AsyncKafkaConsumer.poll() [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue closed pull request #20363: KAFKA-19588: Reduce waiting for event completion in AsyncKafkaConsumer.poll()
URL: https://github.com/apache/kafka/pull/20363
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-19588: Reduce number of events generated in AsyncKafkaConsumer.poll() [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #20363:
URL: https://github.com/apache/kafka/pull/20363#discussion_r2302368414
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -859,16 +865,19 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
}
- do {
- PollEvent event = new PollEvent(timer.currentTimeMs());
- // Make sure to let the background thread know that we are still polling.
- // This will trigger async auto-commits of consumed positions when hitting
- // the interval time or reconciling new assignments
- applicationEventHandler.add(event);
+ PollEvent event = new PollEvent(timer.currentTimeMs());
+ // Make sure to let the background thread know that we are still polling.
+ // This will trigger async auto-commits of consumed positions when hitting
+ // the interval time or reconciling new assignments
+ applicationEventHandler.add(event);
+
+ if (reconciliationInProgress.get() || autoCommitState.shouldAutoCommit()) {
Review Comment:
> Couldn't we end up with a race condition here if the app thread sees `autoCommitState.shouldAutoCommit()` false at this point (because interval hasn't expired just yet), but by the time the background checks the same when processing the poll event the interval expired?
No, simply because nothing updates the timer's sense of the current time between when the application thread submits the `PollEvent` and when it's processed by the network thread.
The application thread determines the timestamp used by the auto-commit timer via `PollEvent` (and `AssignmentChangeEvent`). And it's only checked when `CommitRequestManager.updateTimerAndMaybeCommit()` is invoked by the `ApplicationEventProcessor`. So when we update the timer in the application thread via `AutoCommitState.updateTimer()` and then call `AutoCommitState.shouldAutoCommit()`, we should get the same result in either thread, at least until the auto-commit is actually kicked off.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19588: Reduce waiting for event completion in AsyncKafkaConsumer.poll() [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #20363:
URL: https://github.com/apache/kafka/pull/20363#issuecomment-3267125525
I'm closing this PR as we have decided against going any further with this approach due to thread safety concerns.
cc @lianetm @AndrewJSchofield
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-19588: Reduce number of events generated in AsyncKafkaConsumer.poll() [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #20363:
URL: https://github.com/apache/kafka/pull/20363#issuecomment-3229842281
@chia7712 @frankvicky—would you or someone you know to be familiar with the new `AsyncKafkaConsumer` be willing to help review this PR? Thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19588: Reduce waiting for event completion in AsyncKafkaConsumer.poll() [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue closed pull request #20363: KAFKA-19588: Reduce waiting for event completion in AsyncKafkaConsumer.poll()
URL: https://github.com/apache/kafka/pull/20363
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19588: Reduce number of events generated in AsyncKafkaConsumer.poll() [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #20363:
URL: https://github.com/apache/kafka/pull/20363#discussion_r2308712762
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -140,7 +145,7 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
* assignmentReadyToReconcile. This will be true if {@link #maybeReconcile(boolean)} has been triggered
* after receiving a heartbeat response, or a metadata update.
*/
- private boolean reconciliationInProgress;
+ private final SharedReconciliationState reconciliationState;
Review Comment:
I updated the JavaDoc for `reconciliationState` (and `autoCommitState`) to be more up-to-date and clear. Thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-19588: Reduce number of events generated in AsyncKafkaConsumer.poll() [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #20363:
URL: https://github.com/apache/kafka/pull/20363#discussion_r2290534883
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -859,16 +865,19 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
}
- do {
- PollEvent event = new PollEvent(timer.currentTimeMs());
- // Make sure to let the background thread know that we are still polling.
- // This will trigger async auto-commits of consumed positions when hitting
- // the interval time or reconciling new assignments
- applicationEventHandler.add(event);
+ PollEvent event = new PollEvent(timer.currentTimeMs());
+ // Make sure to let the background thread know that we are still polling.
+ // This will trigger async auto-commits of consumed positions when hitting
+ // the interval time or reconciling new assignments
+ applicationEventHandler.add(event);
+
+ if (reconciliationInProgress.get() || autoCommitState.shouldAutoCommit()) {
Review Comment:
I agree with @lianetm that this is opening up the risk of race conditions. However, I think the principle here is a good one. The risk part here is related to the auto-commit timer. If auto-commit is not enabled, we absolutely know that we are not racing with the auto-commit timer. If it is enabled, we are potentially in a race. So, a slight twist on this can safely optimise when auto-commit is not enabled.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-19588: Reduce number of events generated in AsyncKafkaConsumer.poll() [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #20363:
URL: https://github.com/apache/kafka/pull/20363#discussion_r2306047505
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -140,7 +145,7 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
* assignmentReadyToReconcile. This will be true if {@link #maybeReconcile(boolean)} has been triggered
* after receiving a heartbeat response, or a metadata update.
*/
- private boolean reconciliationInProgress;
+ private final SharedReconciliationState reconciliationState;
Review Comment:
Should we also update the Javadoc of this variable?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-19588: Reduce number of events generated in AsyncKafkaConsumer.poll() [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #20363:
URL: https://github.com/apache/kafka/pull/20363#issuecomment-3229649000
@lianetm @AndrewJSchofield—Could you please add the `ci-approved' label? Thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-19588: Reduce number of events generated in AsyncKafkaConsumer.poll() [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #20363:
URL: https://github.com/apache/kafka/pull/20363#discussion_r2302494786
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AutoCommitState.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Timer;
+import org.slf4j.Logger;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.apache.kafka.clients.consumer.ConsumerConfig.DEFAULT_AUTO_COMMIT_INTERVAL_MS;
+
+/**
+ * Encapsulates the state of auto-committing and manages the auto-commit timer.
+ */
+public interface AutoCommitState {
+
+ /**
+ * @return {@code true} if auto-commit is enabled as defined in the configuration
+ * {@link ConsumerConfig#ENABLE_AUTO_COMMIT_CONFIG}.
+ */
+ boolean isAutoCommitEnabled();
+
+ boolean shouldAutoCommit();
+
+ boolean isExpired();
+
+ /**
+ * Reset the auto-commit timer to the auto-commit interval, so that the next auto-commit is
+ * sent out on the interval starting from now. If auto-commit is disabled this will
+ * perform no action.
+ */
+ void resetTimer();
+
+ /**
+ * Reset the auto-commit timer to the provided time (backoff), so that the next auto-commit is
+ * sent out then. If auto-commit is disabled this will perform no action.
+ */
+ void resetTimer(long retryBackoffMs);
Review Comment:
Done.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AutoCommitState.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Timer;
+import org.slf4j.Logger;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.apache.kafka.clients.consumer.ConsumerConfig.DEFAULT_AUTO_COMMIT_INTERVAL_MS;
+
+/**
+ * Encapsulates the state of auto-committing and manages the auto-commit timer.
+ */
+public interface AutoCommitState {
+
+ /**
+ * @return {@code true} if auto-commit is enabled as defined in the configuration
+ * {@link ConsumerConfig#ENABLE_AUTO_COMMIT_CONFIG}.
+ */
+ boolean isAutoCommitEnabled();
+
+ boolean shouldAutoCommit();
+
+ boolean isExpired();
+
+ /**
+ * Reset the auto-commit timer to the auto-commit interval, so that the next auto-commit is
+ * sent out on the interval starting from now. If auto-commit is disabled this will
+ * perform no action.
+ */
+ void resetTimer();
+
+ /**
+ * Reset the auto-commit timer to the provided time (backoff), so that the next auto-commit is
+ * sent out then. If auto-commit is disabled this will perform no action.
+ */
+ void resetTimer(long retryBackoffMs);
+
+ long remainingMs(final long currentTimeMs);
+
+ void updateTimer(final long currentTimeMs);
+
+ void setInflightCommitStatus(final boolean inflightCommitStatus);
+
+ static AutoCommitState enabled(final LogContext logContext,
+ final Time time,
+ final long autoCommitInterval) {
+ return new AutoCommitStateEnabled(logContext, time, autoCommitInterval);
+ }
+
+ static AutoCommitState enabled(final LogContext logContext, final Time time) {
+ return enabled(logContext, time, DEFAULT_AUTO_COMMIT_INTERVAL_MS);
+ }
+
+ static AutoCommitState disabled() {
+ return new AutoCommitStateDisabled();
+ }
+
+ static AutoCommitState newInstance(final LogContext logContext,
+ final ConsumerConfig config,
+ final Time time) {
+ if (config.getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)) {
+ final long interval = Integer.toUnsignedLong(config.getInt(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG));
+ return enabled(logContext, time, interval);
+ } else {
+ return disabled();
+ }
+ }
+
+ class AutoCommitStateEnabled implements AutoCommitState {
+
+ private final Logger log;
+ private final Timer timer;
+ private final long autoCommitInterval;
+ private final AtomicBoolean hasInflightCommit;
+
+ private AutoCommitStateEnabled(final LogContext logContext,
+ final Time time,
+ final long autoCommitInterval) {
+ this.log = logContext.logger(AutoCommitState.class);
+ this.timer = time.timer(autoCommitInterval);
+ this.autoCommitInterval = autoCommitInterval;
+ this.hasInflightCommit = new AtomicBoolean();
Review Comment:
Yes. 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-19588: Reduce number of events generated in AsyncKafkaConsumer.poll() [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #20363:
URL: https://github.com/apache/kafka/pull/20363#discussion_r2288851118
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AutoCommitState.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Timer;
+import org.slf4j.Logger;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.apache.kafka.clients.consumer.ConsumerConfig.DEFAULT_AUTO_COMMIT_INTERVAL_MS;
+
+/**
+ * Encapsulates the state of auto-committing and manages the auto-commit timer.
+ */
+public interface AutoCommitState {
+
+ /**
+ * @return {@code true} if auto-commit is enabled as defined in the configuration
+ * {@link ConsumerConfig#ENABLE_AUTO_COMMIT_CONFIG}.
+ */
+ boolean isAutoCommitEnabled();
+
+ boolean shouldAutoCommit();
+
+ boolean isExpired();
+
+ /**
+ * Reset the auto-commit timer to the auto-commit interval, so that the next auto-commit is
+ * sent out on the interval starting from now. If auto-commit is disabled this will
+ * perform no action.
+ */
+ void resetTimer();
+
+ /**
+ * Reset the auto-commit timer to the provided time (backoff), so that the next auto-commit is
+ * sent out then. If auto-commit is disabled this will perform no action.
+ */
+ void resetTimer(long retryBackoffMs);
+
+ long remainingMs(final long currentTimeMs);
+
+ void updateTimer(final long currentTimeMs);
+
+ void setInflightCommitStatus(final boolean inflightCommitStatus);
+
+ static AutoCommitState enabled(final LogContext logContext,
+ final Time time,
+ final long autoCommitInterval) {
+ return new AutoCommitStateEnabled(logContext, time, autoCommitInterval);
+ }
+
+ static AutoCommitState enabled(final LogContext logContext, final Time time) {
+ return enabled(logContext, time, DEFAULT_AUTO_COMMIT_INTERVAL_MS);
+ }
+
+ static AutoCommitState disabled() {
+ return new AutoCommitStateDisabled();
+ }
+
+ static AutoCommitState newInstance(final LogContext logContext,
+ final ConsumerConfig config,
+ final Time time) {
+ if (config.getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)) {
+ final long interval = Integer.toUnsignedLong(config.getInt(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG));
+ return enabled(logContext, time, interval);
+ } else {
+ return disabled();
+ }
+ }
+
+ class AutoCommitStateEnabled implements AutoCommitState {
+
+ private final Logger log;
+ private final Timer timer;
+ private final long autoCommitInterval;
+ private final AtomicBoolean hasInflightCommit;
+
+ private AutoCommitStateEnabled(final LogContext logContext,
+ final Time time,
+ final long autoCommitInterval) {
+ this.log = logContext.logger(AutoCommitState.class);
+ this.timer = time.timer(autoCommitInterval);
+ this.autoCommitInterval = autoCommitInterval;
+ this.hasInflightCommit = new AtomicBoolean();
Review Comment:
Doesn't the presence of `synchronized` on all of these methods make the use of `AtomicBoolean` redundant?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AutoCommitState.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Timer;
+import org.slf4j.Logger;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.apache.kafka.clients.consumer.ConsumerConfig.DEFAULT_AUTO_COMMIT_INTERVAL_MS;
+
+/**
+ * Encapsulates the state of auto-committing and manages the auto-commit timer.
+ */
+public interface AutoCommitState {
+
+ /**
+ * @return {@code true} if auto-commit is enabled as defined in the configuration
+ * {@link ConsumerConfig#ENABLE_AUTO_COMMIT_CONFIG}.
+ */
+ boolean isAutoCommitEnabled();
+
+ boolean shouldAutoCommit();
+
+ boolean isExpired();
+
+ /**
+ * Reset the auto-commit timer to the auto-commit interval, so that the next auto-commit is
+ * sent out on the interval starting from now. If auto-commit is disabled this will
+ * perform no action.
+ */
+ void resetTimer();
+
+ /**
+ * Reset the auto-commit timer to the provided time (backoff), so that the next auto-commit is
+ * sent out then. If auto-commit is disabled this will perform no action.
+ */
+ void resetTimer(long retryBackoffMs);
Review Comment:
nit: Why not `final long retryBackoffMs` also?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org