KAFKA-20845: Don't flush empty batch when appending large records in group coordinator by squah-confluent · Pull Request #22969 · apache/kafka

squah-confluent · 2026-07-27T21:13:56Z

When appending records, we create a new batch if no batch exists.
KAFKA-19760 introduced a flush of any existing batch when appending
large records, to maximize our chances of compressing the records under
the max.message.bytes. When these two things happen in the same append
operation, we flush an empty batch.

Flushing an empty batch is written to fail and revert the coordinator
state. This would be harmless, except some group coordinator operations
update the coordinator state directly without replay. Upon appending
their records and triggering the empty batch flush, their state changes
are then reverted while their records are written. The group
coordinator's in-memory state diverges from the on-disk state and
subsequent writes for the group can be invalid for the on-disk state.
eg. we may have a consumer group downgraded to a classic group, followed
by consumer group records which is invalid.

Do not flush empty batches when appending large records.

Reviewers: nileshkumar3 nileshkumar3@gmail.com, Andrew Schofield
aschofield@confluent.io, Dongnuo Lyu dlyu@confluent.io, David Jacot
david.jacot@gmail.com

…group coordinator When appending records, we create a new batch if no batch exists. We also flush any existing batch when appending large records to maximize our chances of compressing the records under the max.message.bytes. When these two things happen in the same append operation, we flush an empty batch. Flushing an empty batch is written to fail and revert the coordinator state. This would be harmless, except some group coordinator operations update the coordinator state directly without replay. Upon appending their records and triggering the empty batch flush, their state changes are then reverted while their records are written. The group coordinator's in-memory state diverges from the on-disk state and subsequent writes for the group can be invalid for the on-disk state. eg. we may have a consumer group downgraded to a classic group, followed by consumer group records which is invalid. Do not flush empty batches when appending large records.

nileshkumar3 · 2026-07-29T03:24:33Z

    }

