junrao commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r3768388018
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -401,6 +460,19 @@ public boolean shouldRetainRecord(RecordBatch batch,
Record record) {
if (outputBuffer.position() > 0) {
outputBuffer.flip();
MemoryRecords retained =
MemoryRecords.readableRecords(outputBuffer);
+
+ // While groupSegmentsBySize() ensures source segments don't
exceed Integer.MAX_VALUE,
Review Comment:
One possibility to fix this issue in trunk is to return the filtered result
to the caller on overflow. Instead of rewinding the input segment, the caller
will roll a new segment and append the returned result to the new segment. We
still need to decouple the updating of the txn index in the filterTo logic.
Claude suggested the following.
```
Stage the index appends instead of writing them during the filter.
1. In CleanedTransactionMetadata, replace the immediate
cleanedIndex.ifPresent(index -> index.append(...)) with adding to a
List<AbortedTxn> pendingAbortedTxns, plus a flushPendingTo(TransactionIndex)
that appends and clears.
2. In cleanInto, flush that list right after a successful
dest.append(result.maxOffset(), retained).
3. On overflow, return the saved records + maxOffset and leave the pending
list untouched. cleanSegments finalizes the outgoing segment, creates the new
one (base offset = first batch of the retained buffer), appends the saved
records, then flushes the pending list into the new segment's index.
filterTo then runs exactly once per chunk, so the destructive parts of the
state machine — ongoingAbortedTxns.remove, ongoingCommittedTxns.remove,
consumeAbortedTxnsUpTo's poll — are each consumed once, and the stats
double-count goes away with it. The index entry lands in whichever segment its
data landed in, by construction. That's the invariant you expected the code to
already have.
```
--
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]
