KAFKA-14482: Move LogLoader to storage module #17042
Conversation
| @@ -1775,7 +1776,7 @@ class UnifiedLog(@volatile var logStartOffset: Long, | |||
| producerExpireCheck.cancel(true) | |||
| leaderEpochCache.foreach(_.clear()) | |||
| val deletedSegments = localLog.deleteAllSegments() | |||
There was a problem hiding this comment.
maybe deleteAllSegments can return java collection directly
| * @param function The function to be executed. | ||
| * @return The value returned by the function after a successful invocation | ||
| */ | ||
| public static <T> T maybeHandleIOException(LogDirFailureChannel logDirFailureChannel, |
There was a problem hiding this comment.
Could you please replace LocalLog.scala#maybeHandleIOException by java version?
There was a problem hiding this comment.
This requires a bunch of changes in LocalLog. I think it's best if we do those when we move LocalLog.
There was a problem hiding this comment.
my point was to leverage LocalLog.java#maybeHandleIOException to implement LocalLog.scala#maybeHandleIOException. That can remove the duplicate code. for example:
private def maybeHandleIOException[T](msg: => String)(fun: => T): T = {
JLocalLog.maybeHandleIOException(logDirFailureChannel, parentDir, () => msg, () => fun)
}There was a problem hiding this comment.
Ah right, I see what you meant now. Done
| @@ -906,7 +907,7 @@ class UnifiedLog(@volatile var logStartOffset: Long, | |||
| updateHighWatermarkWithLogEndOffset() | |||
|
|
|||
| // update the producer state | |||
| updatedProducers.values.foreach(producerAppendInfo => producerStateManager.update(producerAppendInfo)) | |||
| updatedProducers.values.asScala.foreach(producerAppendInfo => producerStateManager.update(producerAppendInfo)) | |||
There was a problem hiding this comment.
how about updatedProducers.values.forEach
There was a problem hiding this comment.
Right we don't need this conversion.
| } | ||
|
|
||
| private[log] def splitOverflowedSegment(segment: LogSegment): List[LogSegment] = lock synchronized { | ||
| val result = UnifiedLog.splitOverflowedSegment(segment, localLog.segments, dir, topicPartition, config, scheduler, logDirFailureChannel, logIdent) | ||
| deleteProducerSnapshots(result.deletedSegments, asyncDelete = true) | ||
| result.newSegments.toList | ||
| result.newSegments.asScala.toList |
There was a problem hiding this comment.
it seems the return value is never used, so maybe we can remove it.
There was a problem hiding this comment.
The return value is used in a few places in LogLoaderTest.
Committer Checklist (excluded from commit message)