You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "majialoong (via GitHub)" <gi...@apache.org> on 2025/10/21 15:35:27 UTC
[PR] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
majialoong opened a new pull request, #20739:
URL: https://github.com/apache/kafka/pull/20739
This PR fixed the time comparison logic in `CoordinatorRuntime#maybeFlushCurrentBatch` to ensure that the batch is flushed when the elapsed time since `appendTimeMs` exceeds the `appendLingerMs` parameter.
This issue is also mentioned [here]( https://github.com/apache/kafka/pull/20653/files#r2442452104).
--
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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "majialoong (via GitHub)" <gi...@apache.org>.
majialoong commented on code in PR #20739:
URL: https://github.com/apache/kafka/pull/20739#discussion_r2452499210
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -833,7 +833,7 @@ private void flushCurrentBatch() {
*/
private void maybeFlushCurrentBatch(long currentTimeMs) {
if (currentBatch != null) {
- if (currentBatch.builder.isTransactional() || (currentBatch.appendTimeMs - currentTimeMs) >= appendLingerMs || !currentBatch.builder.hasRoomFor(0)) {
+ if (currentBatch.builder.isTransactional() || (currentTimeMs - currentBatch.appendTimeMs) >= appendLingerMs || !currentBatch.builder.hasRoomFor(0)) {
Review Comment:
Thank you all for your feedback and suggestions. I will add a unit test to cover this scenario.
--
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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #20739:
URL: https://github.com/apache/kafka/pull/20739#issuecomment-3441504121
I will merge this tomorrow if @dajac and @AndrewJSchofield have no objections. I will also backport it to 4.1 and 4.0.
--
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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20739:
URL: https://github.com/apache/kafka/pull/20739#discussion_r2449056159
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -833,7 +833,7 @@ private void flushCurrentBatch() {
*/
private void maybeFlushCurrentBatch(long currentTimeMs) {
if (currentBatch != null) {
- if (currentBatch.builder.isTransactional() || (currentBatch.appendTimeMs - currentTimeMs) >= appendLingerMs || !currentBatch.builder.hasRoomFor(0)) {
+ if (currentBatch.builder.isTransactional() || (currentTimeMs - currentBatch.appendTimeMs) >= appendLingerMs || !currentBatch.builder.hasRoomFor(0)) {
Review Comment:
Personally, I would prefer to remove this condition, as a time-based task already exists. This approach would also mean we are not changing the current behavior, since the condition was previously a no-op
--
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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "majialoong (via GitHub)" <gi...@apache.org>.
majialoong commented on code in PR #20739:
URL: https://github.com/apache/kafka/pull/20739#discussion_r2458871231
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -4866,6 +4867,83 @@ public void testCoordinatorExecutor() {
assertTrue(write1.isDone());
}
+ @Test
+ public void testLingerTimeComparisonInMaybeFlushCurrentBatch() throws Exception {
+ // Provides the runtime clock; we will advance it.
+ MockTimer clockTimer = new MockTimer();
+ // Used for scheduling timer tasks; we won't advance it to avoid a timer-triggered batch flush.
+ MockTimer schedulerTimer = new MockTimer();
+
+ MockPartitionWriter writer = new MockPartitionWriter();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(clockTimer.time())
+ .withTimer(schedulerTimer)
+ .withDefaultWriteTimeOut(Duration.ofMillis(20))
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(writer)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(new StringSerializer())
+ .withAppendLingerMs(10)
+ .withExecutorService(mock(ExecutorService.class))
+ .build();
+
+ // Schedule the loading.
+ runtime.scheduleLoadOperation(TP, 10);
+
+ // Verify the initial state.
+ CoordinatorRuntime<MockCoordinatorShard, String>.CoordinatorContext ctx = runtime.contextOrThrow(TP);
+ assertEquals(ACTIVE, ctx.state);
+ assertNull(ctx.currentBatch);
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation(
+ "write#1", TP, Duration.ofMillis(20),
+ state -> new CoordinatorResult<>(List.of("record1"), "response1")
+ );
+ assertFalse(write1.isDone());
+ assertNotNull(ctx.currentBatch);
+ assertEquals(0, writer.entries(TP).size());
+
+ // Verify that the linger timeout task is created; there will also be a default write timeout task.
+ assertEquals(2, schedulerTimer.size());
+
+ // Advance past the linger time.
+ clockTimer.advanceClock(11);
Review Comment:
Yes, the purpose of advancing the `clockTimer` is to ensure that `flushCurrentBatch` is executed when writing #2.
However, I think that after advancing the `clockTimer`, we should check the number of tasks in the `schedulerTimer` to ensure that the linger task still exists (has not been executed or canceled) before writing #2.
This ensures that `flushCurrentBatch` is executed when writing #2, rather than being triggered by the linger timer task.
--
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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "majialoong (via GitHub)" <gi...@apache.org>.
majialoong commented on PR #20739:
URL: https://github.com/apache/kafka/pull/20739#issuecomment-3438247435
Hello @chia7712 and @dajac , I have added a unit test to cover this scenario. When you have time, could you please review this patch again? 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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #20739:
URL: https://github.com/apache/kafka/pull/20739#discussion_r2459582312
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -4866,6 +4867,83 @@ public void testCoordinatorExecutor() {
assertTrue(write1.isDone());
}
+ @Test
+ public void testLingerTimeComparisonInMaybeFlushCurrentBatch() throws Exception {
+ // Provides the runtime clock; we will advance it.
+ MockTimer clockTimer = new MockTimer();
+ // Used for scheduling timer tasks; we won't advance it to avoid a timer-triggered batch flush.
+ MockTimer schedulerTimer = new MockTimer();
+
+ MockPartitionWriter writer = new MockPartitionWriter();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(clockTimer.time())
+ .withTimer(schedulerTimer)
+ .withDefaultWriteTimeOut(Duration.ofMillis(20))
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(writer)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(new StringSerializer())
+ .withAppendLingerMs(10)
+ .withExecutorService(mock(ExecutorService.class))
+ .build();
+
+ // Schedule the loading.
+ runtime.scheduleLoadOperation(TP, 10);
+
+ // Verify the initial state.
+ CoordinatorRuntime<MockCoordinatorShard, String>.CoordinatorContext ctx = runtime.contextOrThrow(TP);
+ assertEquals(ACTIVE, ctx.state);
+ assertNull(ctx.currentBatch);
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation(
+ "write#1", TP, Duration.ofMillis(20),
+ state -> new CoordinatorResult<>(List.of("record1"), "response1")
Review Comment:
ditto.
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -4866,6 +4867,83 @@ public void testCoordinatorExecutor() {
assertTrue(write1.isDone());
}
+ @Test
+ public void testLingerTimeComparisonInMaybeFlushCurrentBatch() throws Exception {
+ // Provides the runtime clock; we will advance it.
+ MockTimer clockTimer = new MockTimer();
+ // Used for scheduling timer tasks; we won't advance it to avoid a timer-triggered batch flush.
+ MockTimer schedulerTimer = new MockTimer();
+
+ MockPartitionWriter writer = new MockPartitionWriter();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
Review Comment:
nit: We use four spaces to indent code.
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -4866,6 +4867,83 @@ public void testCoordinatorExecutor() {
assertTrue(write1.isDone());
}
+ @Test
+ public void testLingerTimeComparisonInMaybeFlushCurrentBatch() throws Exception {
+ // Provides the runtime clock; we will advance it.
+ MockTimer clockTimer = new MockTimer();
+ // Used for scheduling timer tasks; we won't advance it to avoid a timer-triggered batch flush.
+ MockTimer schedulerTimer = new MockTimer();
+
+ MockPartitionWriter writer = new MockPartitionWriter();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(clockTimer.time())
+ .withTimer(schedulerTimer)
+ .withDefaultWriteTimeOut(Duration.ofMillis(20))
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(writer)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(new StringSerializer())
+ .withAppendLingerMs(10)
+ .withExecutorService(mock(ExecutorService.class))
+ .build();
+
+ // Schedule the loading.
+ runtime.scheduleLoadOperation(TP, 10);
+
+ // Verify the initial state.
+ CoordinatorRuntime<MockCoordinatorShard, String>.CoordinatorContext ctx = runtime.contextOrThrow(TP);
+ assertEquals(ACTIVE, ctx.state);
+ assertNull(ctx.currentBatch);
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation(
+ "write#1", TP, Duration.ofMillis(20),
+ state -> new CoordinatorResult<>(List.of("record1"), "response1")
+ );
+ assertFalse(write1.isDone());
+ assertNotNull(ctx.currentBatch);
+ assertEquals(0, writer.entries(TP).size());
+
+ // Verify that the linger timeout task is created; there will also be a default write timeout task.
+ assertEquals(2, schedulerTimer.size());
+
+ // Advance past the linger time.
+ clockTimer.advanceClock(11);
+
+ // At this point, there are still two scheduled tasks; the linger task has not fired
+ // because we did not advance the schedulerTimer.
+ assertEquals(2, schedulerTimer.size());
+
+ // Write #2.
+ CompletableFuture<String> write2 = runtime.scheduleWriteOperation(
+ "write#2", TP, Duration.ofMillis(20),
+ state -> new CoordinatorResult<>(List.of("record2"), "response2")
Review Comment:
ditto.
--
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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #20739:
URL: https://github.com/apache/kafka/pull/20739
--
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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #20739:
URL: https://github.com/apache/kafka/pull/20739#discussion_r2452080446
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -833,7 +833,7 @@ private void flushCurrentBatch() {
*/
private void maybeFlushCurrentBatch(long currentTimeMs) {
if (currentBatch != null) {
- if (currentBatch.builder.isTransactional() || (currentBatch.appendTimeMs - currentTimeMs) >= appendLingerMs || !currentBatch.builder.hasRoomFor(0)) {
+ if (currentBatch.builder.isTransactional() || (currentTimeMs - currentBatch.appendTimeMs) >= appendLingerMs || !currentBatch.builder.hasRoomFor(0)) {
Review Comment:
Good catch! I think that we can keep the condition here. I agree with adding a unit test.
--
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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "majialoong (via GitHub)" <gi...@apache.org>.
majialoong commented on PR #20739:
URL: https://github.com/apache/kafka/pull/20739#issuecomment-3446712738
> @majialoong could you please file a PR for 4.0? I got following error during backport
>
> ```
> > Task :coordinator-common:checkstyleTest
> [ant:checkstyle] [ERROR] /home/chia7712/project/kafka/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:17:1: NCSS for this file is 2,026 (max allowed is 2,000). [JavaNCSS]
>
> > Task :group-coordinator:compileTestJava
> Note: /home/chia7712/project/kafka/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java uses unchecked or unsafe operations.
> Note: Recompile with -Xlint:unchecked for details.
>
> > Task :coordinator-common:checkstyleTest FAILED
> [ant:checkstyle] [ERROR] /home/chia7712/project/kafka/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:110:1: Class Fan-Out Complexity is 53 (max allowed is 52). [ClassFanOutComplexity]
> ```
Hi @chia7712 . These errors are caused by code differences between the 4.0 branch and trunk. I have submitted a PR ([20773](https://github.com/apache/kafka/pull/20773)) to backport this fix.
--
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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #20739:
URL: https://github.com/apache/kafka/pull/20739#issuecomment-3432155317
> Hello @AndrewJSchofield and @dajac , if you have time, could you please take a look at this issue? Thank you very much, and I look forward to your suggestions.
This is much more in @dajac's area than mine.
--
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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20739:
URL: https://github.com/apache/kafka/pull/20739#discussion_r2457085871
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -4866,6 +4867,83 @@ public void testCoordinatorExecutor() {
assertTrue(write1.isDone());
}
+ @Test
+ public void testLingerTimeComparisonInMaybeFlushCurrentBatch() throws Exception {
+ // Provides the runtime clock; we will advance it.
+ MockTimer clockTimer = new MockTimer();
+ // Used for scheduling timer tasks; we won't advance it to avoid a timer-triggered batch flush.
+ MockTimer schedulerTimer = new MockTimer();
+
+ MockPartitionWriter writer = new MockPartitionWriter();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(clockTimer.time())
+ .withTimer(schedulerTimer)
+ .withDefaultWriteTimeOut(Duration.ofMillis(20))
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(writer)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(new StringSerializer())
+ .withAppendLingerMs(10)
+ .withExecutorService(mock(ExecutorService.class))
+ .build();
+
+ // Schedule the loading.
+ runtime.scheduleLoadOperation(TP, 10);
+
+ // Verify the initial state.
+ CoordinatorRuntime<MockCoordinatorShard, String>.CoordinatorContext ctx = runtime.contextOrThrow(TP);
+ assertEquals(ACTIVE, ctx.state);
+ assertNull(ctx.currentBatch);
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation(
+ "write#1", TP, Duration.ofMillis(20),
+ state -> new CoordinatorResult<>(List.of("record1"), "response1")
+ );
+ assertFalse(write1.isDone());
+ assertNotNull(ctx.currentBatch);
+ assertEquals(0, writer.entries(TP).size());
+
+ // Verify that the linger timeout task is created; there will also be a default write timeout task.
+ assertEquals(2, schedulerTimer.size());
+
+ // Advance past the linger time.
+ clockTimer.advanceClock(11);
Review Comment:
Could you move this to line#4921? It makes more sense there, since the goal of advancing the `clockTimer` is to ensure `flushCurrentBatch` is executed during writing #2, right?
--
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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #20739:
URL: https://github.com/apache/kafka/pull/20739#issuecomment-3444343093
@majialoong could you please file a PR for 4.0? I got following error during backport
```
> Task :coordinator-common:checkstyleTest
[ant:checkstyle] [ERROR] /home/chia7712/project/kafka/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:17:1: NCSS for this file is 2,026 (max allowed is 2,000). [JavaNCSS]
> Task :group-coordinator:compileTestJava
Note: /home/chia7712/project/kafka/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java uses unchecked or unsafe operations.
Note: Recompile with -Xlint:unchecked for details.
> Task :coordinator-common:checkstyleTest FAILED
[ant:checkstyle] [ERROR] /home/chia7712/project/kafka/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:110:1: Class Fan-Out Complexity is 53 (max allowed is 52). [ClassFanOutComplexity]
```
--
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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "majialoong (via GitHub)" <gi...@apache.org>.
majialoong commented on code in PR #20739:
URL: https://github.com/apache/kafka/pull/20739#discussion_r2460929902
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -4866,6 +4867,83 @@ public void testCoordinatorExecutor() {
assertTrue(write1.isDone());
}
+ @Test
+ public void testLingerTimeComparisonInMaybeFlushCurrentBatch() throws Exception {
+ // Provides the runtime clock; we will advance it.
+ MockTimer clockTimer = new MockTimer();
+ // Used for scheduling timer tasks; we won't advance it to avoid a timer-triggered batch flush.
+ MockTimer schedulerTimer = new MockTimer();
+
+ MockPartitionWriter writer = new MockPartitionWriter();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
Review Comment:
Thanks! I’ve addressed this.
--
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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "majialoong (via GitHub)" <gi...@apache.org>.
majialoong commented on code in PR #20739:
URL: https://github.com/apache/kafka/pull/20739#discussion_r2450249209
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -833,7 +833,7 @@ private void flushCurrentBatch() {
*/
private void maybeFlushCurrentBatch(long currentTimeMs) {
if (currentBatch != null) {
- if (currentBatch.builder.isTransactional() || (currentBatch.appendTimeMs - currentTimeMs) >= appendLingerMs || !currentBatch.builder.hasRoomFor(0)) {
+ if (currentBatch.builder.isTransactional() || (currentTimeMs - currentBatch.appendTimeMs) >= appendLingerMs || !currentBatch.builder.hasRoomFor(0)) {
Review Comment:
As you mentioned, time-based scheduled tasks will ensure that the current batch is flushed to the log once it has passed the append linger time.
Perhaps performing a conditional check here is intended to detect as promptly as possible when the current batch has passed the append linger time, so that flushing to the log can be carried out more promptly.
--
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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "majialoong (via GitHub)" <gi...@apache.org>.
majialoong commented on PR #20739:
URL: https://github.com/apache/kafka/pull/20739#issuecomment-3430274905
Hello @AndrewJSchofield and @dajac , if you have time, could you please take a look at this issue? Thank you very much, and I look forward to your suggestions.
--
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] MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20739:
URL: https://github.com/apache/kafka/pull/20739#discussion_r2450445429
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -833,7 +833,7 @@ private void flushCurrentBatch() {
*/
private void maybeFlushCurrentBatch(long currentTimeMs) {
if (currentBatch != null) {
- if (currentBatch.builder.isTransactional() || (currentBatch.appendTimeMs - currentTimeMs) >= appendLingerMs || !currentBatch.builder.hasRoomFor(0)) {
+ if (currentBatch.builder.isTransactional() || (currentTimeMs - currentBatch.appendTimeMs) >= appendLingerMs || !currentBatch.builder.hasRoomFor(0)) {
Review Comment:
Could you add a unit test here? Since the existing tests don't update the mock time, this bug was not disclosed
--
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