cxm940188 commented on code in PR #23525:
URL: https://github.com/apache/kafka/pull/23525#discussion_r4322202971
##########
metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java:
##########
@@ -1637,6 +1652,54 @@ public void testAppendRecordsAtomically() {
appender)).getMessage());
}
+ @ParameterizedTest(name = "numRecords={0}, maxRecordsPerBatch={1}")
+ @CsvSource({
+ "1, 1",
+ "5, 1",
+ "5, 10",
+ "10, 10",
+ "15, 10",
+ "20, 10"
+ })
+ public void testAppendRecordsNonAtomicBatching(int numRecords, int
maxRecordsPerBatch) {
+ List<ApiMessageAndVersion> records = new ArrayList<>();
+ for (int i = 0; i < numRecords; i++) {
+ records.add(rec(i));
+ }
+ RecordingAppender appender = new RecordingAppender();
+ long returned = QuorumController.appendRecords(log,
+ ControllerResult.of(records, null), maxRecordsPerBatch, appender);
+
+ List<Integer> expectedSizes = new ArrayList<>();
+ for (int i = 0; i < numRecords; i += maxRecordsPerBatch) {
+ expectedSizes.add(Math.min(maxRecordsPerBatch, numRecords - i));
+ }
+ assertEquals(expectedSizes, appender.batchSizes,
+ "batches should be exactly maxRecordsPerBatch-sized except the
final one");
+ assertTrue(appender.batchSizes.stream().allMatch(s -> s > 0 && s <=
maxRecordsPerBatch),
+ "every batch should be non-empty and no larger than
maxRecordsPerBatch");
+ assertEquals(numRecords,
appender.batchSizes.stream().mapToLong(Integer::longValue).sum(),
+ "no record should be dropped or duplicated");
+ assertEquals((long) numRecords, returned);
+ }
+
+ @Test
+ public void testNonAtomicWriteDoesNotTriggerControllerFailover() throws
Throwable {
Review Comment:
This integration test proves the **end-to-end** effect: a non-atomic write
whose record count is an exact multiple of `maxRecordsPerBatch` must not cause
the active controller to renounce leadership.
- `setControllerMaxRecordsPerBatch(1)` makes every non-empty write an exact
multiple.
- `registerBrokersAndUnfence` drives a real `registerBroker` write through
the real `ControllerWriteEvent → appendRecords → prepareAppend` path.
- The trailing assertion checks `curClaimEpoch` is unchanged.
The test throws at `registerBroker(...).get()` (an `ExecutionException`),
not at the epoch assertion. With `maxRecordsPerBatch = 1` and one record, the
unfixed loop appends an empty `subList(1, 1)`; `MockRaftClient.prepareAppend`
throws `IllegalArgumentException("Batch cannot be empty")`;
`EventHandlerExceptionInfo`'s catch-all marks it `causesFailover = true` and
wraps it in `UnknownServerException`; the controller calls `renounce()` and the
future completes exceptionally, so `.get()` throws.
**Actual failure on trunk**
`./gradlew metadata:test --tests
"QuorumControllerTest.testNonAtomicWriteDoesNotTriggerControllerFailover`
```
> Task :metadata:test FAILED
Gradle Test Executor 3 > QuorumControllerTest >
testNonAtomicWriteDoesNotTriggerControllerFailover() FAILED
java.util.concurrent.ExecutionException:
org.apache.kafka.common.errors.UnknownServerException:
java.lang.IllegalArgumentException: Batch cannot be empty
at
java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396)
at
java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073)
at
org.apache.kafka.controller.QuorumControllerIntegrationTestUtils.registerBrokersAndUnfence(QuorumControllerIntegrationTestUtils.java:134)
at
org.apache.kafka.controller.QuorumControllerTest.testNonAtomicWriteDoesNotTriggerControllerFailover(QuorumControllerTest.java:1695)
Caused by:
org.apache.kafka.common.errors.UnknownServerException:
java.lang.IllegalArgumentException: Batch cannot be empty
Caused by:
java.lang.IllegalArgumentException: Batch cannot be empty
at
org.apache.kafka.controller.MockRaftClient.prepareAppend(MockRaftClient.java:681)
at
org.apache.kafka.controller.QuorumController$ControllerWriteEvent.lambda$run$0(QuorumController.java:822)
at
org.apache.kafka.controller.QuorumController.appendRecords(QuorumController.java:923)
at
org.apache.kafka.controller.QuorumController$ControllerWriteEvent.run(QuorumController.java:816)
at
org.apache.kafka.queue.KafkaEventQueue$EventContext.run(KafkaEventQueue.java:134)
...
1 test completed, 1 failed
BUILD FAILED in 45s
```
The epoch assertion is the fixed-code sentinel and never runs on trunk.
---
Do you have any other suggestions for this test — for example, restructuring
it or removing it in favor of the unit-level coverage?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
