chia7712 commented on code in PR #23208:
URL: https://github.com/apache/kafka/pull/23208#discussion_r3824500133
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -4463,6 +4463,73 @@ public void
testCompressibleRecordTriggersFlushAndSucceeds() throws Exception {
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(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);
+
+ // 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 with a non-replaying write operation which
applies its
+ // in-memory changes directly. 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 -> {
+ state.replay(0, RecordBatch.NO_PRODUCER_ID,
RecordBatch.NO_PRODUCER_EPOCH, largeRecord.get(0));
+ return new CoordinatorResult<>(largeRecord, "response1", null,
false);
+ }
+ );
+
+ // Verify the state. Only a single batch must have been created and
flushed so the
Review Comment:
Clever way to test this on the backport branch
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
