You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "squah-confluent (via GitHub)" <gi...@apache.org> on 2025/10/27 22:00:56 UTC
[PR] KAFKA-19826: Implement coordinator adaptive batch linger time [kafka]
squah-confluent opened a new pull request, #20780:
URL: https://github.com/apache/kafka/pull/20780
Add support for an adaptive batch linger time in the group and share
coordinators. When an adaptive batch linger time is enabled, we no
longer create a timer to flush the current batch. Instead, we append a
flush operation at the end of the event queue so that any currently
queued operations are naturally collected into the batch.
To avoid double flushing from hitting the maximum batch size or
transactional writes, we number batches with an epoch to check whether
the batch has already been flushed.
The group.coordinator.append.linger.ms and
share.coordinator.append.linger.ms configs are extended to allow -1, to
specify an adaptive append linger time. The default for these configs is
also updated to -1.
--
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-19826: Implement coordinator adaptive batch linger time [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #20780:
URL: https://github.com/apache/kafka/pull/20780#discussion_r2528265383
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -300,7 +302,7 @@ public class GroupCoordinatorConfig {
.define(GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, LIST, GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DEFAULT,
ConfigDef.ValidList.in(false, Group.GroupType.documentValidValues()), MEDIUM, GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DOC)
.define(GROUP_COORDINATOR_NUM_THREADS_CONFIG, INT, GROUP_COORDINATOR_NUM_THREADS_DEFAULT, atLeast(1), HIGH, GROUP_COORDINATOR_NUM_THREADS_DOC)
- .define(GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG, INT, GROUP_COORDINATOR_APPEND_LINGER_MS_DEFAULT, atLeast(0), MEDIUM, GROUP_COORDINATOR_APPEND_LINGER_MS_DOC)
+ .define(GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG, INT, GROUP_COORDINATOR_APPEND_LINGER_MS_DEFAULT, atLeast(-1), MEDIUM, GROUP_COORDINATOR_APPEND_LINGER_MS_DOC)
Review Comment:
Thanks for the review. I tested -0.5 and it's already disallowed
```
org.apache.kafka.common.config.ConfigException: Invalid value -0.5 for configuration group.coordinator.append.linger.ms: Not a number of type INT
at app//org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:776)
at app//org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:531)
at app//org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:524)
at app//org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:118)
at app//org.apache.kafka.server.config.AbstractKafkaConfig.<init>(AbstractKafkaConfig.java:78)
```
--
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-19826: Implement coordinator adaptive batch linger time [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #20780:
URL: https://github.com/apache/kafka/pull/20780#discussion_r2469378305
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -220,8 +226,10 @@ public CoordinatorRuntime<S, U> build() {
throw new IllegalArgumentException("Serializer must be set.");
if (compression == null)
compression = Compression.NONE;
- if (appendLingerMs < 0)
- throw new IllegalArgumentException("AppendLinger must be >= 0");
+ if (appendLingerMs == null)
+ appendLingerMs = OptionalInt.empty();
+ if (appendLingerMs.isPresent() && appendLingerMs.getAsInt() < -1)
+ throw new IllegalArgumentException("AppendLinger must be empty or >= 0");
Review Comment:
If the `appendLingerMs` is `-1`, it will be input as `OptionalInt.empty()`. How about checking the value is >= 0 here?
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -220,8 +226,10 @@ public CoordinatorRuntime<S, U> build() {
throw new IllegalArgumentException("Serializer must be set.");
if (compression == null)
compression = Compression.NONE;
- if (appendLingerMs < 0)
- throw new IllegalArgumentException("AppendLinger must be >= 0");
+ if (appendLingerMs == null)
+ appendLingerMs = OptionalInt.empty();
Review Comment:
From the production code, the `appendLingerMs` input cannot be null. How about we set a default value `OptionalInt.empty()` to `appendLingerMs` and remove `appendLingerMs == null` check?
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -186,6 +186,11 @@ public Builder<S, U> withCompression(Compression compression) {
}
public Builder<S, U> withAppendLingerMs(int appendLingerMs) {
+ this.appendLingerMs = OptionalInt.of(appendLingerMs);
+ return this;
+ }
Review Comment:
From the references, this function is only used by test code. Should we remove it and update test cases to use `OptionalInt` one?
--
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-19826: Implement coordinator adaptive batch linger time [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on PR #20780:
URL: https://github.com/apache/kafka/pull/20780#issuecomment-3453509467
The changes are a little long, so it may be easier to review this PR commit by commit, at least for the first 5 commits.
--
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-19826: Implement coordinator adaptive batch linger time [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #20780:
URL: https://github.com/apache/kafka/pull/20780#discussion_r2470862404
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -220,8 +226,10 @@ public CoordinatorRuntime<S, U> build() {
throw new IllegalArgumentException("Serializer must be set.");
if (compression == null)
compression = Compression.NONE;
- if (appendLingerMs < 0)
- throw new IllegalArgumentException("AppendLinger must be >= 0");
+ if (appendLingerMs == null)
+ appendLingerMs = OptionalInt.empty();
+ if (appendLingerMs.isPresent() && appendLingerMs.getAsInt() < -1)
+ throw new IllegalArgumentException("AppendLinger must be empty or >= 0");
Review Comment:
Good catch! The check is supposed to reject -1.
--
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-19826: Implement coordinator adaptive batch linger time [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #20780:
URL: https://github.com/apache/kafka/pull/20780#issuecomment-3533816639
@squah-confluent There are some checkstyle errors. Could you please fix them?
--
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-19826: Implement coordinator adaptive batch linger time [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #20780:
URL: https://github.com/apache/kafka/pull/20780
--
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-19826: Implement coordinator adaptive batch linger time [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #20780:
URL: https://github.com/apache/kafka/pull/20780#discussion_r2527994420
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -300,7 +302,7 @@ public class GroupCoordinatorConfig {
.define(GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, LIST, GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DEFAULT,
ConfigDef.ValidList.in(false, Group.GroupType.documentValidValues()), MEDIUM, GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DOC)
.define(GROUP_COORDINATOR_NUM_THREADS_CONFIG, INT, GROUP_COORDINATOR_NUM_THREADS_DEFAULT, atLeast(1), HIGH, GROUP_COORDINATOR_NUM_THREADS_DOC)
- .define(GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG, INT, GROUP_COORDINATOR_APPEND_LINGER_MS_DEFAULT, atLeast(0), MEDIUM, GROUP_COORDINATOR_APPEND_LINGER_MS_DOC)
+ .define(GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG, INT, GROUP_COORDINATOR_APPEND_LINGER_MS_DEFAULT, atLeast(-1), MEDIUM, GROUP_COORDINATOR_APPEND_LINGER_MS_DOC)
Review Comment:
I wonder whether we could strengthen the validation here because -0.5 is not a valid value for instance. Would it be possible to accept -1 or atLeast(0)?
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorConfig.java:
##########
@@ -89,7 +91,7 @@ public class ShareCoordinatorConfig {
.define(SNAPSHOT_UPDATE_RECORDS_PER_SNAPSHOT_CONFIG, INT, SNAPSHOT_UPDATE_RECORDS_PER_SNAPSHOT_DEFAULT, atLeast(0), MEDIUM, SNAPSHOT_UPDATE_RECORDS_PER_SNAPSHOT_DOC)
.define(LOAD_BUFFER_SIZE_CONFIG, INT, LOAD_BUFFER_SIZE_DEFAULT, atLeast(1), HIGH, LOAD_BUFFER_SIZE_DOC)
.define(STATE_TOPIC_COMPRESSION_CODEC_CONFIG, INT, (int) STATE_TOPIC_COMPRESSION_CODEC_DEFAULT.id, HIGH, STATE_TOPIC_COMPRESSION_CODEC_DOC)
- .define(APPEND_LINGER_MS_CONFIG, INT, APPEND_LINGER_MS_DEFAULT, atLeast(0), MEDIUM, APPEND_LINGER_MS_DOC)
+ .define(APPEND_LINGER_MS_CONFIG, INT, APPEND_LINGER_MS_DEFAULT, atLeast(-1), MEDIUM, APPEND_LINGER_MS_DOC)
Review Comment:
ditto about the validation.
--
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-19826: Implement coordinator adaptive batch linger time [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #20780:
URL: https://github.com/apache/kafka/pull/20780#discussion_r2467188692
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/DirectEventProcessor.java:
##########
@@ -16,31 +16,52 @@
*/
package org.apache.kafka.coordinator.common.runtime;
+import java.util.Deque;
+import java.util.LinkedList;
import java.util.concurrent.RejectedExecutionException;
/**
* A CoordinatorEventProcessor that directly executes the operations. This is
* useful in unit tests where execution in threads is not required.
*/
public class DirectEventProcessor implements CoordinatorEventProcessor {
+ private final Deque<CoordinatorEvent> queue;
+ private boolean inEvent;
+
+ public DirectEventProcessor() {
+ this.queue = new LinkedList<>();
+ this.inEvent = false;
+ }
+
Review Comment:
We can now enqueue flush events while in the middle of another event. The flush event must be run _after_ the current event, so we have to introduce a queue here.
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2615,6 +2615,7 @@ public void testHighWatermarkUpdate() {
.withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
.withCoordinatorMetrics(mock(CoordinatorMetrics.class))
.withSerializer(new StringSerializer())
+ .withAppendLingerMs(0)
Review Comment:
Many of the tests in this file run with an adaptive linger time now, except for those with an explicit linger time.
I updated a handful of tests using the `ManualEventProcessor` to use an append linger time of 0 (preserving the existing behavior) to avoid test churn.
--
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-19826: Implement coordinator adaptive batch linger time [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #20780:
URL: https://github.com/apache/kafka/pull/20780#discussion_r2470864819
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -220,8 +226,10 @@ public CoordinatorRuntime<S, U> build() {
throw new IllegalArgumentException("Serializer must be set.");
if (compression == null)
compression = Compression.NONE;
- if (appendLingerMs < 0)
- throw new IllegalArgumentException("AppendLinger must be >= 0");
+ if (appendLingerMs == null)
+ appendLingerMs = OptionalInt.empty();
Review Comment:
I was following the existing pattern in the builder. We don't set a default value for `logPrefix`, `logContext` and `compression` and initialize them in `build()`.
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -186,6 +186,11 @@ public Builder<S, U> withCompression(Compression compression) {
}
public Builder<S, U> withAppendLingerMs(int appendLingerMs) {
+ this.appendLingerMs = OptionalInt.of(appendLingerMs);
+ return this;
+ }
Review Comment:
Sure, we can remove the method.
--
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-19826: Implement coordinator adaptive batch linger time [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on PR #20780:
URL: https://github.com/apache/kafka/pull/20780#issuecomment-3458238992
@FrankYang0529 Thanks for the review! I updated the code. I also added some tests for the config logic.
--
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