junrao commented on code in PR #24072:
URL: https://github.com/apache/kafka/pull/24072#discussion_r4449102094
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -728,7 +728,13 @@ public void buildOffsetMap(UnifiedLog log,
logger.info("Building offset map for log {} for {} segments in offset
range [{}, {}).", log.name(), dirty.size(), start, end);
CleanedTransactionMetadata transactionMetadata = new
CleanedTransactionMetadata();
- List<AbortedTxn> abortedTransactions =
log.collectAbortedTransactions(start, end);
+ // collectAbortedTransactions uses the log start offset, not start, as
its lower bound, so it also returns
+ // aborted transactions that ended before start. This pass may not
read their ABORT markers (reading begins at
+ // the offset index entry at or below start), and seeding them would
make every later transactional batch of
+ // those producers, including committed ones, count as aborted and
leave its keys out of the offset map.
+ List<AbortedTxn> abortedTransactions =
log.collectAbortedTransactions(start, end).stream()
Review Comment:
Hmm, should we just fix the implementation of
log.collectAbortedTransactions? It seems that
localLog.collectAbortedTransactions() doesn't need to take in logStartOffset.
It can just use startOffset to locate the starting segment.
--
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]