+    @Test
+    public void testLargeCompressibleRecordSucceeds() throws Exception {


can we add regression test for a large uncompressible record when the current batch was just allocated and is still empty?

I'm not 100% on board with this because that's a case where we have already run into a bug. In normal operation there should never be an empty batch lying around. I'm trying to keep KAFKA-20845 scoped to the happy path - the unhappy path is broken for non-replaying operations in a few ways*

* Off the top of my head, these include:

A non-replaying operation that throws an exception midway through will leave group coordinator state partially updated with no records written.

Same for an exception midway through append(), except we may leave an empty batch lying around. It's wrong to continue using that batch and commit it, which we may currently do (even before this PR).

nileshkumar3 · 2026-07-29T03:25:15Z


-                    if (!currentBatch.builder.hasRoomFor(estimatedSizeUpperBound)) {
+                    if (!currentBatch.builder.hasRoomFor(estimatedSizeUpperBound) &&
+                        currentBatch.builder.numRecords() > 0) {


Optional: consider reordering this condition to currentBatch.builder.numRecords() > 0 && !currentBatch.builder.hasRoomFor(estimatedSizeUpperBound)

Sure! That check is cheaper so it makes sense to do it first.

AndrewJSchofield · 2026-07-31T12:40:08Z

This merits backporting to 4.2 and 4.3 I think.

dongnuo123 · 2026-07-31T20:34:46Z

Could we make it on info level?

Yes, good idea. I think it should be warn level since it's supposed to be impossible unless we had a previous exception or bug in the append logic.

dongnuo123 · 2026-07-31T20:35:37Z

                        recordsToAppend
                    );

-                    if (!currentBatch.builder.hasRoomFor(estimatedSizeUpperBound)) {


I believe this is the only place where downgrade could possibly cause flushCurrentBatch

dajac · 2026-08-07T08:49:42Z

I wonder whether we should also change this. The assumption is kind of wrong as in-mem changes could have been made so reverting is not alway correct. Should we just free to batch if the batch is empty?

My concern is that we have many other paths calling flushCurrentBatch() so I am not sure whether gating the flush in append in enough to cover all the possible cases. It is good to have it for sure but it may not be enough.

@squah-confluent @dongnuo123 What do you think?

Yeah i think it makes sense. The only concern might be that if we really have another path that can result in the same bug with flushing empty batch, directly freeing the batch might cause divergence between the memory and logs.

eg the scenario we wanted to fix with this if (currentBatch.builder.numRecords() == 0) { branch. If it's a downgrade and the in-mem state has been updated and serializeKey/serializeValue throws, the exception is not caught, no record is written so we could end up in an empty batch with in-mem change. In this case we still need the fail the batch to revert the state.

Also I wonder what if we don't end up with an empty batch because there's already something in the batch when the serialization error happens? Looks like no revert will be triggered and we will have memory changed while no changes in log?

The comment is indeed wrong and needs updating. Both succeeding and reverting an empty batch are incorrect in different ways. Basically we need state changes from non-replaying writes to be applied or reverted atomically with the records. If we have an empty batch with state changes attached it means we have separated them and that's wrong.

Regarding other paths:
There aren't any other happy paths that flush an empty batch. The non-atomic flush later in the method will never do it because hasRoomFor returns true when the batch is empty. Even without that, it's invalid to have a non-replaying non-atomic write anyway, since we can't revert the correct part of the in-memory updates when the write fails partially.

If we start considering unhappy paths, that's a huge can of worms. The only way we can have an empty batch after an append() call is if it threw. Both succeeding and failing an empty batch are incorrect in different situations. If the empty batch is from a previous failed non-replaying write, succeeding the batch would keep its in-memory changes without writing corresponding records. Arguably we should add exception handling to append() to guarantee that it never leaves an empty batch.

tl;dr I'm not keen on this right now because it doesn't really fix the unhappy paths and I want to avoid scope creeping this PR.

To close all the unhappy paths, ideally we would:

Add exception handling to append() to never leave empty batches

which lets us clean up flushCurrentBatch() to succeed empty batches, ie. handle 0-record batches the same way as 1-record batches.

Add exception handling to CoordinatorWriteEvent to revert in-memory changes when an exception is thrown before appending records.

And even then, large non-replaying writes would still be bugged, since the in-memory changes would be attached to any existing batch, but the records would go in the next batch.

(Filed as KAFKA-20912)

dajac · 2026-08-07T08:55:32Z

    }

+    @Test
+    public void testLargeCompressibleRecordSucceeds() throws Exception {


The name of the test does not fully represent the bug in my opinion. Could we come up with a better one?

Thanks for reviewing, renamed it to testLargeCompressibleRecordDoesNotFlushEmptyBatch

github-actions Bot added group-coordinator small Small PRs triage PRs from the community labels Jul 27, 2026

nileshkumar3 reviewed Jul 29, 2026

View reviewed changes

github-actions Bot removed the triage PRs from the community label Jul 29, 2026

AndrewJSchofield requested a review from dajac July 31, 2026 08:24

dongnuo123 reviewed Jul 31, 2026

View reviewed changes

squah-confluent added 2 commits August 5, 2026 15:20

fixup: log empty batch flushes at warn level

6ea6adb

fixup: swap condition evaluation order

a2def8f

dajac reviewed Aug 7, 2026

View reviewed changes

squah-confluent added 3 commits August 10, 2026 14:54

fixup: rename new test

faaa90c

fixup: throw when failing empty batch

231b1e6

fixup: update comment in new test to add explanation

bae4157

github-actions Bot removed the small Small PRs label Aug 10, 2026

Navigation Menu

Uh oh!

KAFKA-20845: Don't flush empty batch when appending large records in group coordinator - #22969

KAFKA-20845: Don't flush empty batch when appending large records in group coordinator#22969
squah-confluent wants to merge 6 commits into
apache:trunkfrom
confluentinc:squah-fix-empty-batch-flush-on-large-append

squah-confluent commented Jul 27, 2026 •

edited by github-actions Bot

Loading

Uh oh!

nileshkumar3 Jul 29, 2026

Uh oh!

squah-confluent Aug 6, 2026

Uh oh!

nileshkumar3 Jul 29, 2026 •

edited

Loading

Uh oh!

squah-confluent Aug 6, 2026

Uh oh!

AndrewJSchofield commented Jul 31, 2026

Uh oh!

dongnuo123 Jul 31, 2026

Uh oh!

squah-confluent Aug 6, 2026

Uh oh!

dongnuo123 Jul 31, 2026

Uh oh!

dajac Aug 7, 2026

Uh oh!

dongnuo123 Aug 7, 2026

Uh oh!

squah-confluent Aug 10, 2026 •

edited

Loading

Uh oh!

dajac Aug 7, 2026

Uh oh!

squah-confluent Aug 10, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

Uh oh!

Conversation

squah-confluent commented Jul 27, 2026 • 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!

nileshkumar3 Jul 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield commented Jul 31, 2026

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!

squah-confluent Aug 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

squah-confluent commented Jul 27, 2026 •

edited by github-actions Bot

Loading

nileshkumar3 Jul 29, 2026 •

edited

Loading

squah-confluent Aug 10, 2026 •

edited

Loading