You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mingyen066 (via GitHub)" <gi...@apache.org> on 2025/06/25 20:59:29 UTC
[PR] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
mingyen066 opened a new pull request, #20040:
URL: https://github.com/apache/kafka/pull/20040
Delete this text and replace it with a detailed description of your change. The
PR title and body will become the squashed commit message.
If you would like to tag individuals, add some commentary, upload images, or
include other supplemental information that should not be part of the eventual
commit message, please use a separate comment.
If applicable, please include a summary of the testing strategy (including
rationale) for the proposed change. Unit and/or integration tests are expected
for any behavior change and system tests should be considered for larger
changes.
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2194771745
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2937,9 +2939,206 @@ public void testAppendRecordBatchSize() {
assertFalse(write1.isCompletedExceptionally());
int batchSize = writer.entries(TP).get(0).sizeInBytes();
- assertTrue(batchSize > MIN_BUFFER_SIZE && batchSize < maxBatchSize);
+ assertTrue(batchSize > INITIAL_BUFFER_SIZE && batchSize < maxBatchSize);
}
+ @Test
+ public void testCoordinatorDoNotRetainBufferLargeThanMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate a record larger than the maxBatchSize.
+ List<String> largeRecords = List.of("A".repeat(100 * 1024 * 1024));
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(largeRecords, "response1", null, true, false)
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the initial small one, not the large buffer.
+ assertEquals(INITIAL_BUFFER_SIZE, ctx.bufferSupplier.get(1).capacity());
+ }
+
+ @Test
+ public void testCoordinatorRetainExpandedBufferLessOrEqualToMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024 * 1024) // 1GB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate enough records to create a batch that has INITIAL_BUFFER_SIZE < batchSize < maxBatchSize
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the expanded buffer.
+ assertTrue(ctx.bufferSupplier.get(1).capacity() > INITIAL_BUFFER_SIZE);
+ }
+
+ @Test
+ public void testBufferShrinkWhenMaxMessageSizeReducedBelowInitialBufferSize() {
+ MockTimer timer = new MockTimer();
+ var mockWriter = new InMemoryPartitionWriter(false) {
+ private LogConfig config = new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return config;
+ }
+
+ public void updateConfig(LogConfig newConfig) {
+ this.config = newConfig;
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
Review Comment:
ditto
```suggestion
int batchSize = writer.entries(TP).get(0).sizeInBytes();
assertTrue(records <= batchSize);
```
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "mingyen066 (via GitHub)" <gi...@apache.org>.
mingyen066 commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2173250668
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2940,6 +2942,135 @@ public void testAppendRecordBatchSize() {
assertTrue(batchSize > MIN_BUFFER_SIZE && batchSize < maxBatchSize);
}
+ @Test
+ public void testCoordinatorDoNotRetainLargeBuffer() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ var props = new Properties();
+ props.put(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024)); // 1MB
+ return new LogConfig(props);
+ }
+
+ @Override
+ public long append(
+ TopicPartition tp,
+ VerificationGuard verificationGuard,
+ MemoryRecords batch
+ ) {
+ return super.append(tp, verificationGuard, batch);
+ }
Review Comment:
My mistake — I meant to use MockPartitionWriter, but after replacing it with InMemoryPartitionWriter, I forgot to clean it up.
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2194534919
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2937,9 +2939,206 @@ public void testAppendRecordBatchSize() {
assertFalse(write1.isCompletedExceptionally());
int batchSize = writer.entries(TP).get(0).sizeInBytes();
- assertTrue(batchSize > MIN_BUFFER_SIZE && batchSize < maxBatchSize);
+ assertTrue(batchSize > INITIAL_BUFFER_SIZE && batchSize < maxBatchSize);
}
+ @Test
+ public void testCoordinatorDoNotRetainBufferLargeThanMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate a record larger than the maxBatchSize.
+ List<String> largeRecords = List.of("A".repeat(100 * 1024 * 1024));
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(largeRecords, "response1", null, true, false)
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the initial small one, not the large buffer.
+ assertEquals(INITIAL_BUFFER_SIZE, ctx.bufferSupplier.get(1).capacity());
+ }
+
+ @Test
+ public void testCoordinatorRetainExpandedBufferLessOrEqualToMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024 * 1024) // 1GB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate enough records to create a batch that has INITIAL_BUFFER_SIZE < batchSize < maxBatchSize
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the expanded buffer.
+ assertTrue(ctx.bufferSupplier.get(1).capacity() > INITIAL_BUFFER_SIZE);
+ }
+
+ @Test
+ public void testBufferShrinkWhenMaxMessageSizeReducedBelowInitialBufferSize() {
+ MockTimer timer = new MockTimer();
+ var mockWriter = new InMemoryPartitionWriter(false) {
+ private LogConfig config = new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return config;
+ }
+
+ public void updateConfig(LogConfig newConfig) {
+ this.config = newConfig;
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ ByteBuffer cachedBuffer = ctx.bufferSupplier.get(1);
+ assertEquals(INITIAL_BUFFER_SIZE, cachedBuffer.capacity());
+ ctx.bufferSupplier.release(cachedBuffer);
+
+ // Reduce max message size below initial buffer size.
+ mockWriter.updateConfig(new LogConfig(
+ Map.of(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(INITIAL_BUFFER_SIZE - 66))));
+ assertEquals(INITIAL_BUFFER_SIZE - 66, mockWriter.config(TP).maxMessageSize());
+ assertTrue(mockWriter.config(TP).maxMessageSize() < INITIAL_BUFFER_SIZE);
Review Comment:
This line is already covered by the test right above.
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "mingyen066 (via GitHub)" <gi...@apache.org>.
mingyen066 commented on PR #20040:
URL: https://github.com/apache/kafka/pull/20040#issuecomment-3015671445
> @mingyen066 this PR also fix the issue that decreasing the message size dynamically does not re-create the buffer.
Hence, could you please add test for the scenario?
Thanks @chia7712 , I've added a test to cover the scenario where `max.message.bytes` is dynamically reduced.
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "mingyen066 (via GitHub)" <gi...@apache.org>.
mingyen066 commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2173383416
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -859,7 +866,7 @@ private void maybeAllocateNewBatch(
LogConfig logConfig = partitionWriter.config(tp);
int maxBatchSize = logConfig.maxMessageSize();
long prevLastWrittenOffset = coordinator.lastWrittenOffset();
- ByteBuffer buffer = bufferSupplier.get(maxBatchSize);
+ ByteBuffer buffer = bufferSupplier.get(min(MIN_BUFFER_SIZE, maxBatchSize));
Review Comment:
Good suggestion! I’ve renamed MIN_BUFFER_SIZE to INITIAL_BUFFER_SIZE.
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2167660567
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -627,6 +637,8 @@ private CoordinatorContext(
defaultWriteTimeout
);
this.bufferSupplier = new BufferSupplier.GrowableBufferSupplier();
+ this.minBufferSize = 16384; // 16KB
+ this.maxBatchSize = partitionWriter.config(tp).maxMessageSize();
Review Comment:
the config is dynamic, 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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2188507749
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2937,9 +2939,206 @@ public void testAppendRecordBatchSize() {
assertFalse(write1.isCompletedExceptionally());
int batchSize = writer.entries(TP).get(0).sizeInBytes();
- assertTrue(batchSize > MIN_BUFFER_SIZE && batchSize < maxBatchSize);
+ assertTrue(batchSize > INITIAL_BUFFER_SIZE && batchSize < maxBatchSize);
}
+ @Test
+ public void testCoordinatorDoNotRetainBufferLargeThanMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate a record larger than the maxBatchSize.
+ List<String> largeRecords = List.of("A".repeat(100 * 1024 * 1024));
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(largeRecords, "response1", null, true, false)
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the initial small one, not the large buffer.
+ assertEquals(INITIAL_BUFFER_SIZE, ctx.bufferSupplier.get(1).capacity());
+ }
+
+ @Test
+ public void testCoordinatorRetainExpandedBufferLessOrEqualToMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024 * 1024) // 1GB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate enough records to create a batch that has INITIAL_BUFFER_SIZE < batchSize < maxBatchSize
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the expanded buffer.
+ assertTrue(ctx.bufferSupplier.get(1).capacity() > INITIAL_BUFFER_SIZE);
+ }
+
+ @Test
+ public void testBufferShrinkWhenMaxMessageSizeReducedBelowInitialBufferSize() {
+ MockTimer timer = new MockTimer();
+ var mockWriter = new InMemoryPartitionWriter(false) {
+ private LogConfig config = new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));;
Review Comment:
`;;` -> `;`
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2937,9 +2939,206 @@ public void testAppendRecordBatchSize() {
assertFalse(write1.isCompletedExceptionally());
int batchSize = writer.entries(TP).get(0).sizeInBytes();
- assertTrue(batchSize > MIN_BUFFER_SIZE && batchSize < maxBatchSize);
+ assertTrue(batchSize > INITIAL_BUFFER_SIZE && batchSize < maxBatchSize);
}
+ @Test
+ public void testCoordinatorDoNotRetainBufferLargeThanMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate a record larger than the maxBatchSize.
+ List<String> largeRecords = List.of("A".repeat(100 * 1024 * 1024));
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(largeRecords, "response1", null, true, false)
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the initial small one, not the large buffer.
+ assertEquals(INITIAL_BUFFER_SIZE, ctx.bufferSupplier.get(1).capacity());
+ }
+
+ @Test
+ public void testCoordinatorRetainExpandedBufferLessOrEqualToMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024 * 1024) // 1GB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate enough records to create a batch that has INITIAL_BUFFER_SIZE < batchSize < maxBatchSize
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the expanded buffer.
+ assertTrue(ctx.bufferSupplier.get(1).capacity() > INITIAL_BUFFER_SIZE);
+ }
+
+ @Test
+ public void testBufferShrinkWhenMaxMessageSizeReducedBelowInitialBufferSize() {
+ MockTimer timer = new MockTimer();
+ var mockWriter = new InMemoryPartitionWriter(false) {
+ private LogConfig config = new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));;
+
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return config;
+ }
+
+ public void updateConfig(LogConfig newConfig) {
+ this.config = newConfig;
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ ByteBuffer cachedBuffer = ctx.bufferSupplier.get(1);
+ assertEquals(INITIAL_BUFFER_SIZE, cachedBuffer.capacity());
+ ctx.bufferSupplier.release(cachedBuffer);
Review Comment:
what is it's purpose?
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2173285324
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -859,7 +866,7 @@ private void maybeAllocateNewBatch(
LogConfig logConfig = partitionWriter.config(tp);
int maxBatchSize = logConfig.maxMessageSize();
long prevLastWrittenOffset = coordinator.lastWrittenOffset();
- ByteBuffer buffer = bufferSupplier.get(maxBatchSize);
+ ByteBuffer buffer = bufferSupplier.get(min(MIN_BUFFER_SIZE, maxBatchSize));
Review Comment:
If so, since we allow users to assign `maxBatchSize` smaller than `MIN_BUFFER_SIZE`. `MIN_BUFFER_SIZE` variable name does not match its semantics. Maybe rename it to "INITIAL_BUFFER_SIZE" or something similar.
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2173285324
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -859,7 +866,7 @@ private void maybeAllocateNewBatch(
LogConfig logConfig = partitionWriter.config(tp);
int maxBatchSize = logConfig.maxMessageSize();
long prevLastWrittenOffset = coordinator.lastWrittenOffset();
- ByteBuffer buffer = bufferSupplier.get(maxBatchSize);
+ ByteBuffer buffer = bufferSupplier.get(min(MIN_BUFFER_SIZE, maxBatchSize));
Review Comment:
If so, since we allow users to assign `maxBatchSize` smaller than `MIN_BUFFER_SIZE`. `MIN_BUFFER_SIZE` variable name does not match its semantics. Maybe rename it to "DEFAULT_BUFFER_SIZE" or something similar.
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #20040:
URL: https://github.com/apache/kafka/pull/20040
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2179954736
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2937,9 +2939,208 @@ public void testAppendRecordBatchSize() {
assertFalse(write1.isCompletedExceptionally());
int batchSize = writer.entries(TP).get(0).sizeInBytes();
- assertTrue(batchSize > MIN_BUFFER_SIZE && batchSize < maxBatchSize);
+ assertTrue(batchSize > INITIAL_BUFFER_SIZE && batchSize < maxBatchSize);
}
+ @Test
+ public void testCoordinatorDoNotRetainBufferLargeThanMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate a record larger than the maxBatchSize.
+ List<String> largeRecords = List.of("A".repeat(100 * 1024 * 1024));
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(largeRecords, "response1", null, true, false)
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the initial small one, not the large buffer.
+ assertEquals(INITIAL_BUFFER_SIZE, ctx.bufferSupplier.get(1).capacity());
+ }
+
+ @Test
+ public void testCoordinatorRetainExpandedBufferLessOrEqualToMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024 * 1024) // 1GB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate enough records to create a batch that has INITIAL_BUFFER_SIZE < batchSize < maxBatchSize
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the expanded buffer.
+ assertTrue(ctx.bufferSupplier.get(1).capacity() > INITIAL_BUFFER_SIZE);
+ }
+
+ @Test
+ public void testBufferShrinkWhenMaxMessageSizeReducedBelowInitialBufferSize() {
+ MockTimer timer = new MockTimer();
+ var mockWriter = new InMemoryPartitionWriter(false) {
+ private LogConfig config;
+
+ {
+ config = new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+ }
+
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return config;
+ }
+
+ public void updateConfig(LogConfig newConfig) {
+ this.config = newConfig;
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ ByteBuffer cachedBuffer = ctx.bufferSupplier.get(1);
+ assertEquals(INITIAL_BUFFER_SIZE, cachedBuffer.capacity());
+ ctx.bufferSupplier.release(cachedBuffer);
+
+ // Reduce max message size below initial buffer size.
+ mockWriter.updateConfig(new LogConfig(
+ Map.of(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(INITIAL_BUFFER_SIZE - 66))));
+
+ // Write #2.
+ CompletableFuture<String> write2 = runtime.scheduleWriteOperation("write#2", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response2")
+ );
+ assertFalse(write2.isCompletedExceptionally());
+
+ // Verify that there is no cached buffer.
+ assertEquals(1, ctx.bufferSupplier.get(1).capacity());
+
+ // Write #3.
Review Comment:
Could you please add unit test to ensure the maximum value of capacity of buffer is equal to `maxMessageSize`?
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2175414376
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2937,9 +2939,208 @@ public void testAppendRecordBatchSize() {
assertFalse(write1.isCompletedExceptionally());
int batchSize = writer.entries(TP).get(0).sizeInBytes();
- assertTrue(batchSize > MIN_BUFFER_SIZE && batchSize < maxBatchSize);
+ assertTrue(batchSize > INITIAL_BUFFER_SIZE && batchSize < maxBatchSize);
}
+ @Test
+ public void testCoordinatorDoNotRetainBufferLargeThanMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate a record larger than the maxBatchSize.
+ List<String> largeRecords = List.of("A".repeat(100 * 1024 * 1024));
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(largeRecords, "response1", null, true, false)
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the initial small one, not the large buffer.
+ assertEquals(INITIAL_BUFFER_SIZE, ctx.bufferSupplier.get(1).capacity());
+ }
+
+ @Test
+ public void testCoordinatorRetainExpandedBufferLessOrEqualToMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024 * 1024) // 1GB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate enough records to create a batch that has INITIAL_BUFFER_SIZE < batchSize < maxBatchSize
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the expanded buffer.
+ assertTrue(ctx.bufferSupplier.get(1).capacity() > INITIAL_BUFFER_SIZE);
+ }
+
+ @Test
+ public void testBufferShrinkWhenMaxMessageSizeReducedBelowInitialBufferSize() {
+ MockTimer timer = new MockTimer();
+ var mockWriter = new InMemoryPartitionWriter(false) {
+ private LogConfig config;
+
+ {
+ config = new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+ }
Review Comment:
How about
```suggestion
private LogConfig config = new LogConfig(Map.of(
TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
));
```
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #20040:
URL: https://github.com/apache/kafka/pull/20040#issuecomment-3015125009
@mingyen066 this PR also fix the issue that decreasing the message size dynamically does not re-create the buffer.
Hence, could you please add test for the 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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2173151714
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2940,6 +2942,135 @@ public void testAppendRecordBatchSize() {
assertTrue(batchSize > MIN_BUFFER_SIZE && batchSize < maxBatchSize);
}
+ @Test
+ public void testCoordinatorDoNotRetainLargeBuffer() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ var props = new Properties();
+ props.put(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024)); // 1MB
+ return new LogConfig(props);
+ }
+
+ @Override
+ public long append(
+ TopicPartition tp,
+ VerificationGuard verificationGuard,
+ MemoryRecords batch
+ ) {
+ return super.append(tp, verificationGuard, batch);
+ }
Review Comment:
The method is same as super class, why we need to override it?
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2173262513
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2940,6 +2942,135 @@ public void testAppendRecordBatchSize() {
assertTrue(batchSize > MIN_BUFFER_SIZE && batchSize < maxBatchSize);
}
+ @Test
+ public void testCoordinatorDoNotRetainLargeBuffer() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ var props = new Properties();
+ props.put(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024)); // 1MB
+ return new LogConfig(props);
+ }
+
+ @Override
+ public long append(
+ TopicPartition tp,
+ VerificationGuard verificationGuard,
+ MemoryRecords batch
+ ) {
+ return super.append(tp, verificationGuard, batch);
+ }
Review Comment:
There is a same issue below :)
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2194771745
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2937,9 +2939,206 @@ public void testAppendRecordBatchSize() {
assertFalse(write1.isCompletedExceptionally());
int batchSize = writer.entries(TP).get(0).sizeInBytes();
- assertTrue(batchSize > MIN_BUFFER_SIZE && batchSize < maxBatchSize);
+ assertTrue(batchSize > INITIAL_BUFFER_SIZE && batchSize < maxBatchSize);
}
+ @Test
+ public void testCoordinatorDoNotRetainBufferLargeThanMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate a record larger than the maxBatchSize.
+ List<String> largeRecords = List.of("A".repeat(100 * 1024 * 1024));
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(largeRecords, "response1", null, true, false)
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the initial small one, not the large buffer.
+ assertEquals(INITIAL_BUFFER_SIZE, ctx.bufferSupplier.get(1).capacity());
+ }
+
+ @Test
+ public void testCoordinatorRetainExpandedBufferLessOrEqualToMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024 * 1024) // 1GB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate enough records to create a batch that has INITIAL_BUFFER_SIZE < batchSize < maxBatchSize
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the expanded buffer.
+ assertTrue(ctx.bufferSupplier.get(1).capacity() > INITIAL_BUFFER_SIZE);
+ }
+
+ @Test
+ public void testBufferShrinkWhenMaxMessageSizeReducedBelowInitialBufferSize() {
+ MockTimer timer = new MockTimer();
+ var mockWriter = new InMemoryPartitionWriter(false) {
+ private LogConfig config = new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return config;
+ }
+
+ public void updateConfig(LogConfig newConfig) {
+ this.config = newConfig;
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
Review Comment:
ditto
```suggestion
int batchSize = writer.entries(TP).get(0).sizeInBytes();
assertTrue(records < batchSize);
```
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2937,9 +2939,206 @@ public void testAppendRecordBatchSize() {
assertFalse(write1.isCompletedExceptionally());
int batchSize = writer.entries(TP).get(0).sizeInBytes();
- assertTrue(batchSize > MIN_BUFFER_SIZE && batchSize < maxBatchSize);
+ assertTrue(batchSize > INITIAL_BUFFER_SIZE && batchSize < maxBatchSize);
}
+ @Test
+ public void testCoordinatorDoNotRetainBufferLargeThanMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate a record larger than the maxBatchSize.
+ List<String> largeRecords = List.of("A".repeat(100 * 1024 * 1024));
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(largeRecords, "response1", null, true, false)
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the initial small one, not the large buffer.
+ assertEquals(INITIAL_BUFFER_SIZE, ctx.bufferSupplier.get(1).capacity());
+ }
+
+ @Test
+ public void testCoordinatorRetainExpandedBufferLessOrEqualToMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024 * 1024) // 1GB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate enough records to create a batch that has INITIAL_BUFFER_SIZE < batchSize < maxBatchSize
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the expanded buffer.
+ assertTrue(ctx.bufferSupplier.get(1).capacity() > INITIAL_BUFFER_SIZE);
+ }
+
+ @Test
+ public void testBufferShrinkWhenMaxMessageSizeReducedBelowInitialBufferSize() {
+ MockTimer timer = new MockTimer();
+ var mockWriter = new InMemoryPartitionWriter(false) {
+ private LogConfig config = new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return config;
+ }
+
+ public void updateConfig(LogConfig newConfig) {
+ this.config = newConfig;
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ ByteBuffer cachedBuffer = ctx.bufferSupplier.get(1);
+ assertEquals(INITIAL_BUFFER_SIZE, cachedBuffer.capacity());
+ ctx.bufferSupplier.release(cachedBuffer);
+
+ // Reduce max message size below initial buffer size.
+ mockWriter.updateConfig(new LogConfig(
+ Map.of(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(INITIAL_BUFFER_SIZE - 66))));
+ assertEquals(INITIAL_BUFFER_SIZE - 66, mockWriter.config(TP).maxMessageSize());
+ assertTrue(mockWriter.config(TP).maxMessageSize() < INITIAL_BUFFER_SIZE);
+
+ // Write #2.
+ CompletableFuture<String> write2 = runtime.scheduleWriteOperation("write#2", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response2")
+ );
+ assertFalse(write2.isCompletedExceptionally());
+
+ // Verify that there is no cached buffer since the cached buffer size is greater than new maxMessageSize.
+ assertEquals(1, ctx.bufferSupplier.get(1).capacity());
+
+ // Write #3.
+ CompletableFuture<String> write3 = runtime.scheduleWriteOperation("write#3", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response3")
+ );
+ assertFalse(write3.isCompletedExceptionally());
+
+ // Verify that the cached buffer size is equals to new maxMessageSize that less than INITIAL_BUFFER_SIZE.
+ assertEquals(mockWriter.config(TP).maxMessageSize(), ctx.bufferSupplier.get(1).capacity());
+ }
+
+
+
Review Comment:
Nit: remove extra lines.
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2937,9 +2939,206 @@ public void testAppendRecordBatchSize() {
assertFalse(write1.isCompletedExceptionally());
int batchSize = writer.entries(TP).get(0).sizeInBytes();
- assertTrue(batchSize > MIN_BUFFER_SIZE && batchSize < maxBatchSize);
+ assertTrue(batchSize > INITIAL_BUFFER_SIZE && batchSize < maxBatchSize);
}
+ @Test
+ public void testCoordinatorDoNotRetainBufferLargeThanMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate a record larger than the maxBatchSize.
+ List<String> largeRecords = List.of("A".repeat(100 * 1024 * 1024));
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(largeRecords, "response1", null, true, false)
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the initial small one, not the large buffer.
+ assertEquals(INITIAL_BUFFER_SIZE, ctx.bufferSupplier.get(1).capacity());
+ }
+
+ @Test
+ public void testCoordinatorRetainExpandedBufferLessOrEqualToMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024 * 1024) // 1GB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate enough records to create a batch that has INITIAL_BUFFER_SIZE < batchSize < maxBatchSize
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
Review Comment:
It’s clearer to check if INITIAL_BUFFER_SIZE < batchSize < maxBatchSize first.
```suggestion
int batchSize = writer.entries(TP).get(0).sizeInBytes();
assertTrue(batchSize > INITIAL_BUFFER_SIZE && batchSize <= maxBatchSize);
```
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2167643266
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -627,6 +636,8 @@ private CoordinatorContext(
defaultWriteTimeout
);
this.bufferSupplier = new BufferSupplier.GrowableBufferSupplier();
+ this.minBufferSize = 512 * 1024; // 512 KB
+ this.maxBatchSize = 1024 * 1024 * 1024; // 1 GB
Review Comment:
what is magic number? we need to honor the value of maxMessageSize
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "mingyen066 (via GitHub)" <gi...@apache.org>.
mingyen066 commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2180892266
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2937,9 +2939,208 @@ public void testAppendRecordBatchSize() {
assertFalse(write1.isCompletedExceptionally());
int batchSize = writer.entries(TP).get(0).sizeInBytes();
- assertTrue(batchSize > MIN_BUFFER_SIZE && batchSize < maxBatchSize);
+ assertTrue(batchSize > INITIAL_BUFFER_SIZE && batchSize < maxBatchSize);
}
+ @Test
+ public void testCoordinatorDoNotRetainBufferLargeThanMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate a record larger than the maxBatchSize.
+ List<String> largeRecords = List.of("A".repeat(100 * 1024 * 1024));
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(largeRecords, "response1", null, true, false)
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the initial small one, not the large buffer.
+ assertEquals(INITIAL_BUFFER_SIZE, ctx.bufferSupplier.get(1).capacity());
+ }
+
+ @Test
+ public void testCoordinatorRetainExpandedBufferLessOrEqualToMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024 * 1024) // 1GB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate enough records to create a batch that has INITIAL_BUFFER_SIZE < batchSize < maxBatchSize
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the expanded buffer.
+ assertTrue(ctx.bufferSupplier.get(1).capacity() > INITIAL_BUFFER_SIZE);
+ }
+
+ @Test
+ public void testBufferShrinkWhenMaxMessageSizeReducedBelowInitialBufferSize() {
+ MockTimer timer = new MockTimer();
+ var mockWriter = new InMemoryPartitionWriter(false) {
+ private LogConfig config;
+
+ {
+ config = new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+ }
+
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return config;
+ }
+
+ public void updateConfig(LogConfig newConfig) {
+ this.config = newConfig;
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ ByteBuffer cachedBuffer = ctx.bufferSupplier.get(1);
+ assertEquals(INITIAL_BUFFER_SIZE, cachedBuffer.capacity());
+ ctx.bufferSupplier.release(cachedBuffer);
+
+ // Reduce max message size below initial buffer size.
+ mockWriter.updateConfig(new LogConfig(
+ Map.of(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(INITIAL_BUFFER_SIZE - 66))));
+
+ // Write #2.
+ CompletableFuture<String> write2 = runtime.scheduleWriteOperation("write#2", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response2")
+ );
+ assertFalse(write2.isCompletedExceptionally());
+
+ // Verify that there is no cached buffer.
+ assertEquals(1, ctx.bufferSupplier.get(1).capacity());
+
+ // Write #3.
Review Comment:
I replaced the previous assertion with `assertEquals(mockWriter.config(TP).maxMessageSize(), ctx.bufferSupplier.get(1).capacity())` to ensure that
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2207399148
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2937,7 +2939,210 @@ public void testAppendRecordBatchSize() {
assertFalse(write1.isCompletedExceptionally());
int batchSize = writer.entries(TP).get(0).sizeInBytes();
- assertTrue(batchSize > MIN_BUFFER_SIZE && batchSize < maxBatchSize);
+ assertTrue(batchSize > INITIAL_BUFFER_SIZE && batchSize < maxBatchSize);
+ }
+
+ @Test
+ public void testCoordinatorDoNotRetainBufferLargeThanMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate a record larger than the maxBatchSize.
+ List<String> largeRecords = List.of("A".repeat(100 * 1024 * 1024));
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(largeRecords, "response1", null, true, false)
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the initial small one, not the large buffer.
+ assertEquals(INITIAL_BUFFER_SIZE, ctx.bufferSupplier.get(1).capacity());
+ }
+
+ @Test
+ public void testCoordinatorRetainExpandedBufferLessOrEqualToMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024 * 1024) // 1GB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate enough records to create a batch that has INITIAL_BUFFER_SIZE < batchSize < maxBatchSize
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ int batchSize = mockWriter.entries(TP).get(0).sizeInBytes();
+ int maxBatchSize = mockWriter.config(TP).maxMessageSize();
+ assertTrue(INITIAL_BUFFER_SIZE < batchSize && batchSize <= maxBatchSize);
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the expanded buffer.
+ assertTrue(ctx.bufferSupplier.get(1).capacity() > INITIAL_BUFFER_SIZE);
+ }
+
+ @Test
+ public void testBufferShrinkWhenMaxMessageSizeReducedBelowInitialBufferSize() {
+ MockTimer timer = new MockTimer();
+ var mockWriter = new InMemoryPartitionWriter(false) {
+ private LogConfig config = new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return config;
+ }
+
+ public void updateConfig(LogConfig newConfig) {
+ this.config = newConfig;
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ int batchSize = mockWriter.entries(TP).get(0).sizeInBytes();
+ int maxBatchSize = mockWriter.config(TP).maxMessageSize();
+ assertTrue(batchSize < INITIAL_BUFFER_SIZE && INITIAL_BUFFER_SIZE <= maxBatchSize);
Review Comment:
nit:
```suggestion
assertTrue(batchSize <= INITIAL_BUFFER_SIZE && INITIAL_BUFFER_SIZE <= maxBatchSize);
```
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #20040:
URL: https://github.com/apache/kafka/pull/20040#issuecomment-3078785240
Thanks for all reviews and thanks to @mingyen066 fox fixing it. I'm going to merge it to trunk, 4.1 ( after getting approval from RM), and 4.0. Please feel free to leave more comments. We can address them in the follow-up.
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2173285324
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -859,7 +866,7 @@ private void maybeAllocateNewBatch(
LogConfig logConfig = partitionWriter.config(tp);
int maxBatchSize = logConfig.maxMessageSize();
long prevLastWrittenOffset = coordinator.lastWrittenOffset();
- ByteBuffer buffer = bufferSupplier.get(maxBatchSize);
+ ByteBuffer buffer = bufferSupplier.get(min(MIN_BUFFER_SIZE, maxBatchSize));
Review Comment:
If so, we allow users to assign maxBatchSize smaller than `MIN_BUFFER_SIZE`. `MIN_BUFFER_SIZE` variable name does not match its semantics. Maybe rename it to "INITIAL_BUFFER_SIZE" or something similar.
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2167668045
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -859,7 +866,7 @@ private void maybeAllocateNewBatch(
LogConfig logConfig = partitionWriter.config(tp);
int maxBatchSize = logConfig.maxMessageSize();
long prevLastWrittenOffset = coordinator.lastWrittenOffset();
- ByteBuffer buffer = bufferSupplier.get(maxBatchSize);
+ ByteBuffer buffer = bufferSupplier.get(MIN_BUFFER_SIZE);
Review Comment:
what happens if the `maxBatchSize` is smaller than `MIN_BUFFER_SIZE`?
--
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] KAFKA-19427: Allow the coordinator to grow its buffer dynamically [kafka]
Posted by "mingyen066 (via GitHub)" <gi...@apache.org>.
mingyen066 commented on code in PR #20040:
URL: https://github.com/apache/kafka/pull/20040#discussion_r2194451463
##########
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java:
##########
@@ -2937,9 +2939,206 @@ public void testAppendRecordBatchSize() {
assertFalse(write1.isCompletedExceptionally());
int batchSize = writer.entries(TP).get(0).sizeInBytes();
- assertTrue(batchSize > MIN_BUFFER_SIZE && batchSize < maxBatchSize);
+ assertTrue(batchSize > INITIAL_BUFFER_SIZE && batchSize < maxBatchSize);
}
+ @Test
+ public void testCoordinatorDoNotRetainBufferLargeThanMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate a record larger than the maxBatchSize.
+ List<String> largeRecords = List.of("A".repeat(100 * 1024 * 1024));
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(largeRecords, "response1", null, true, false)
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the initial small one, not the large buffer.
+ assertEquals(INITIAL_BUFFER_SIZE, ctx.bufferSupplier.get(1).capacity());
+ }
+
+ @Test
+ public void testCoordinatorRetainExpandedBufferLessOrEqualToMaxMessageSize() {
+ MockTimer timer = new MockTimer();
+ InMemoryPartitionWriter mockWriter = new InMemoryPartitionWriter(false) {
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024 * 1024) // 1GB
+ ));
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Generate enough records to create a batch that has INITIAL_BUFFER_SIZE < batchSize < maxBatchSize
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ // Verify that the next buffer retrieved from the bufferSupplier is the expanded buffer.
+ assertTrue(ctx.bufferSupplier.get(1).capacity() > INITIAL_BUFFER_SIZE);
+ }
+
+ @Test
+ public void testBufferShrinkWhenMaxMessageSizeReducedBelowInitialBufferSize() {
+ MockTimer timer = new MockTimer();
+ var mockWriter = new InMemoryPartitionWriter(false) {
+ private LogConfig config = new LogConfig(Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, String.valueOf(1024 * 1024) // 1MB
+ ));;
+
+ @Override
+ public LogConfig config(TopicPartition tp) {
+ return config;
+ }
+
+ public void updateConfig(LogConfig newConfig) {
+ this.config = newConfig;
+ }
+ };
+ StringSerializer serializer = new StringSerializer();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(mockWriter)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
+ .withSerializer(serializer)
+ .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(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(List.of(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ List<String> records = new ArrayList<>();
+ for (int i = 0; i < 1000; i++) {
+ records.add("record-" + i);
+ }
+
+ // Write #1.
+ CompletableFuture<String> write1 = runtime.scheduleWriteOperation("write#1", TP, DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(records, "response1")
+ );
+
+ // Verify that the write has not completed exceptionally.
+ // This will catch any exceptions thrown including RecordTooLargeException.
+ assertFalse(write1.isCompletedExceptionally());
+
+ ByteBuffer cachedBuffer = ctx.bufferSupplier.get(1);
+ assertEquals(INITIAL_BUFFER_SIZE, cachedBuffer.capacity());
+ ctx.bufferSupplier.release(cachedBuffer);
Review Comment:
`ctx.bufferSupplier.get(1);` will clear `cachedBuffer` in `bufferSupplier`, so I want to put it back to `bufferSupplier`
--
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