You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2026/08/20 18:37:14 UTC
(kafka) branch 4.2 updated: KAFKA-20845: Don't flush empty batch when appending large records in group coordinator (4.2) (#23207)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch 4.2
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.2 by this push:
new 4ffb20312f5 KAFKA-20845: Don't flush empty batch when appending large records in group coordinator (4.2) (#23207)
4ffb20312f5 is described below
commit 4ffb20312f5051d42d68bf7413ec112a09d2c1f3
Author: David Jacot <da...@gmail.com>
AuthorDate: Thu Aug 20 20:37:07 2026 +0200
KAFKA-20845: Don't flush empty batch when appending large records in group coordinator (4.2) (#23207)
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.
Backport of #22969 to 4.2. The test is adapted to the
`CoordinatorRuntime` API of this branch.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
Co-authored-by: Sean Quah <sq...@confluent.io>
---
.../common/runtime/CoordinatorRuntime.java | 5 +-
.../common/runtime/CoordinatorRuntimeTest.java | 67 ++++++++++++++++++++++
2 files changed, 70 insertions(+), 2 deletions(-)
diff --git a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java
index 3692060c38a..d176a454bb1 100644
--- a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java
+++ b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java
@@ -828,7 +828,7 @@ public class CoordinatorRuntime<S extends CoordinatorShard<U>, U> implements Aut
if (currentBatch.builder.numRecords() == 0) {
// The only way we can get here is if append() has failed in an unexpected
// way and left an empty batch. Try to clean it up.
- log.debug("Tried to flush an empty batch for {}.", tp);
+ log.warn("Tried to flush an empty batch for {}.", tp);
// There should not be any deferred events attached to the batch. We fail
// the batch just in case. As a side effect, coordinator state is also
// reverted, but there should be no changes since the batch was empty.
@@ -1092,7 +1092,8 @@ public class CoordinatorRuntime<S extends CoordinatorShard<U>, U> implements Aut
recordsToAppend
);
- if (!currentBatch.builder.hasRoomFor(estimatedSizeUpperBound)) {
+ if (currentBatch.builder.numRecords() > 0 &&
+ !currentBatch.builder.hasRoomFor(estimatedSizeUpperBound)) {
// Start a new batch when the total uncompressed data size would exceed
// the max batch size. We still allow atomic writes with an uncompressed size
// larger than the max batch size as long as they compress down to under the max
diff --git a/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java b/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java
index b9cb131f019..e32b0a1ccbe 100644
--- a/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java
+++ b/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java
@@ -5212,6 +5212,73 @@ public class CoordinatorRuntimeTest {
assertEquals("response2", write2.get(5, TimeUnit.SECONDS));
}
+ @Test
+ public void testLargeCompressibleRecordDoesNotFlushEmptyBatch() throws Exception {
+ MockTimer timer = new MockTimer();
+ MockPartitionWriter writer = new MockPartitionWriter();
+ Compression compression = Compression.gzip().build();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(Duration.ofMillis(30))
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(writer)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withCompression(compression)
+ .withSerializer(new StringSerializer())
+ .withAppendLingerMs(OptionalInt.of(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);
+ assertNull(ctx.currentBatch);
+ assertEquals(0, ctx.batchEpoch);
+
+ // Get the max batch size.
+ int maxBatchSize = writer.config(TP).maxMessageSize();
+
+ // Create a large record of highly compressible data.
+ List<String> largeRecord = List.of("a".repeat(3 * maxBatchSize));
+
+ // Write the large record. The record does not fit in an empty batch uncompressed,
+ // but it fits once compressed, so it goes into a batch on its own. The batch is
+ // flushed immediately because it has no room left.
+ long batchTimestamp = timer.time().milliseconds();
+ CompletableFuture<String> write = runtime.scheduleWriteOperation("write#1", TP, Duration.ofMillis(50),
+ state -> new CoordinatorResult<>(largeRecord, "response1")
+ );
+
+ // Verify the state.
+ assertEquals(1L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(
+ new MockCoordinatorShard.RecordAndMetadata(0, largeRecord.get(0))
+ ), ctx.coordinator.coordinator().fullRecords());
+ assertEquals(List.of(
+ records(batchTimestamp, compression, largeRecord)
+ ), writer.entries(TP));
+
+ // KAFKA-20845: Only a single batch must have been created and flushed. When the write is
+ // non-replaying and has made in-memory changes, both the in-memory changes and
+ // the records must be attached to the same batch.
+ assertEquals(1, ctx.batchEpoch);
+
+ // Commit and verify that the write is completed.
+ writer.commit(TP);
+ assertTrue(write.isDone());
+ assertEquals(1L, ctx.coordinator.lastCommittedOffset());
+ assertEquals("response1", write.get(5, TimeUnit.SECONDS));
+ }
+
@Test
public void testLargeCompressibleRecordTriggersFlushAndSucceeds() throws Exception {
MockTimer timer = new MockTimer();