This is an automated email from the ASF dual-hosted git repository.
AndrewJSchofield pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 998ed9b2b68 KAFKA-20690: silent test-coverage hole in
PersisterStateBatchCombinerTest (#22571)
998ed9b2b68 is described below
commit 998ed9b2b68f9008aa5088b67e79176417bce295
Author: Shekhar Prasad Rajak <[email protected]>
AuthorDate: Mon Jun 15 12:57:37 2026 +0530
KAFKA-20690: silent test-coverage hole in PersisterStateBatchCombinerTest
(#22571)
Fixes https://issues.apache.org/jira/browse/KAFKA-20690 a silent
test-coverage hole in PersisterStateBatchCombinerTest, corrects one
wrong expected output it exposed, and removes a few hot-path allocations
in PersisterStateBatchCombiner.
Corrected the expectedResult for the "Handle overlapping batches in
newBatches, same state" case to include the trailing (124..130, 0, 1)
range the algorithm correctly emits.
No output change.
Updated for the improvements and more tests :
pruneBatches short-circuits when startOffset == -1 or the list is empty.
getMergeCandidatePair reuses a single ArrayList buffer instead of
allocating a LinkedList per call; uses bulk removeAll.
mergeBatches caches compareBatchDeliveryInfo(candidate, prev) into a
local instead of recomputing.
Reviewers: Sushant Mahajan <[email protected]>, Andrew Schofield
<[email protected]>
---
.../share/PersisterStateBatchCombiner.java | 64 ++++++------
.../PersisterStateBatchCombinerStressTest.java | 116 +++++++++++++++++++++
.../share/PersisterStateBatchCombinerTest.java | 91 +++++++---------
3 files changed, 184 insertions(+), 87 deletions(-)
diff --git
a/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombiner.java
b/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombiner.java
index 9078c5fee65..b4eda98cd66 100644
---
a/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombiner.java
+++
b/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombiner.java
@@ -21,7 +21,6 @@ import
org.apache.kafka.server.share.persister.PersisterStateBatch;
import java.util.ArrayList;
import java.util.Iterator;
-import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
import java.util.TreeSet;
@@ -31,6 +30,7 @@ public class PersisterStateBatchCombiner {
private final long startOffset;
private TreeSet<PersisterStateBatch> sortedBatches;
private List<PersisterStateBatch> finalBatchList; // final list is built
here
+ private final List<PersisterStateBatch> nonOverlappingBuffer = new
ArrayList<>(); // reused per findMergeCandidatePair call
public PersisterStateBatchCombiner(
List<PersisterStateBatch> batchesSoFar,
@@ -111,7 +111,8 @@ public class PersisterStateBatchCombiner {
sortedBatches.remove(prev);
sortedBatches.remove(candidate);
- if (compareBatchDeliveryInfo(candidate, prev) == 0) { // same
state and overlap or contiguous
+ int cmp = compareBatchDeliveryInfo(candidate, prev);
+ if (cmp == 0) { // same state and overlap or contiguous
// overlap and same state (prev.firstOffset <=
candidate.firstOffset) due to sort
// covers:
// case: 1 2 3 4 5
6 7 (contiguous)
@@ -180,27 +181,27 @@ public class PersisterStateBatchCombiner {
}
Iterator<PersisterStateBatch> iter = sortedBatches.iterator();
PersisterStateBatch prev = iter.next();
- List<PersisterStateBatch> nonOverlapping = new LinkedList<>();
+ nonOverlappingBuffer.clear();
while (iter.hasNext()) {
PersisterStateBatch candidate = iter.next();
if (candidate.firstOffset() <= prev.lastOffset() || // overlap
prev.lastOffset() + 1 == candidate.firstOffset() &&
compareBatchDeliveryInfo(prev, candidate) == 0) { // contiguous
- updateBatchContainers(nonOverlapping);
+ updateBatchContainers(nonOverlappingBuffer);
return new MergeCandidatePair(
prev,
candidate
);
}
- nonOverlapping.add(prev);
+ nonOverlappingBuffer.add(prev);
prev = candidate;
}
- updateBatchContainers(nonOverlapping);
+ updateBatchContainers(nonOverlappingBuffer);
return MergeCandidatePair.EMPTY;
}
private void updateBatchContainers(List<PersisterStateBatch>
nonOverlappingBatches) {
- nonOverlappingBatches.forEach(sortedBatches::remove);
+ sortedBatches.removeAll(nonOverlappingBatches);
finalBatchList.addAll(nonOverlappingBatches);
}
@@ -212,31 +213,32 @@ public class PersisterStateBatchCombiner {
* the part after it is preserved.
*/
private void pruneBatches() {
- if (startOffset != -1) {
- List<PersisterStateBatch> retainedBatches = new
ArrayList<>(combinedBatchList.size());
- combinedBatchList.forEach(batch -> {
- if (batch.lastOffset() < startOffset) {
- // batch is expired, skip current iteration
- // -------
- // | -> start offset
- return;
- }
-
- if (batch.firstOffset() >= startOffset) {
- // complete batch is valid
- // ---------
- // | -> start offset
- retainedBatches.add(batch);
- } else {
- // start offset intersects batch
- // ---------
- // | -> start offset
- retainedBatches.add(new PersisterStateBatch(startOffset,
batch.lastOffset(), batch.deliveryState(), batch.deliveryCount()));
- }
- });
- // update the instance variable
- combinedBatchList = retainedBatches;
+ if (startOffset == -1 || combinedBatchList.isEmpty()) {
+ return;
}
+ List<PersisterStateBatch> retainedBatches = new
ArrayList<>(combinedBatchList.size());
+ combinedBatchList.forEach(batch -> {
+ if (batch.lastOffset() < startOffset) {
+ // batch is expired, skip current iteration
+ // -------
+ // | -> start offset
+ return;
+ }
+
+ if (batch.firstOffset() >= startOffset) {
+ // complete batch is valid
+ // ---------
+ // | -> start offset
+ retainedBatches.add(batch);
+ } else {
+ // start offset intersects batch
+ // ---------
+ // | -> start offset
+ retainedBatches.add(new PersisterStateBatch(startOffset,
batch.lastOffset(), batch.deliveryState(), batch.deliveryCount()));
+ }
+ });
+ // update the instance variable
+ combinedBatchList = retainedBatches;
}
private void handleSameStateMerge(PersisterStateBatch prev,
PersisterStateBatch candidate) {
diff --git
a/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombinerStressTest.java
b/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombinerStressTest.java
new file mode 100644
index 00000000000..5eafff1554d
--- /dev/null
+++
b/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombinerStressTest.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.coordinator.share;
+
+import org.apache.kafka.server.share.persister.PersisterStateBatch;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class PersisterStateBatchCombinerStressTest {
+
+ @Test
+ public void manyContiguousSameStateBatchesCoalesceIntoOne() {
+ final int batchCount = 10_000;
+ List<PersisterStateBatch> input = new ArrayList<>(batchCount);
+ for (int i = 0; i < batchCount; i++) {
+ long first = i * 10L;
+ input.add(new PersisterStateBatch(first, first + 9, (byte) 0,
(short) 1));
+ }
+
+ List<PersisterStateBatch> result = new
PersisterStateBatchCombiner(input, List.of(), -1)
+ .combineStateBatches();
+
+ assertEquals(1, result.size(), "all contiguous same-state batches must
coalesce");
+ PersisterStateBatch only = result.get(0);
+ assertEquals(0L, only.firstOffset());
+ assertEquals(batchCount * 10L - 1, only.lastOffset());
+ assertEquals((byte) 0, only.deliveryState());
+ assertEquals((short) 1, only.deliveryCount());
+ }
+
+ @Test
+ public void manyShuffledSameStateBatchesCoalesceIntoOne() {
+ final int batchCount = 5_000;
+ List<PersisterStateBatch> input = new ArrayList<>(batchCount);
+ for (int i = 0; i < batchCount; i++) {
+ long first = i * 10L;
+ input.add(new PersisterStateBatch(first, first + 9, (byte) 0,
(short) 1));
+ }
+ Collections.shuffle(input, new Random(42));
+
+ List<PersisterStateBatch> result = new
PersisterStateBatchCombiner(input, List.of(), -1)
+ .combineStateBatches();
+
+ assertEquals(1, result.size());
+ assertEquals(0L, result.get(0).firstOffset());
+ assertEquals(batchCount * 10L - 1, result.get(0).lastOffset());
+ }
+
+ @Test
+ public void alternatingStateBatchesAreNotMerged() {
+ final int batchCount = 1_000;
+ List<PersisterStateBatch> input = new ArrayList<>(batchCount);
+ for (int i = 0; i < batchCount; i++) {
+ long first = i * 10L;
+ byte state = (byte) (i % 2 == 0 ? 0 : 2);
+ input.add(new PersisterStateBatch(first, first + 9, state, (short)
1));
+ }
+
+ List<PersisterStateBatch> result = new
PersisterStateBatchCombiner(input, List.of(), -1)
+ .combineStateBatches();
+
+ assertEquals(batchCount, result.size(), "alternating-state batches
must not coalesce");
+ for (int i = 0; i < batchCount; i++) {
+ assertEquals(i * 10L, result.get(i).firstOffset());
+ assertEquals(i * 10L + 9, result.get(i).lastOffset());
+ }
+ }
+
+ @Test
+ public void resultIsSortedDisjointCoverage() {
+ final int batchCount = 2_000;
+ List<PersisterStateBatch> input = new ArrayList<>(batchCount);
+ Random rng = new Random(7);
+ for (int i = 0; i < batchCount; i++) {
+ long first = rng.nextInt(100_000);
+ long last = first + rng.nextInt(50);
+ byte state = (byte) (rng.nextInt(4));
+ short count = (short) (rng.nextInt(3) + 1);
+ input.add(new PersisterStateBatch(first, last, state, count));
+ }
+
+ List<PersisterStateBatch> result = new
PersisterStateBatchCombiner(input, List.of(), -1)
+ .combineStateBatches();
+
+ for (int i = 1; i < result.size(); i++) {
+ PersisterStateBatch a = result.get(i - 1);
+ PersisterStateBatch b = result.get(i);
+ assertTrue(a.firstOffset() <= a.lastOffset(), "batch " + a + "
must have first <= last");
+ assertTrue(a.lastOffset() < b.firstOffset(),
+ "batches must be strictly disjoint: " + a + " vs " + b);
+ }
+ }
+}
diff --git
a/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombinerTest.java
b/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombinerTest.java
index 26ddbea2c14..1a74a9bd8b3 100644
---
a/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombinerTest.java
+++
b/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombinerTest.java
@@ -35,7 +35,6 @@ public class PersisterStateBatchCombinerTest {
final List<PersisterStateBatch> newBatches;
final List<PersisterStateBatch> expectedResult;
final long startOffset;
- final boolean shouldRun;
BatchTestHolder(
String testName,
@@ -43,24 +42,12 @@ public class PersisterStateBatchCombinerTest {
List<PersisterStateBatch> newBatches,
List<PersisterStateBatch> expectedResult,
long startOffset
- ) {
- this(testName, batchesSoFar, newBatches, expectedResult,
startOffset, false);
- }
-
- BatchTestHolder(
- String testName,
- List<PersisterStateBatch> batchesSoFar,
- List<PersisterStateBatch> newBatches,
- List<PersisterStateBatch> expectedResult,
- long startOffset,
- boolean shouldRun
) {
this.testName = testName;
this.batchesSoFar = batchesSoFar;
this.newBatches = newBatches;
this.expectedResult = expectedResult;
this.startOffset = startOffset;
- this.shouldRun = shouldRun;
}
static List<PersisterStateBatch> singleBatch(
@@ -162,8 +149,7 @@ public class PersisterStateBatchCombinerTest {
BatchTestHolder.singleBatch(100, 110, 0, 1),
BatchTestHolder.singleBatch(105, 108, 0, 1),
BatchTestHolder.singleBatch(100, 110, 0, 1),
- -1,
- true
+ -1
),
new BatchTestHolder(
@@ -203,10 +189,11 @@ public class PersisterStateBatchCombinerTest {
BatchTestHolder.multiBatch()
.addBatch(111, 119, 2, 2)
.addBatch(116, 123, 2, 2) // overlap with first batch
- .build(), // , //[(111-123, 2, 2)]
+ .build(),
BatchTestHolder.multiBatch()
.addBatch(100, 110, 0, 1)
.addBatch(111, 123, 2, 2)
+ .addBatch(124, 130, 0, 1)
.build(),
-1
),
@@ -385,60 +372,52 @@ public class PersisterStateBatchCombinerTest {
@ParameterizedTest
@MethodSource("generatorDifferentStates")
public void testStateBatchCombineDifferentStates(BatchTestHolder test) {
- if (test.shouldRun) {
- assertEquals(test.expectedResult,
- new PersisterStateBatchCombiner(
- test.batchesSoFar,
- test.newBatches,
- test.startOffset)
- .combineStateBatches(),
- test.testName
- );
- }
+ assertEquals(test.expectedResult,
+ new PersisterStateBatchCombiner(
+ test.batchesSoFar,
+ test.newBatches,
+ test.startOffset)
+ .combineStateBatches(),
+ test.testName
+ );
}
@ParameterizedTest
@MethodSource("generatorSameState")
public void testStateBatchCombineSameState(BatchTestHolder test) {
- if (test.shouldRun) {
- assertEquals(test.expectedResult,
- new PersisterStateBatchCombiner(
- test.batchesSoFar,
- test.newBatches,
- test.startOffset)
- .combineStateBatches(),
- test.testName
- );
- }
+ assertEquals(test.expectedResult,
+ new PersisterStateBatchCombiner(
+ test.batchesSoFar,
+ test.newBatches,
+ test.startOffset)
+ .combineStateBatches(),
+ test.testName
+ );
}
@ParameterizedTest
@MethodSource("generatorComplex")
public void testStateBatchCombineComplexCases(BatchTestHolder test) {
- if (test.shouldRun) {
- assertEquals(test.expectedResult,
- new PersisterStateBatchCombiner(
- test.batchesSoFar,
- test.newBatches,
- test.startOffset)
- .combineStateBatches(),
- test.testName
- );
- }
+ assertEquals(test.expectedResult,
+ new PersisterStateBatchCombiner(
+ test.batchesSoFar,
+ test.newBatches,
+ test.startOffset)
+ .combineStateBatches(),
+ test.testName
+ );
}
@ParameterizedTest
@MethodSource("generatorCornerCases")
public void testStateBatchCombineCornerCases(BatchTestHolder test) {
- if (test.shouldRun) {
- assertEquals(test.expectedResult,
- new PersisterStateBatchCombiner(
- test.batchesSoFar,
- test.newBatches,
- test.startOffset)
- .combineStateBatches(),
- test.testName
- );
- }
+ assertEquals(test.expectedResult,
+ new PersisterStateBatchCombiner(
+ test.batchesSoFar,
+ test.newBatches,
+ test.startOffset)
+ .combineStateBatches(),
+ test.testName
+ );
}
}
