MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch by majialoong · Pull Request #20739 · apache/kafka

majialoong · 2025-10-21T15:35:25Z

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.

Reviewers: David Jacot djacot@confluent.io, Chia-Ping Tsai
chia7712@gmail.com

…maybeFlushCurrentBatch

chia7712 · 2025-10-21T16:58:24Z

...tor-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java

        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)) {


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

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.

Could you add a unit test here? Since the existing tests don't update the mock time, this bug was not disclosed

Good catch! I think that we can keep the condition here. I agree with adding a unit test.

Thank you all for your feedback and suggestions. I will add a unit test to cover this scenario.

majialoong · 2025-10-22T02:36:34Z

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.

AndrewJSchofield · 2025-10-22T12:38:02Z

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.

majialoong · 2025-10-23T17:29:25Z

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 !

chia7712

@majialoong thanks for this patch. overall LGTM.

chia7712 · 2025-10-23T20:04:01Z

...common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java

+        assertEquals(2, schedulerTimer.size());
+
+        // Advance past the linger time.
+        clockTimer.advanceClock(11);


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?

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.

chia7712

LGTM

chia7712 · 2025-10-24T07:25:35Z

I will merge this tomorrow if @dajac and @AndrewJSchofield have no objections. I will also backport it to 4.1 and 4.0.

dajac

Thanks for the fix, @majialoong! I left a few nits. Otherwise, LGTM.

dajac · 2025-10-24T09:48:18Z

...common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java

+        MockPartitionWriter writer = new MockPartitionWriter();
+
+        CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+                new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()


nit: We use four spaces to indent code.

Thanks! I’ve addressed this.

dajac · 2025-10-24T09:48:24Z

...common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java

+                "write#1", TP, Duration.ofMillis(20),
+                state -> new CoordinatorResult<>(List.of("record1"), "response1")


dajac · 2025-10-24T09:48:32Z

...common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java

+                "write#2", TP, Duration.ofMillis(20),
+                state -> new CoordinatorResult<>(List.of("record2"), "response2")


dajac

lgtm, thanks

…maybeFlushCurrentBatch (#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). Reviewers: David Jacot <djacot@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>

chia7712 · 2025-10-24T18:15:59Z

@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]

majialoong · 2025-10-25T13:36:43Z

@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) to backport this fix.

…maybeFlushCurrentBatch (apache#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). Reviewers: David Jacot <djacot@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>

…maybeFlushCurrentBatch (#20773) 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). The fix for this issue was originally in [this PR](#20739) in the trunk branch, which was backported to the 4.0 branch. Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

…maybeFlushCurrentBatch (apache#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). Reviewers: David Jacot <djacot@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>

MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#…

7bfbe9b

…maybeFlushCurrentBatch

github-actions bot added triage PRs from the community group-coordinator small Small PRs labels Oct 21, 2025

chia7712 added the ci-approved label Oct 21, 2025

chia7712 mentioned this pull request Oct 21, 2025

KAFKA-19760: RecordTooLargeExceptions in group coordinator when offsets.topic.compression.codec is used #20653

Merged

chia7712 reviewed Oct 21, 2025

View reviewed changes

chia7712 requested review from AndrewJSchofield and dajac October 21, 2025 16:58

github-actions bot removed the triage PRs from the community label Oct 22, 2025

AndrewJSchofield removed their request for review October 23, 2025 10:09

Add unit test.

90c3f59

chia7712 reviewed Oct 23, 2025

View reviewed changes

chia7712 approved these changes Oct 24, 2025

View reviewed changes

dajac reviewed Oct 24, 2025

View reviewed changes

Adjust code style.

69963c6

dajac approved these changes Oct 24, 2025

View reviewed changes

chia7712 merged commit 82a1935 into apache:trunk Oct 24, 2025
24 checks passed

majialoong deleted the fix_time_comparison branch October 25, 2025 06:50

majialoong mentioned this pull request Oct 25, 2025

MINOR: Fix time comparison with appendLingerMs in CoordinatorRuntime#maybeFlushCurrentBatch #20773

Merged

		"write#1", TP, Duration.ofMillis(20),
		state -> new CoordinatorResult<>(List.of("record1"), "response1")

		"write#2", TP, Duration.ofMillis(20),
		state -> new CoordinatorResult<>(List.of("record2"), "response2")

Conversation

majialoong commented Oct 21, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

majialoong commented Oct 22, 2025

Uh oh!

AndrewJSchofield commented Oct 22, 2025

Uh oh!

majialoong commented Oct 23, 2025

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

chia7712 commented Oct 24, 2025

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

chia7712 commented Oct 24, 2025

Uh oh!

majialoong commented Oct 25, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

majialoong commented Oct 21, 2025 •

edited by github-actions bot

Loading