nileshkumar3 commented on code in PR #23765:
URL: https://github.com/apache/kafka/pull/23765#discussion_r4436943005
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java:
##########
@@ -524,6 +539,8 @@ public int recover(ProducerStateManager
producerStateManager, LeaderEpochFileCac
// A normally closed segment always appends the biggest timestamp ever
seen into log segment, we do this as well.
timeIndex().maybeAppend(maxTimestampSoFar(),
shallowOffsetOfMaxTimestampSoFar(), true);
timeIndex().trimToValidSize();
+ if (overflowException != null)
+ throw overflowException;
Review Comment:
Thanks for the detailed review and for catching the second problem. I've
reworked the PR along the lines you proposed .
LogSegment.recover() no longer throws on offset overflow. It records the
first overflowing offset, continues validating subsequent batches, truncates
any invalid tail, and returns RecoveryResult(truncatedBytes, overflowOffset).
LogLoader.recoverLog() deletes the remaining segments when truncatedBytes >
0, then throws LogSegmentOffsetOverflowException if overflow was detected. This
ensures deletion happens before the retry sees the already-trimmed segment.
retryOnOffsetOverflow() still wraps recoverLog(). After splitting, it
retries recovery, rebuilding the transaction indexes on the split segments.
loadSegmentFiles() no longer calls recoverSegment() and is no longer wrapped
in retryOnOffsetOverflow(). For a missing or corrupt index, it sets
hadCleanShutdown = false and updates recoveryPointCheckpoint to
Math.min(recoveryPointCheckpoint, segment.baseOffset()). This preserves the
earliest recovery point and lets the subsequent recoverLog() handle recovery
and any required splitting.
--
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]
