You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Shekharrajak (via GitHub)" <gi...@apache.org> on 2026/06/14 18:16:33 UTC
[PR] KAFKA-20691: PersisterStateBatchCombiner.combineStateBatches() with an event-driven sweep-line [kafka]
Shekharrajak opened a new pull request, #22572:
URL: https://github.com/apache/kafka/pull/22572
Fix https://issues.apache.org/jira/browse/KAFKA-20691
Current merge rescans the TreeSet from the start on every step. 10k contiguous same-state batches → 10k rescans of a shrinking set, surfacing as non-linear WriteShareGroupState latency when many small acknowledge ranges accumulate per partition.
How:
- Emit (firstOffset, BEGIN) and (lastOffset + 1, END) for each pruned batch.
- Sort events by (offset asc, END before BEGIN).
- Sweep events; keep a max-priority PriorityQueue of active batches + lazy-deletion HashSet.
- At every offset advance, emit a slice with the heap-top's (state, count); coalesce contiguous slices with identical (state, count).
Result
- Worst case: O(n^2) → O((n + k) log n).
- Public API unchanged.
--
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-20691: PersisterStateBatchCombiner.combineStateBatches() with an event-driven sweep-line [kafka]
Posted by "Shekharrajak (via GitHub)" <gi...@apache.org>.
Shekharrajak commented on code in PR #22572:
URL: https://github.com/apache/kafka/pull/22572#discussion_r3412538692
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombiner.java:
##########
@@ -20,385 +20,170 @@
import org.apache.kafka.server.share.persister.PersisterStateBatch;
import java.util.ArrayList;
-import java.util.Iterator;
+import java.util.Comparator;
import java.util.List;
-import java.util.Objects;
-import java.util.TreeSet;
+import java.util.TreeMap;
+/**
+ * Combines an existing list of {@link PersisterStateBatch} entries with a list of newly produced
+ * entries and returns the shortest non-overlapping, {@code (deliveryState, deliveryCount)}-distinct
+ * cover of the union, clipped at {@code startOffset} (SPSO).
+ *
+ * <p>The merge is performed by an event-driven sweep-line over the union of both inputs. Each
+ * input batch contributes one BEGIN event at {@code firstOffset} and one END event at
+ * {@code lastOffset + 1}. Events are processed in offset order (END before BEGIN at the same
+ * offset). A counted ordered map tracks the currently active priorities; the first key defines the
+ * {@code (state, count)} that wins on the current sub-range. Successive sub-ranges with identical
+ * {@code (state, count)} are coalesced on the fly.
+ *
+ * <p>Complexity: {@code O((n + k) log p)} where {@code n} is the total number of input batches,
+ * {@code k} is the number of overlap transitions encountered, and {@code p} is the number of
+ * distinct {@code (deliveryState, deliveryCount)} priorities.
+ */
public class PersisterStateBatchCombiner {
- private List<PersisterStateBatch> combinedBatchList; // link between pruning and merging
+ private static final Comparator<BatchPriority> PRIORITY_DESC = (a, b) -> {
+ int cmpCount = Short.compare(b.deliveryCount(), a.deliveryCount());
Review Comment:
State
0 = AVAILABLE
1 = ACQUIRED
2 = ACKNOWLEDGED
3 = ARCHIVING
4 = ARCHIVED
---
existing: (100,110,state0,count1)
new:
(101,105,state1,count2)
(101,115,state2,count2)
(101,120,state3,count2)
result:
(100,100,state0,count1)
(101,120,state3,count2)
Meaning: all new ranges have higher count 2, and among them state 3 wins over state 2 and 1.
---
--
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-20691: PersisterStateBatchCombiner.combineStateBatches() with an event-driven sweep-line [kafka]
Posted by "Shekharrajak (via GitHub)" <gi...@apache.org>.
Shekharrajak commented on code in PR #22572:
URL: https://github.com/apache/kafka/pull/22572#discussion_r3412498192
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombiner.java:
##########
@@ -20,385 +20,170 @@
import org.apache.kafka.server.share.persister.PersisterStateBatch;
import java.util.ArrayList;
-import java.util.Iterator;
+import java.util.Comparator;
import java.util.List;
-import java.util.Objects;
-import java.util.TreeSet;
+import java.util.TreeMap;
+/**
+ * Combines an existing list of {@link PersisterStateBatch} entries with a list of newly produced
+ * entries and returns the shortest non-overlapping, {@code (deliveryState, deliveryCount)}-distinct
+ * cover of the union, clipped at {@code startOffset} (SPSO).
+ *
+ * <p>The merge is performed by an event-driven sweep-line over the union of both inputs. Each
+ * input batch contributes one BEGIN event at {@code firstOffset} and one END event at
+ * {@code lastOffset + 1}. Events are processed in offset order (END before BEGIN at the same
+ * offset). A counted ordered map tracks the currently active priorities; the first key defines the
+ * {@code (state, count)} that wins on the current sub-range. Successive sub-ranges with identical
+ * {@code (state, count)} are coalesced on the fly.
+ *
+ * <p>Complexity: {@code O((n + k) log p)} where {@code n} is the total number of input batches,
+ * {@code k} is the number of overlap transitions encountered, and {@code p} is the number of
+ * distinct {@code (deliveryState, deliveryCount)} priorities.
+ */
public class PersisterStateBatchCombiner {
- private List<PersisterStateBatch> combinedBatchList; // link between pruning and merging
+ private static final Comparator<BatchPriority> PRIORITY_DESC = (a, b) -> {
+ int cmpCount = Short.compare(b.deliveryCount(), a.deliveryCount());
Review Comment:
1. Higher deliveryCount wins.
2. If deliveryCount is equal, higher deliveryState wins.
--
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-20691: PersisterStateBatchCombiner.combineStateBatches() with an event-driven sweep-line [kafka]
Posted by "Shekharrajak (via GitHub)" <gi...@apache.org>.
Shekharrajak commented on code in PR #22572:
URL: https://github.com/apache/kafka/pull/22572#discussion_r3412538692
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombiner.java:
##########
@@ -20,385 +20,170 @@
import org.apache.kafka.server.share.persister.PersisterStateBatch;
import java.util.ArrayList;
-import java.util.Iterator;
+import java.util.Comparator;
import java.util.List;
-import java.util.Objects;
-import java.util.TreeSet;
+import java.util.TreeMap;
+/**
+ * Combines an existing list of {@link PersisterStateBatch} entries with a list of newly produced
+ * entries and returns the shortest non-overlapping, {@code (deliveryState, deliveryCount)}-distinct
+ * cover of the union, clipped at {@code startOffset} (SPSO).
+ *
+ * <p>The merge is performed by an event-driven sweep-line over the union of both inputs. Each
+ * input batch contributes one BEGIN event at {@code firstOffset} and one END event at
+ * {@code lastOffset + 1}. Events are processed in offset order (END before BEGIN at the same
+ * offset). A counted ordered map tracks the currently active priorities; the first key defines the
+ * {@code (state, count)} that wins on the current sub-range. Successive sub-ranges with identical
+ * {@code (state, count)} are coalesced on the fly.
+ *
+ * <p>Complexity: {@code O((n + k) log p)} where {@code n} is the total number of input batches,
+ * {@code k} is the number of overlap transitions encountered, and {@code p} is the number of
+ * distinct {@code (deliveryState, deliveryCount)} priorities.
+ */
public class PersisterStateBatchCombiner {
- private List<PersisterStateBatch> combinedBatchList; // link between pruning and merging
+ private static final Comparator<BatchPriority> PRIORITY_DESC = (a, b) -> {
+ int cmpCount = Short.compare(b.deliveryCount(), a.deliveryCount());
Review Comment:
State
0 = AVAILABLE
1 = ACQUIRED
2 = ACKNOWLEDGED
3 = ARCHIVING
4 = ARCHIVED
---
existing: (100,110,state0,count1)
new:
(101,105,state1,count2)
(101,115,state2,count2)
(101,120,state3,count2)
result:
(100,100,state0,count1)
(101,120,state3,count2)
Meaning: all new ranges have higher count 2, and among them state 3 wins over state 2 and 1.
---
--
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-20691: PersisterStateBatchCombiner.combineStateBatches() with an event-driven sweep-line [kafka]
Posted by "Shekharrajak (via GitHub)" <gi...@apache.org>.
Shekharrajak commented on PR #22572:
URL: https://github.com/apache/kafka/pull/22572#issuecomment-4706685968
Thanks @smjn , added the unit test and simplified the sweep line algo : https://github.com/apache/kafka/pull/22572/changes#diff-05d00e514823f6655dc646f8d8cd297f60dc339f91609beaa9f3a8bfa9da05cbR140
--
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-20691: PersisterStateBatchCombiner.combineStateBatches() with an event-driven sweep-line [kafka]
Posted by "Shekharrajak (via GitHub)" <gi...@apache.org>.
Shekharrajak commented on code in PR #22572:
URL: https://github.com/apache/kafka/pull/22572#discussion_r3410005478
##########
share-coordinator/src/test/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombinerTest.java:
##########
@@ -35,32 +35,19 @@ static class BatchTestHolder {
final List<PersisterStateBatch> newBatches;
final List<PersisterStateBatch> expectedResult;
final long startOffset;
- final boolean shouldRun;
BatchTestHolder(
String testName,
List<PersisterStateBatch> batchesSoFar,
List<PersisterStateBatch> newBatches,
List<PersisterStateBatch> expectedResult,
long startOffset
- ) {
Review Comment:
Ref https://github.com/apache/kafka/pull/22571 -
we must merge https://github.com/apache/kafka/pull/22571 first.
--
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-20691: PersisterStateBatchCombiner.combineStateBatches() with an event-driven sweep-line [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #22572:
URL: https://github.com/apache/kafka/pull/22572#issuecomment-4702973821
@Shekharrajak
Thanks for the PR and improvements.
The code does not handle duplicate batches correctly. For example:
## Same state
```
(0, 5,state0,count1), (0,5,state0,count1), (11,15,state0,count1)
```
is producing
```
(0,15,state0,count1)
```
which means its inventing (6,10,state0,count1)
## Different states
```
(0, 5,state2,count1), (0,5,state2,count1), (11,15,state0,count1)
```
is producing
```
(0,15,state2,count1) // BIGGER CONCERN AS STATE CHANGING OF OLDER BATCHES
```
which means its inventing (6,10,state2,count1) and modifying older batch state.
Since there is no restriction on the the batches coming from SharePartition and to the coordinator - duplicates need to be handled possible.
--
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-20691: PersisterStateBatchCombiner.combineStateBatches() with an event-driven sweep-line [kafka]
Posted by "Shekharrajak (via GitHub)" <gi...@apache.org>.
Shekharrajak commented on code in PR #22572:
URL: https://github.com/apache/kafka/pull/22572#discussion_r3412305517
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/PersisterStateBatchCombiner.java:
##########
@@ -20,385 +20,170 @@
import org.apache.kafka.server.share.persister.PersisterStateBatch;
import java.util.ArrayList;
-import java.util.Iterator;
+import java.util.Comparator;
import java.util.List;
-import java.util.Objects;
-import java.util.TreeSet;
+import java.util.TreeMap;
+/**
+ * Combines an existing list of {@link PersisterStateBatch} entries with a list of newly produced
+ * entries and returns the shortest non-overlapping, {@code (deliveryState, deliveryCount)}-distinct
+ * cover of the union, clipped at {@code startOffset} (SPSO).
+ *
+ * <p>The merge is performed by an event-driven sweep-line over the union of both inputs. Each
+ * input batch contributes one BEGIN event at {@code firstOffset} and one END event at
+ * {@code lastOffset + 1}. Events are processed in offset order (END before BEGIN at the same
+ * offset). A counted ordered map tracks the currently active priorities; the first key defines the
+ * {@code (state, count)} that wins on the current sub-range. Successive sub-ranges with identical
+ * {@code (state, count)} are coalesced on the fly.
+ *
+ * <p>Complexity: {@code O((n + k) log p)} where {@code n} is the total number of input batches,
+ * {@code k} is the number of overlap transitions encountered, and {@code p} is the number of
+ * distinct {@code (deliveryState, deliveryCount)} priorities.
+ */
public class PersisterStateBatchCombiner {
- private List<PersisterStateBatch> combinedBatchList; // link between pruning and merging
+ private static final Comparator<BatchPriority> PRIORITY_DESC = (a, b) -> {
+ int cmpCount = Short.compare(b.deliveryCount(), a.deliveryCount());
+ if (cmpCount != 0) {
+ return cmpCount;
+ }
+ return Byte.compare(b.deliveryState(), a.deliveryState());
+ };
+
+ private final List<PersisterStateBatch> batchesSoFar;
+ private final List<PersisterStateBatch> newBatches;
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,
List<PersisterStateBatch> newBatches,
long startOffset
) {
- initializeCombinedList(batchesSoFar, newBatches);
- int estimatedResultSize = (combinedBatchList.size() * 3) / 2; // heuristic size - 50% overallocation
- finalBatchList = new ArrayList<>(estimatedResultSize);
+ this.batchesSoFar = batchesSoFar == null ? List.of() : batchesSoFar;
+ this.newBatches = newBatches == null ? List.of() : newBatches;
this.startOffset = startOffset;
}
- private void initializeCombinedList(List<PersisterStateBatch> batchesSoFar, List<PersisterStateBatch> newBatches) {
- boolean soFarEmpty = batchesSoFar == null || batchesSoFar.isEmpty();
- boolean newBatchesEmpty = newBatches == null || newBatches.isEmpty();
-
- if (soFarEmpty && newBatchesEmpty) {
- combinedBatchList = new ArrayList<>();
- } else if (soFarEmpty) {
- combinedBatchList = new ArrayList<>(newBatches); // new list as the original one could be unmodifiable
- } else if (newBatchesEmpty) {
- combinedBatchList = new ArrayList<>(batchesSoFar); // new list as the original one could be unmodifiable
- } else {
- combinedBatchList = new ArrayList<>(batchesSoFar.size() + newBatches.size());
- combinedBatchList.addAll(batchesSoFar);
- combinedBatchList.addAll(newBatches);
- }
- }
-
/**
- * Algorithm: Merge current state batches and new batches into a single non-overlapping batch list.
- * Input: batchesSoFar, newBatches, startOffset
- * Output: combined list with non-overlapping batches (finalBatchList)
- * <p>
- * - Add both currentBatches and newBatches into a single list combinedBatchList
- * - if combinedBatchList.size() <= 1 return combinedBatchList
- * <p>
- * - Remove/prune any batches from the combinedBatchList:
- * - if batch.lastOffset < startOffset then remove batch from combinedBatchList
- * - else if batch.firstOffset > startOffset then we will keep the batch
- * - else if batch.firstOffset <= startOffset <= batch.lastOffset then keep [startOffset, batch.lastOffset] part only and discard rest.
- * <p>
- * - create a treeset sortedBatches using pruned combinedBatchList
- * - find first 2 mergeable batches in sortedBatches set, say, prev and candidate.
- * - remove any non-overlapping batches from sortedBatches encountered during the find operation and add them to a finalBatchList
- * - do repeat until a mergeable pair is not found:
- * - based on various conditions of offset overlap and batch state differences combine the batches or
- * create new batches, if required, and add to the sortedBatches.
- * - find first 2 mergeable batches in sortedBatches set, say, prev and candidate.
- * - remove any non-mergeable batches from sortedBatches encountered during the find operation and add them to a finalBatchList
- * - done
- * - return the finalBatchList
- *
- * @return list of {@link PersisterStateBatch} representing non-overlapping combined batches
+ * Produces the merged, pruned, non-overlapping batch list.
*/
public List<PersisterStateBatch> combineStateBatches() {
- pruneBatches();
- mergeBatches();
- return finalBatchList;
- }
-
- private void mergeBatches() {
- if (combinedBatchList.size() < 2) {
- finalBatchList = combinedBatchList;
- return;
- }
-
- sortedBatches = new TreeSet<>(combinedBatchList);
-
- MergeCandidatePair overlapState = getMergeCandidatePair();
-
- while (overlapState != MergeCandidatePair.EMPTY) {
- PersisterStateBatch prev = overlapState.prev();
- PersisterStateBatch candidate = overlapState.candidate();
-
- // remove both previous and candidate for easier
- // assessment about adding batches to sortedBatches
- sortedBatches.remove(prev);
- sortedBatches.remove(candidate);
-
- 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)
- // prev: ------ ------- ------- ------- ------- -------- -------
- // candidate: ------ ---- ---------- --- ---- ------- -------
- handleSameStateMerge(prev, candidate); // pair can be contiguous or overlapping
- } else {
- // If we reach here then it is guaranteed that the batch pair is overlapping and
- // non-contiguous because getMergeCandidatePair only returns contiguous pair if
- // the constituents have the same delivery count and state.
- // covers:
- // case: 1 2* 3 4 5 6 7*
- // prev: ------ ------- ------- ------- ------- -------- ------
- // candidate: ------ ---- --------- ---- ---- ------- -------
- // max batches: 1 2 2 3 2 2 2
- // min batches: 1 1 1 1 1 2 1
- // * not possible with treeset
- handleDiffStateOverlap(prev, candidate);
- }
- overlapState = getMergeCandidatePair();
+ List<PersisterStateBatch> pruned = prune();
+ if (pruned.isEmpty()) {
+ return pruned;
}
- finalBatchList.addAll(sortedBatches); // some non overlapping batches might have remained
- }
-
- /**
- * Compares the non-offset state of 2 batches i.e. the deliveryCount and deliverState.
- * <p>
- * Uses standard compareTo contract x < y => +int, x > y => -int, x == y => 0
- *
- * @param b1 - {@link PersisterStateBatch} to compare
- * @param b2 - {@link PersisterStateBatch} to compare
- * @return int representing comparison result.
- */
- private int compareBatchDeliveryInfo(PersisterStateBatch b1, PersisterStateBatch b2) {
- int deltaCount = Short.compare(b1.deliveryCount(), b2.deliveryCount());
-
- // Delivery state could be:
- // 0 - AVAILABLE (non-terminal)
- // 1 - ACQUIRED - should not be persisted yet
- // 2 - ACKNOWLEDGED (terminal)
- // 3 - ARCHIVING - not implemented in KIP-932 - non-terminal - leads only to ARCHIVED
- // 4 - ARCHIVED (terminal)
-
- if (deltaCount == 0) { // same delivery count
- return Byte.compare(b1.deliveryState(), b2.deliveryState());
+ if (pruned.size() == 1) {
+ return pruned;
}
- return deltaCount;
+ return sweepMerge(pruned);
}
/**
- * Accepts a sorted set of state batches and finds the first 2 batches which can be merged.
- * Merged implies that they have some offsets in common or, they are contiguous with the same state.
- * <p>
- * Any non-mergeable batches prefixing a good mergeable pair are removed from the sortedBatches.
- * For example:
- * ----- ---- ----- ----- -----
- * ------
- * <---------------> <-------->
- * non-overlapping 1st overlapping pair
- *
- * @return object representing the overlap state
+ * Drops or clips ranges below {@code startOffset}. Returns a new list; never modifies inputs.
*/
- private MergeCandidatePair getMergeCandidatePair() {
- if (sortedBatches == null || sortedBatches.isEmpty()) {
- return MergeCandidatePair.EMPTY;
- }
- Iterator<PersisterStateBatch> iter = sortedBatches.iterator();
- PersisterStateBatch prev = iter.next();
- 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(nonOverlappingBuffer);
- return new MergeCandidatePair(
- prev,
- candidate
- );
+ private List<PersisterStateBatch> prune() {
+ int estimate = batchesSoFar.size() + newBatches.size();
+ List<PersisterStateBatch> out = new ArrayList<>(estimate);
+ addPruned(out, batchesSoFar);
+ addPruned(out, newBatches);
+ return out;
+ }
+
+ private void addPruned(List<PersisterStateBatch> out, List<PersisterStateBatch> src) {
+ for (PersisterStateBatch b : src) {
+ if (startOffset != -1 && b.lastOffset() < startOffset) {
+ // batch fully expired
+ continue;
+ }
+ if (startOffset == -1 || b.firstOffset() >= startOffset) {
+ out.add(b);
+ } else {
+ // start offset intersects batch -> clip
+ out.add(new PersisterStateBatch(startOffset, b.lastOffset(), b.deliveryState(), b.deliveryCount()));
}
- nonOverlappingBuffer.add(prev);
- prev = candidate;
}
-
- updateBatchContainers(nonOverlappingBuffer);
- return MergeCandidatePair.EMPTY;
- }
-
- private void updateBatchContainers(List<PersisterStateBatch> nonOverlappingBatches) {
- sortedBatches.removeAll(nonOverlappingBatches);
- finalBatchList.addAll(nonOverlappingBatches);
}
/**
- * Accepts a list of {@link PersisterStateBatch} and checks:
- * - last offset is < start offset => batch is removed
- * - first offset > start offset => batch is preserved
- * - start offset intersects the batch => part of batch before start offset is removed and
- * the part after it is preserved.
+ * Event-driven sweep. Linear time after the initial event sort.
*/
- private void pruneBatches() {
- 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()));
+ private List<PersisterStateBatch> sweepMerge(List<PersisterStateBatch> batches) {
+ int n = batches.size();
+ Event[] events = new Event[n * 2];
+ for (int i = 0; i < n; i++) {
+ PersisterStateBatch b = batches.get(i);
+ BatchPriority priority = BatchPriority.from(b);
+ events[i * 2] = new Event(b.firstOffset(), true, priority);
+ events[i * 2 + 1] = new Event(b.lastOffset() + 1, false, priority);
+ }
+ // END (isBegin=false) sorts before BEGIN (isBegin=true) at the same offset so that
+ // contiguous same-state ranges meeting at offset X collapse to a single emit.
+ java.util.Arrays.sort(events, (e1, e2) -> {
Review Comment:
smaller offset first and If two events have the same offset, END goes before BEGIN because:
false = END
true = BEGIN
Boolean.compare(false, true) puts END first.
--
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-20691: PersisterStateBatchCombiner.combineStateBatches() with an event-driven sweep-line [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #22572:
URL: https://github.com/apache/kafka/pull/22572
--
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