You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/04/05 15:57:09 UTC
[GitHub] [kafka] kowshik opened a new pull request #10478: KAFKA-12553: Refactor recovery logic
kowshik opened a new pull request #10478:
URL: https://github.com/apache/kafka/pull/10478
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#issuecomment-821333025
@junrao: Thanks for the review! I have addressed the latest comments and also fixed the broken `PartitionLockTest` test. This PR is ready for review.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r610968190
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -255,19 +261,21 @@ case object SnapshotGenerated extends LogStartOffsetIncrementReason {
@threadsafe
class Log(@volatile private var _dir: File,
@volatile var config: LogConfig,
+ val segments: LogSegments,
@volatile var logStartOffset: Long,
@volatile var recoveryPoint: Long,
+ @volatile var nextOffsetMetadata: LogOffsetMetadata,
scheduler: Scheduler,
brokerTopicStats: BrokerTopicStats,
val time: Time,
val maxProducerIdExpirationMs: Int,
val producerIdExpirationCheckIntervalMs: Int,
val topicPartition: TopicPartition,
+ @volatile var leaderEpochCache: Option[LeaderEpochFileCache],
val producerStateManager: ProducerStateManager,
logDirFailureChannel: LogDirFailureChannel,
- private val hadCleanShutdown: Boolean = true,
@volatile var topicId: Option[Uuid],
- val keepPartitionMetadataFile: Boolean) extends Logging with KafkaMetricsGroup {
+ val keepPartitionMetadataFile: Boolean = true) extends Logging with KafkaMetricsGroup {
Review comment:
Is the default value needed?
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -2586,11 +1999,15 @@ object Log {
logDirFailureChannel: LogDirFailureChannel,
lastShutdownClean: Boolean = true,
topicId: Option[Uuid],
- keepPartitionMetadataFile: Boolean): Log = {
+ keepPartitionMetadataFile: Boolean = true): Log = {
+ // create the log directory if it doesn't exist
+ Files.createDirectories(dir.toPath)
val topicPartition = Log.parseTopicPartitionName(dir)
- val producerStateManager = new ProducerStateManager(topicPartition, dir, maxProducerIdExpirationMs)
- new Log(dir, config, logStartOffset, recoveryPoint, scheduler, brokerTopicStats, time, maxProducerIdExpirationMs,
- producerIdExpirationCheckIntervalMs, topicPartition, producerStateManager, logDirFailureChannel, lastShutdownClean, topicId, keepPartitionMetadataFile)
+ val logLoader = new LogLoader(dir, topicPartition, config, scheduler, time, logDirFailureChannel)
Review comment:
Do we need LogLoader to be a class? Another option is to have LogLoader as an Object with a single public method `load()`.
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -2586,11 +1999,15 @@ object Log {
logDirFailureChannel: LogDirFailureChannel,
lastShutdownClean: Boolean = true,
topicId: Option[Uuid],
- keepPartitionMetadataFile: Boolean): Log = {
+ keepPartitionMetadataFile: Boolean = true): Log = {
Review comment:
Is the default value needed? If so, it seems many tests are not taking advantage of this.
##########
File path: core/src/test/scala/unit/kafka/log/LogTestUtils.scala
##########
@@ -37,4 +47,246 @@ object LogTestUtils {
new LogSegment(ms, idx, timeIdx, txnIndex, offset, indexIntervalBytes, 0, time)
}
+
+ def createLogConfig(segmentMs: Long = Defaults.SegmentMs,
+ segmentBytes: Int = Defaults.SegmentSize,
+ retentionMs: Long = Defaults.RetentionMs,
+ retentionBytes: Long = Defaults.RetentionSize,
+ segmentJitterMs: Long = Defaults.SegmentJitterMs,
+ cleanupPolicy: String = Defaults.CleanupPolicy,
+ maxMessageBytes: Int = Defaults.MaxMessageSize,
+ indexIntervalBytes: Int = Defaults.IndexInterval,
+ segmentIndexBytes: Int = Defaults.MaxIndexSize,
+ messageFormatVersion: String = Defaults.MessageFormatVersion,
+ fileDeleteDelayMs: Long = Defaults.FileDeleteDelayMs): LogConfig = {
+ val logProps = new Properties()
+
+ logProps.put(LogConfig.SegmentMsProp, segmentMs: java.lang.Long)
+ logProps.put(LogConfig.SegmentBytesProp, segmentBytes: Integer)
+ logProps.put(LogConfig.RetentionMsProp, retentionMs: java.lang.Long)
+ logProps.put(LogConfig.RetentionBytesProp, retentionBytes: java.lang.Long)
+ logProps.put(LogConfig.SegmentJitterMsProp, segmentJitterMs: java.lang.Long)
+ logProps.put(LogConfig.CleanupPolicyProp, cleanupPolicy)
+ logProps.put(LogConfig.MaxMessageBytesProp, maxMessageBytes: Integer)
+ logProps.put(LogConfig.IndexIntervalBytesProp, indexIntervalBytes: Integer)
+ logProps.put(LogConfig.SegmentIndexBytesProp, segmentIndexBytes: Integer)
+ logProps.put(LogConfig.MessageFormatVersionProp, messageFormatVersion)
+ logProps.put(LogConfig.FileDeleteDelayMsProp, fileDeleteDelayMs: java.lang.Long)
+ LogConfig(logProps)
+ }
+
+ def createLog(dir: File,
+ config: LogConfig,
+ brokerTopicStats: BrokerTopicStats,
+ scheduler: Scheduler,
+ time: Time,
+ logStartOffset: Long = 0L,
+ recoveryPoint: Long = 0L,
+ maxProducerIdExpirationMs: Int = 60 * 60 * 1000,
+ producerIdExpirationCheckIntervalMs: Int = LogManager.ProducerIdExpirationCheckIntervalMs,
+ lastShutdownClean: Boolean = true,
+ topicId: Option[Uuid] = None): Log = {
+ Log(dir = dir,
+ config = config,
+ logStartOffset = logStartOffset,
+ recoveryPoint = recoveryPoint,
+ scheduler = scheduler,
+ brokerTopicStats = brokerTopicStats,
+ time = time,
+ maxProducerIdExpirationMs = maxProducerIdExpirationMs,
+ producerIdExpirationCheckIntervalMs = producerIdExpirationCheckIntervalMs,
+ logDirFailureChannel = new LogDirFailureChannel(10),
+ lastShutdownClean = lastShutdownClean,
+ topicId = topicId)
+ }
+
+ /**
+ * Check if the given log contains any segment with records that cause offset overflow.
+ * @param log Log to check
+ * @return true if log contains at least one segment with offset overflow; false otherwise
+ */
+ def hasOffsetOverflow(log: Log): Boolean = firstOverflowSegment(log).isDefined
+
+ def firstOverflowSegment(log: Log): Option[LogSegment] = {
+ def hasOverflow(baseOffset: Long, batch: RecordBatch): Boolean =
+ batch.lastOffset > baseOffset + Int.MaxValue || batch.baseOffset < baseOffset
+
+ for (segment <- log.logSegments) {
+ val overflowBatch = segment.log.batches.asScala.find(batch => hasOverflow(segment.baseOffset, batch))
+ if (overflowBatch.isDefined)
+ return Some(segment)
+ }
+ None
+ }
+
+ private def rawSegment(logDir: File, baseOffset: Long): FileRecords =
+ FileRecords.open(Log.logFile(logDir, baseOffset))
+
+ /**
+ * Initialize the given log directory with a set of segments, one of which will have an
+ * offset which overflows the segment
+ */
+ def initializeLogDirWithOverflowedSegment(logDir: File): Unit = {
+ def writeSampleBatches(baseOffset: Long, segment: FileRecords): Long = {
+ def record(offset: Long) = {
+ val data = offset.toString.getBytes
+ new SimpleRecord(data, data)
+ }
+
+ segment.append(MemoryRecords.withRecords(baseOffset, CompressionType.NONE, 0,
+ record(baseOffset)))
+ segment.append(MemoryRecords.withRecords(baseOffset + 1, CompressionType.NONE, 0,
+ record(baseOffset + 1),
+ record(baseOffset + 2)))
+ segment.append(MemoryRecords.withRecords(baseOffset + Int.MaxValue - 1, CompressionType.NONE, 0,
+ record(baseOffset + Int.MaxValue - 1)))
+ // Need to create the offset files explicitly to avoid triggering segment recovery to truncate segment.
+ Log.offsetIndexFile(logDir, baseOffset).createNewFile()
+ Log.timeIndexFile(logDir, baseOffset).createNewFile()
+ baseOffset + Int.MaxValue
+ }
+
+ def writeNormalSegment(baseOffset: Long): Long = {
+ val segment = rawSegment(logDir, baseOffset)
+ try writeSampleBatches(baseOffset, segment)
+ finally segment.close()
+ }
+
+ def writeOverflowSegment(baseOffset: Long): Long = {
+ val segment = rawSegment(logDir, baseOffset)
+ try {
+ val nextOffset = writeSampleBatches(baseOffset, segment)
+ writeSampleBatches(nextOffset, segment)
+ } finally segment.close()
+ }
+
+ // We create three segments, the second of which contains offsets which overflow
+ var nextOffset = 0L
+ nextOffset = writeNormalSegment(nextOffset)
+ nextOffset = writeOverflowSegment(nextOffset)
+ writeNormalSegment(nextOffset)
+ }
+
+ def allRecords(log: Log): List[Record] = {
Review comment:
Could this be private?
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -2797,63 +2590,65 @@ object LogMetricNames {
}
}
-sealed trait SegmentDeletionReason {
- def logReason(log: Log, toDelete: List[LogSegment]): Unit
+abstract class SegmentDeletionReason {
+ def logReason(logger: Logging, toDelete: List[LogSegment]): Unit = {}
Review comment:
Hmm, this method is being used, but no subclass implements it. Are we losing logging because of this?
##########
File path: core/src/test/scala/unit/kafka/log/LogTestUtils.scala
##########
@@ -37,4 +47,246 @@ object LogTestUtils {
new LogSegment(ms, idx, timeIdx, txnIndex, offset, indexIntervalBytes, 0, time)
}
+
+ def createLogConfig(segmentMs: Long = Defaults.SegmentMs,
+ segmentBytes: Int = Defaults.SegmentSize,
+ retentionMs: Long = Defaults.RetentionMs,
+ retentionBytes: Long = Defaults.RetentionSize,
+ segmentJitterMs: Long = Defaults.SegmentJitterMs,
+ cleanupPolicy: String = Defaults.CleanupPolicy,
+ maxMessageBytes: Int = Defaults.MaxMessageSize,
+ indexIntervalBytes: Int = Defaults.IndexInterval,
+ segmentIndexBytes: Int = Defaults.MaxIndexSize,
+ messageFormatVersion: String = Defaults.MessageFormatVersion,
+ fileDeleteDelayMs: Long = Defaults.FileDeleteDelayMs): LogConfig = {
+ val logProps = new Properties()
+
+ logProps.put(LogConfig.SegmentMsProp, segmentMs: java.lang.Long)
+ logProps.put(LogConfig.SegmentBytesProp, segmentBytes: Integer)
+ logProps.put(LogConfig.RetentionMsProp, retentionMs: java.lang.Long)
+ logProps.put(LogConfig.RetentionBytesProp, retentionBytes: java.lang.Long)
+ logProps.put(LogConfig.SegmentJitterMsProp, segmentJitterMs: java.lang.Long)
+ logProps.put(LogConfig.CleanupPolicyProp, cleanupPolicy)
+ logProps.put(LogConfig.MaxMessageBytesProp, maxMessageBytes: Integer)
+ logProps.put(LogConfig.IndexIntervalBytesProp, indexIntervalBytes: Integer)
+ logProps.put(LogConfig.SegmentIndexBytesProp, segmentIndexBytes: Integer)
+ logProps.put(LogConfig.MessageFormatVersionProp, messageFormatVersion)
+ logProps.put(LogConfig.FileDeleteDelayMsProp, fileDeleteDelayMs: java.lang.Long)
+ LogConfig(logProps)
+ }
+
+ def createLog(dir: File,
+ config: LogConfig,
+ brokerTopicStats: BrokerTopicStats,
+ scheduler: Scheduler,
+ time: Time,
+ logStartOffset: Long = 0L,
+ recoveryPoint: Long = 0L,
+ maxProducerIdExpirationMs: Int = 60 * 60 * 1000,
+ producerIdExpirationCheckIntervalMs: Int = LogManager.ProducerIdExpirationCheckIntervalMs,
+ lastShutdownClean: Boolean = true,
+ topicId: Option[Uuid] = None): Log = {
+ Log(dir = dir,
+ config = config,
+ logStartOffset = logStartOffset,
+ recoveryPoint = recoveryPoint,
+ scheduler = scheduler,
+ brokerTopicStats = brokerTopicStats,
+ time = time,
+ maxProducerIdExpirationMs = maxProducerIdExpirationMs,
+ producerIdExpirationCheckIntervalMs = producerIdExpirationCheckIntervalMs,
+ logDirFailureChannel = new LogDirFailureChannel(10),
+ lastShutdownClean = lastShutdownClean,
+ topicId = topicId)
+ }
+
+ /**
+ * Check if the given log contains any segment with records that cause offset overflow.
+ * @param log Log to check
+ * @return true if log contains at least one segment with offset overflow; false otherwise
+ */
+ def hasOffsetOverflow(log: Log): Boolean = firstOverflowSegment(log).isDefined
+
+ def firstOverflowSegment(log: Log): Option[LogSegment] = {
+ def hasOverflow(baseOffset: Long, batch: RecordBatch): Boolean =
+ batch.lastOffset > baseOffset + Int.MaxValue || batch.baseOffset < baseOffset
+
+ for (segment <- log.logSegments) {
+ val overflowBatch = segment.log.batches.asScala.find(batch => hasOverflow(segment.baseOffset, batch))
+ if (overflowBatch.isDefined)
+ return Some(segment)
+ }
+ None
+ }
+
+ private def rawSegment(logDir: File, baseOffset: Long): FileRecords =
+ FileRecords.open(Log.logFile(logDir, baseOffset))
+
+ /**
+ * Initialize the given log directory with a set of segments, one of which will have an
+ * offset which overflows the segment
+ */
+ def initializeLogDirWithOverflowedSegment(logDir: File): Unit = {
+ def writeSampleBatches(baseOffset: Long, segment: FileRecords): Long = {
+ def record(offset: Long) = {
+ val data = offset.toString.getBytes
+ new SimpleRecord(data, data)
+ }
+
+ segment.append(MemoryRecords.withRecords(baseOffset, CompressionType.NONE, 0,
+ record(baseOffset)))
+ segment.append(MemoryRecords.withRecords(baseOffset + 1, CompressionType.NONE, 0,
+ record(baseOffset + 1),
+ record(baseOffset + 2)))
+ segment.append(MemoryRecords.withRecords(baseOffset + Int.MaxValue - 1, CompressionType.NONE, 0,
+ record(baseOffset + Int.MaxValue - 1)))
+ // Need to create the offset files explicitly to avoid triggering segment recovery to truncate segment.
+ Log.offsetIndexFile(logDir, baseOffset).createNewFile()
+ Log.timeIndexFile(logDir, baseOffset).createNewFile()
+ baseOffset + Int.MaxValue
+ }
+
+ def writeNormalSegment(baseOffset: Long): Long = {
+ val segment = rawSegment(logDir, baseOffset)
+ try writeSampleBatches(baseOffset, segment)
+ finally segment.close()
+ }
+
+ def writeOverflowSegment(baseOffset: Long): Long = {
+ val segment = rawSegment(logDir, baseOffset)
+ try {
+ val nextOffset = writeSampleBatches(baseOffset, segment)
+ writeSampleBatches(nextOffset, segment)
+ } finally segment.close()
+ }
+
+ // We create three segments, the second of which contains offsets which overflow
+ var nextOffset = 0L
+ nextOffset = writeNormalSegment(nextOffset)
+ nextOffset = writeOverflowSegment(nextOffset)
+ writeNormalSegment(nextOffset)
+ }
+
+ def allRecords(log: Log): List[Record] = {
+ val recordsFound = ListBuffer[Record]()
+ for (logSegment <- log.logSegments) {
+ for (batch <- logSegment.log.batches.asScala) {
+ recordsFound ++= batch.iterator().asScala
+ }
+ }
+ recordsFound.toList
+ }
+
+ def verifyRecordsInLog(log: Log, expectedRecords: List[Record]): Unit = {
Review comment:
Is this method used?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613804832
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
+ * directory failure
+ * @param hadCleanShutdown Boolean flag to indicate whether the associated log previously had a
+ * clean shutdown
+ * @param segments The LogSegments instance into which segments recovered from disk will be
+ * populated
+ * @param logStartOffsetCheckpoint The checkpoint of the log start offset
+ * @param recoveryPointCheckpoint The checkpoint of the offset at which to begin the recovery
+ * @param maxProducerIdExpirationMs The maximum amount of time to wait before a producer id is
+ * considered expired
+ * @param leaderEpochCache An optional LeaderEpochFileCache instance to be updated during recovery
+ * @param producerStateManager The ProducerStateManager instance to be updated during recovery
+ */
+case class LoadLogParams(dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ time: Time,
+ logDirFailureChannel: LogDirFailureChannel,
+ hadCleanShutdown: Boolean,
+ segments: LogSegments,
+ logStartOffsetCheckpoint: Long,
+ recoveryPointCheckpoint: Long,
+ maxProducerIdExpirationMs: Int,
+ leaderEpochCache: Option[LeaderEpochFileCache],
+ producerStateManager: ProducerStateManager) {
+ val logIdentifier: String = s"[LogLoader partition=$topicPartition, dir=${dir.getParent}]"
+}
+
+/**
+ * This object is responsible for all activities related with recovery of log segments from disk.
+ */
+object LogLoader extends Logging {
+ /**
+ * Load the log segments from the log files on disk, and return the components of the loaded log.
+ * Additionally, it also suitably updates the provided LeaderEpochFileCache and ProducerStateManager
+ * to reflect the contents of the loaded log.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is only
+ * called before all logs are loaded.
Review comment:
Agree this comment was a bit misleading. I've improved it now to say the following, do you feel its better?
```
In the context of the calling thread, this function does not need to convert
IOException to KafkaStorageException because it is only called before all logs are loaded.
```
In the recovery path, we always asynchronously delete segments. The `IOException` conversion happens only within the background work that's scheduled for segment deletions. https://github.com/apache/kafka/blob/193a9dfc44fa2c0df163a8518a804c4579d37e36/core/src/main/scala/kafka/log/Log.scala#L2374-L2389
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r611438309
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -255,19 +261,21 @@ case object SnapshotGenerated extends LogStartOffsetIncrementReason {
@threadsafe
class Log(@volatile private var _dir: File,
@volatile var config: LogConfig,
+ val segments: LogSegments,
@volatile var logStartOffset: Long,
@volatile var recoveryPoint: Long,
+ @volatile var nextOffsetMetadata: LogOffsetMetadata,
scheduler: Scheduler,
brokerTopicStats: BrokerTopicStats,
val time: Time,
val maxProducerIdExpirationMs: Int,
val producerIdExpirationCheckIntervalMs: Int,
val topicPartition: TopicPartition,
+ @volatile var leaderEpochCache: Option[LeaderEpochFileCache],
val producerStateManager: ProducerStateManager,
logDirFailureChannel: LogDirFailureChannel,
- private val hadCleanShutdown: Boolean = true,
@volatile var topicId: Option[Uuid],
- val keepPartitionMetadataFile: Boolean) extends Logging with KafkaMetricsGroup {
+ val keepPartitionMetadataFile: Boolean = true) extends Logging with KafkaMetricsGroup {
Review comment:
Done. I've removed it. It's not needed, I believe it showed up accidentally. My bad.
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -2586,11 +1999,15 @@ object Log {
logDirFailureChannel: LogDirFailureChannel,
lastShutdownClean: Boolean = true,
topicId: Option[Uuid],
- keepPartitionMetadataFile: Boolean): Log = {
+ keepPartitionMetadataFile: Boolean = true): Log = {
Review comment:
Done. I've removed it. It's not needed, I believe it showed up accidentally. My bad.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613804832
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
+ * directory failure
+ * @param hadCleanShutdown Boolean flag to indicate whether the associated log previously had a
+ * clean shutdown
+ * @param segments The LogSegments instance into which segments recovered from disk will be
+ * populated
+ * @param logStartOffsetCheckpoint The checkpoint of the log start offset
+ * @param recoveryPointCheckpoint The checkpoint of the offset at which to begin the recovery
+ * @param maxProducerIdExpirationMs The maximum amount of time to wait before a producer id is
+ * considered expired
+ * @param leaderEpochCache An optional LeaderEpochFileCache instance to be updated during recovery
+ * @param producerStateManager The ProducerStateManager instance to be updated during recovery
+ */
+case class LoadLogParams(dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ time: Time,
+ logDirFailureChannel: LogDirFailureChannel,
+ hadCleanShutdown: Boolean,
+ segments: LogSegments,
+ logStartOffsetCheckpoint: Long,
+ recoveryPointCheckpoint: Long,
+ maxProducerIdExpirationMs: Int,
+ leaderEpochCache: Option[LeaderEpochFileCache],
+ producerStateManager: ProducerStateManager) {
+ val logIdentifier: String = s"[LogLoader partition=$topicPartition, dir=${dir.getParent}]"
+}
+
+/**
+ * This object is responsible for all activities related with recovery of log segments from disk.
+ */
+object LogLoader extends Logging {
+ /**
+ * Load the log segments from the log files on disk, and return the components of the loaded log.
+ * Additionally, it also suitably updates the provided LeaderEpochFileCache and ProducerStateManager
+ * to reflect the contents of the loaded log.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is only
+ * called before all logs are loaded.
Review comment:
Agree this comment was a bit misleading. I've improved it now to say the following, do you feel its better?
```
In the context of the calling thread, this function does not need to convert
IOException to KafkaStorageException because it is only called before all logs are loaded.
```
In the recovery path, we always asynchronously delete segments. The `IOException` conversion happens only within the background work that's scheduled for segment deletions. https://github.com/apache/kafka/blob/193a9dfc44fa2c0df163a8518a804c4579d37e36/core/src/main/scala/kafka/log/Log.scala#L2377
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#issuecomment-817634015
@junrao @satishd: Thanks for the review! I've responded to your comments and addressed them in 8f068caa78a9998c14a18f4583eaaa8a90cba71c.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613486870
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
+ * directory failure
+ * @param hadCleanShutdown Boolean flag to indicate whether the associated log previously had a
+ * clean shutdown
+ * @param segments The LogSegments instance into which segments recovered from disk will be
+ * populated
+ * @param logStartOffsetCheckpoint The checkpoint of the log start offset
+ * @param recoveryPointCheckpoint The checkpoint of the offset at which to begin the recovery
+ * @param maxProducerIdExpirationMs The maximum amount of time to wait before a producer id is
+ * considered expired
+ * @param leaderEpochCache An optional LeaderEpochFileCache instance to be updated during recovery
+ * @param producerStateManager The ProducerStateManager instance to be updated during recovery
+ */
+case class LoadLogParams(dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ time: Time,
+ logDirFailureChannel: LogDirFailureChannel,
+ hadCleanShutdown: Boolean,
+ segments: LogSegments,
+ logStartOffsetCheckpoint: Long,
+ recoveryPointCheckpoint: Long,
+ maxProducerIdExpirationMs: Int,
+ leaderEpochCache: Option[LeaderEpochFileCache],
+ producerStateManager: ProducerStateManager) {
+ val logIdentifier: String = s"[LogLoader partition=$topicPartition, dir=${dir.getParent}]"
+}
+
+/**
+ * This object is responsible for all activities related with recovery of log segments from disk.
+ */
+object LogLoader extends Logging {
+ /**
+ * Load the log segments from the log files on disk, and return the components of the loaded log.
+ * Additionally, it also suitably updates the provided LeaderEpochFileCache and ProducerStateManager
+ * to reflect the contents of the loaded log.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is only
+ * called before all logs are loaded.
+ *
+ * @param params The parameters for the log being loaded from disk
+ *
+ * @return the offsets of the Log successfully loaded from disk
+ *
+ * @throws LogSegmentOffsetOverflowException if we encounter a .swap file with messages that
+ * overflow index offset; or when we find an unexpected
+ * number of .log files with overflow
Review comment:
It seems that this comment is no longer accurate. The only place that can throw LogSegmentOffsetOverflowException is from completeSwapOperations(). The rest of the places are all wrapped with retryOnOffsetOverflow().
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
+ * directory failure
+ * @param hadCleanShutdown Boolean flag to indicate whether the associated log previously had a
+ * clean shutdown
+ * @param segments The LogSegments instance into which segments recovered from disk will be
+ * populated
+ * @param logStartOffsetCheckpoint The checkpoint of the log start offset
+ * @param recoveryPointCheckpoint The checkpoint of the offset at which to begin the recovery
+ * @param maxProducerIdExpirationMs The maximum amount of time to wait before a producer id is
+ * considered expired
+ * @param leaderEpochCache An optional LeaderEpochFileCache instance to be updated during recovery
+ * @param producerStateManager The ProducerStateManager instance to be updated during recovery
+ */
+case class LoadLogParams(dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ time: Time,
+ logDirFailureChannel: LogDirFailureChannel,
+ hadCleanShutdown: Boolean,
+ segments: LogSegments,
+ logStartOffsetCheckpoint: Long,
+ recoveryPointCheckpoint: Long,
+ maxProducerIdExpirationMs: Int,
+ leaderEpochCache: Option[LeaderEpochFileCache],
+ producerStateManager: ProducerStateManager) {
+ val logIdentifier: String = s"[LogLoader partition=$topicPartition, dir=${dir.getParent}]"
+}
+
+/**
+ * This object is responsible for all activities related with recovery of log segments from disk.
+ */
+object LogLoader extends Logging {
+ /**
+ * Load the log segments from the log files on disk, and return the components of the loaded log.
+ * Additionally, it also suitably updates the provided LeaderEpochFileCache and ProducerStateManager
+ * to reflect the contents of the loaded log.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is only
+ * called before all logs are loaded.
+ *
+ * @param params The parameters for the log being loaded from disk
+ *
+ * @return the offsets of the Log successfully loaded from disk
+ *
+ * @throws LogSegmentOffsetOverflowException if we encounter a .swap file with messages that
+ * overflow index offset; or when we find an unexpected
+ * number of .log files with overflow
+ */
+ def load(params: LoadLogParams): LoadedLogOffsets = {
+ // first do a pass through the files in the log directory and remove any temporary files
+ // and find any interrupted swap operations
+ val swapFiles = removeTempFilesAndCollectSwapFiles(params)
+
+ // Now do a second pass and load all the log and index files.
+ // We might encounter legacy log segments with offset overflow (KAFKA-6264). We need to split such segments. When
+ // this happens, restart loading segment files from scratch.
+ retryOnOffsetOverflow(params, {
+ // In case we encounter a segment with offset overflow, the retry logic will split it after which we need to retry
+ // loading of segments. In that case, we also need to close all segments that could have been left open in previous
+ // call to loadSegmentFiles().
+ params.segments.close()
+ params.segments.clear()
+ loadSegmentFiles(params)
+ })
+
+ completeSwapOperations(swapFiles, params)
+
+ val (newRecoveryPoint: Long, nextOffset: Long) = {
+ if (!params.dir.getAbsolutePath.endsWith(Log.DeleteDirSuffix)) {
+ val (newRecoveryPoint, nextOffset) = retryOnOffsetOverflow(params, {
+ recoverLog(params)
+ })
+
+ // reset the index size of the currently active log segment to allow more entries
+ params.segments.lastSegment.get.resizeIndexes(params.config.maxIndexSize)
+ (newRecoveryPoint, nextOffset)
+ } else {
+ if (params.segments.isEmpty) {
+ params.segments.add(
+ LogSegment.open(
+ dir = params.dir,
+ baseOffset = 0,
+ params.config,
+ time = params.time,
+ initFileSize = params.config.initFileSize))
+ }
+ (params.recoveryPointCheckpoint, 0L)
Review comment:
Since the nextOffset is 0, recoveryPoint needs to be 0 too.
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
Review comment:
Currently, if we hit any IOException during loading, we just kill the broker. So, it seems that there is no need to pass in logDirFailureChannel during log loading.
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
+ * directory failure
+ * @param hadCleanShutdown Boolean flag to indicate whether the associated log previously had a
+ * clean shutdown
+ * @param segments The LogSegments instance into which segments recovered from disk will be
+ * populated
+ * @param logStartOffsetCheckpoint The checkpoint of the log start offset
+ * @param recoveryPointCheckpoint The checkpoint of the offset at which to begin the recovery
+ * @param maxProducerIdExpirationMs The maximum amount of time to wait before a producer id is
+ * considered expired
+ * @param leaderEpochCache An optional LeaderEpochFileCache instance to be updated during recovery
+ * @param producerStateManager The ProducerStateManager instance to be updated during recovery
+ */
+case class LoadLogParams(dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ time: Time,
+ logDirFailureChannel: LogDirFailureChannel,
+ hadCleanShutdown: Boolean,
+ segments: LogSegments,
+ logStartOffsetCheckpoint: Long,
+ recoveryPointCheckpoint: Long,
+ maxProducerIdExpirationMs: Int,
+ leaderEpochCache: Option[LeaderEpochFileCache],
+ producerStateManager: ProducerStateManager) {
+ val logIdentifier: String = s"[LogLoader partition=$topicPartition, dir=${dir.getParent}]"
+}
+
+/**
+ * This object is responsible for all activities related with recovery of log segments from disk.
+ */
+object LogLoader extends Logging {
+ /**
+ * Load the log segments from the log files on disk, and return the components of the loaded log.
+ * Additionally, it also suitably updates the provided LeaderEpochFileCache and ProducerStateManager
+ * to reflect the contents of the loaded log.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is only
+ * called before all logs are loaded.
Review comment:
The loading code still converts IOException to KafkaStorageException through Log.deleteSegmentFiles().
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -222,25 +222,31 @@ case object SnapshotGenerated extends LogStartOffsetIncrementReason {
*
* @param _dir The directory in which log segments are created.
* @param config The log configuration settings
+ * @param segments The log segments, these may be non-empty when recovered from disk
Review comment:
The segments is always non-empty since we initialize it with an empty segment if no segments are present.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613795734
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -255,19 +261,21 @@ case object SnapshotGenerated extends LogStartOffsetIncrementReason {
@threadsafe
class Log(@volatile private var _dir: File,
@volatile var config: LogConfig,
+ val segments: LogSegments,
Review comment:
I took a stab at this. Things changed a bit after I did the refactoring that Jun suggested [here](https://github.com/apache/kafka/pull/10478#discussion_r610933966). In the latest version of the PR, `LogComponents` has been renamed to a class called `LoadedLogOffsets` which now only contains (logStartOffset, recoveryPoint, nextOffsetMetadata). It seemed reasonable to just pass these in as 3 separate parameters into `Log` since the `LeaderEpochFileCache` and `ProducerStateManager` are no longer inside the `LoadedLogOffsets` object, so those have to be passed separately anyway.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik edited a comment on pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik edited a comment on pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#issuecomment-821333025
@junrao: Thanks for the review! I have addressed the latest comments and also fixed the broken unit test in `PartitionLockTest` suite. This PR is ready for review.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] satishd commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
satishd commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r611288890
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -255,19 +261,21 @@ case object SnapshotGenerated extends LogStartOffsetIncrementReason {
@threadsafe
class Log(@volatile private var _dir: File,
@volatile var config: LogConfig,
+ val segments: LogSegments,
Review comment:
Can we pass `LogComponents` as a single field instead of initializing with the respective fields? This makes the code easy to read, constructor simpler with less vars.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613787194
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
+ * directory failure
+ * @param hadCleanShutdown Boolean flag to indicate whether the associated log previously had a
+ * clean shutdown
+ * @param segments The LogSegments instance into which segments recovered from disk will be
+ * populated
+ * @param logStartOffsetCheckpoint The checkpoint of the log start offset
+ * @param recoveryPointCheckpoint The checkpoint of the offset at which to begin the recovery
+ * @param maxProducerIdExpirationMs The maximum amount of time to wait before a producer id is
+ * considered expired
+ * @param leaderEpochCache An optional LeaderEpochFileCache instance to be updated during recovery
+ * @param producerStateManager The ProducerStateManager instance to be updated during recovery
+ */
+case class LoadLogParams(dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ time: Time,
+ logDirFailureChannel: LogDirFailureChannel,
+ hadCleanShutdown: Boolean,
+ segments: LogSegments,
+ logStartOffsetCheckpoint: Long,
+ recoveryPointCheckpoint: Long,
+ maxProducerIdExpirationMs: Int,
+ leaderEpochCache: Option[LeaderEpochFileCache],
+ producerStateManager: ProducerStateManager) {
+ val logIdentifier: String = s"[LogLoader partition=$topicPartition, dir=${dir.getParent}]"
+}
+
+/**
+ * This object is responsible for all activities related with recovery of log segments from disk.
+ */
+object LogLoader extends Logging {
+ /**
+ * Load the log segments from the log files on disk, and return the components of the loaded log.
+ * Additionally, it also suitably updates the provided LeaderEpochFileCache and ProducerStateManager
+ * to reflect the contents of the loaded log.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is only
+ * called before all logs are loaded.
+ *
+ * @param params The parameters for the log being loaded from disk
+ *
+ * @return the offsets of the Log successfully loaded from disk
+ *
+ * @throws LogSegmentOffsetOverflowException if we encounter a .swap file with messages that
+ * overflow index offset; or when we find an unexpected
+ * number of .log files with overflow
Review comment:
Done. I've fixed it.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#issuecomment-815353566
cc @dhruvilshah3 @junrao @satishd -- this PR is ready for a review.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r611830920
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -255,19 +261,21 @@ case object SnapshotGenerated extends LogStartOffsetIncrementReason {
@threadsafe
class Log(@volatile private var _dir: File,
@volatile var config: LogConfig,
+ val segments: LogSegments,
Review comment:
Sounds good. I'll give this a shot.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613799479
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
Review comment:
In the recovery path, the `logDirFailureChannel` is used to asynchronously handle `IOException` within the `Log.deleteSegmentFiles()` calls. This was the existing behavior prior to this refactor as well. Should we change the existing behavior?https://github.com/apache/kafka/blob/193a9dfc44fa2c0df163a8518a804c4579d37e36/core/src/main/scala/kafka/log/Log.scala#L2374-L2384.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613799479
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
Review comment:
In the recovery path, the `logDirFailureChannel` is used to asynchronously handle `IOException` within the `Log.deleteSegmentFiles()` calls. This was the existing behavior prior to this refactor as well. Should we change the existing behavior?https://github.com/apache/kafka/blob/193a9dfc44fa2c0df163a8518a804c4579d37e36/core/src/main/scala/kafka/log/Log.scala#L2374-L2384
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r611437760
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -2797,63 +2590,65 @@ object LogMetricNames {
}
}
-sealed trait SegmentDeletionReason {
- def logReason(log: Log, toDelete: List[LogSegment]): Unit
+abstract class SegmentDeletionReason {
+ def logReason(logger: Logging, toDelete: List[LogSegment]): Unit = {}
Review comment:
Great catch. After an investigation, I realized this change is no longer required and I've reverted it now.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613787269
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
+ * directory failure
+ * @param hadCleanShutdown Boolean flag to indicate whether the associated log previously had a
+ * clean shutdown
+ * @param segments The LogSegments instance into which segments recovered from disk will be
+ * populated
+ * @param logStartOffsetCheckpoint The checkpoint of the log start offset
+ * @param recoveryPointCheckpoint The checkpoint of the offset at which to begin the recovery
+ * @param maxProducerIdExpirationMs The maximum amount of time to wait before a producer id is
+ * considered expired
+ * @param leaderEpochCache An optional LeaderEpochFileCache instance to be updated during recovery
+ * @param producerStateManager The ProducerStateManager instance to be updated during recovery
+ */
+case class LoadLogParams(dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ time: Time,
+ logDirFailureChannel: LogDirFailureChannel,
+ hadCleanShutdown: Boolean,
+ segments: LogSegments,
+ logStartOffsetCheckpoint: Long,
+ recoveryPointCheckpoint: Long,
+ maxProducerIdExpirationMs: Int,
+ leaderEpochCache: Option[LeaderEpochFileCache],
+ producerStateManager: ProducerStateManager) {
+ val logIdentifier: String = s"[LogLoader partition=$topicPartition, dir=${dir.getParent}]"
+}
+
+/**
+ * This object is responsible for all activities related with recovery of log segments from disk.
+ */
+object LogLoader extends Logging {
+ /**
+ * Load the log segments from the log files on disk, and return the components of the loaded log.
+ * Additionally, it also suitably updates the provided LeaderEpochFileCache and ProducerStateManager
+ * to reflect the contents of the loaded log.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is only
+ * called before all logs are loaded.
+ *
+ * @param params The parameters for the log being loaded from disk
+ *
+ * @return the offsets of the Log successfully loaded from disk
+ *
+ * @throws LogSegmentOffsetOverflowException if we encounter a .swap file with messages that
+ * overflow index offset; or when we find an unexpected
+ * number of .log files with overflow
+ */
+ def load(params: LoadLogParams): LoadedLogOffsets = {
+ // first do a pass through the files in the log directory and remove any temporary files
+ // and find any interrupted swap operations
+ val swapFiles = removeTempFilesAndCollectSwapFiles(params)
+
+ // Now do a second pass and load all the log and index files.
+ // We might encounter legacy log segments with offset overflow (KAFKA-6264). We need to split such segments. When
+ // this happens, restart loading segment files from scratch.
+ retryOnOffsetOverflow(params, {
+ // In case we encounter a segment with offset overflow, the retry logic will split it after which we need to retry
+ // loading of segments. In that case, we also need to close all segments that could have been left open in previous
+ // call to loadSegmentFiles().
+ params.segments.close()
+ params.segments.clear()
+ loadSegmentFiles(params)
+ })
+
+ completeSwapOperations(swapFiles, params)
+
+ val (newRecoveryPoint: Long, nextOffset: Long) = {
+ if (!params.dir.getAbsolutePath.endsWith(Log.DeleteDirSuffix)) {
+ val (newRecoveryPoint, nextOffset) = retryOnOffsetOverflow(params, {
+ recoverLog(params)
+ })
+
+ // reset the index size of the currently active log segment to allow more entries
+ params.segments.lastSegment.get.resizeIndexes(params.config.maxIndexSize)
+ (newRecoveryPoint, nextOffset)
+ } else {
+ if (params.segments.isEmpty) {
+ params.segments.add(
+ LogSegment.open(
+ dir = params.dir,
+ baseOffset = 0,
+ params.config,
+ time = params.time,
+ initFileSize = params.config.initFileSize))
+ }
+ (params.recoveryPointCheckpoint, 0L)
Review comment:
Great catch. I've fixed it now.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] satishd commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
satishd commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r616265276
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -255,19 +261,21 @@ case object SnapshotGenerated extends LogStartOffsetIncrementReason {
@threadsafe
class Log(@volatile private var _dir: File,
@volatile var config: LogConfig,
+ val segments: LogSegments,
Review comment:
Sure, that looks reasonable to me.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613804832
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
+ * directory failure
+ * @param hadCleanShutdown Boolean flag to indicate whether the associated log previously had a
+ * clean shutdown
+ * @param segments The LogSegments instance into which segments recovered from disk will be
+ * populated
+ * @param logStartOffsetCheckpoint The checkpoint of the log start offset
+ * @param recoveryPointCheckpoint The checkpoint of the offset at which to begin the recovery
+ * @param maxProducerIdExpirationMs The maximum amount of time to wait before a producer id is
+ * considered expired
+ * @param leaderEpochCache An optional LeaderEpochFileCache instance to be updated during recovery
+ * @param producerStateManager The ProducerStateManager instance to be updated during recovery
+ */
+case class LoadLogParams(dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ time: Time,
+ logDirFailureChannel: LogDirFailureChannel,
+ hadCleanShutdown: Boolean,
+ segments: LogSegments,
+ logStartOffsetCheckpoint: Long,
+ recoveryPointCheckpoint: Long,
+ maxProducerIdExpirationMs: Int,
+ leaderEpochCache: Option[LeaderEpochFileCache],
+ producerStateManager: ProducerStateManager) {
+ val logIdentifier: String = s"[LogLoader partition=$topicPartition, dir=${dir.getParent}]"
+}
+
+/**
+ * This object is responsible for all activities related with recovery of log segments from disk.
+ */
+object LogLoader extends Logging {
+ /**
+ * Load the log segments from the log files on disk, and return the components of the loaded log.
+ * Additionally, it also suitably updates the provided LeaderEpochFileCache and ProducerStateManager
+ * to reflect the contents of the loaded log.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is only
+ * called before all logs are loaded.
Review comment:
Thise comment was trying to convey that in the context of the calling thread, this function doesn't convert the `IOException` to `KafkaStorageException`. In the recovery path, we always asynchronously delete segments. The `IOException` conversion happens only within the background work that's scheduled for segment deletions. https://github.com/apache/kafka/blob/193a9dfc44fa2c0df163a8518a804c4579d37e36/core/src/main/scala/kafka/log/Log.scala#L2374-L2389
In case this is misleading, does the below improved comment feel better to you?
```
In the context of the calling thread, this function does not need to convert
IOException to KafkaStorageException because it is only called before all logs are loaded.
```
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#issuecomment-823118678
Thanks for the review @junrao and @satishd! I've updated this PR after rebasing onto latest commit in trunk.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#issuecomment-823441394
I checked the failed tests in CI this morning and none of them are related with this PR:
```
org.apache.kafka.connect.*
kafka.admin.ResetConsumerGroupOffsetTest.testResetOffsetsToZonedDateTime()
kafka.server.RaftClusterTest.testCreateClusterAndCreateListDeleteTopic()
kafka.integration.MetricsDuringTopicCreationDeletionTest.testMetricsDuringTopicCreateDelete()
```
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613804832
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
+ * directory failure
+ * @param hadCleanShutdown Boolean flag to indicate whether the associated log previously had a
+ * clean shutdown
+ * @param segments The LogSegments instance into which segments recovered from disk will be
+ * populated
+ * @param logStartOffsetCheckpoint The checkpoint of the log start offset
+ * @param recoveryPointCheckpoint The checkpoint of the offset at which to begin the recovery
+ * @param maxProducerIdExpirationMs The maximum amount of time to wait before a producer id is
+ * considered expired
+ * @param leaderEpochCache An optional LeaderEpochFileCache instance to be updated during recovery
+ * @param producerStateManager The ProducerStateManager instance to be updated during recovery
+ */
+case class LoadLogParams(dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ time: Time,
+ logDirFailureChannel: LogDirFailureChannel,
+ hadCleanShutdown: Boolean,
+ segments: LogSegments,
+ logStartOffsetCheckpoint: Long,
+ recoveryPointCheckpoint: Long,
+ maxProducerIdExpirationMs: Int,
+ leaderEpochCache: Option[LeaderEpochFileCache],
+ producerStateManager: ProducerStateManager) {
+ val logIdentifier: String = s"[LogLoader partition=$topicPartition, dir=${dir.getParent}]"
+}
+
+/**
+ * This object is responsible for all activities related with recovery of log segments from disk.
+ */
+object LogLoader extends Logging {
+ /**
+ * Load the log segments from the log files on disk, and return the components of the loaded log.
+ * Additionally, it also suitably updates the provided LeaderEpochFileCache and ProducerStateManager
+ * to reflect the contents of the loaded log.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is only
+ * called before all logs are loaded.
Review comment:
Agree this comment was a bit misleading. I've improved it now to say:
```
In the context of the calling thread, this function does not need to convert
IOException to KafkaStorageException because it is only called before all logs are loaded.
```
In the recovery path, we always asynchronously delete segments. The `IOException` conversion happens only within the background work that's scheduled for segment deletions. https://github.com/apache/kafka/blob/193a9dfc44fa2c0df163a8518a804c4579d37e36/core/src/main/scala/kafka/log/Log.scala#L2374-L2389
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613804832
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
+ * directory failure
+ * @param hadCleanShutdown Boolean flag to indicate whether the associated log previously had a
+ * clean shutdown
+ * @param segments The LogSegments instance into which segments recovered from disk will be
+ * populated
+ * @param logStartOffsetCheckpoint The checkpoint of the log start offset
+ * @param recoveryPointCheckpoint The checkpoint of the offset at which to begin the recovery
+ * @param maxProducerIdExpirationMs The maximum amount of time to wait before a producer id is
+ * considered expired
+ * @param leaderEpochCache An optional LeaderEpochFileCache instance to be updated during recovery
+ * @param producerStateManager The ProducerStateManager instance to be updated during recovery
+ */
+case class LoadLogParams(dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ time: Time,
+ logDirFailureChannel: LogDirFailureChannel,
+ hadCleanShutdown: Boolean,
+ segments: LogSegments,
+ logStartOffsetCheckpoint: Long,
+ recoveryPointCheckpoint: Long,
+ maxProducerIdExpirationMs: Int,
+ leaderEpochCache: Option[LeaderEpochFileCache],
+ producerStateManager: ProducerStateManager) {
+ val logIdentifier: String = s"[LogLoader partition=$topicPartition, dir=${dir.getParent}]"
+}
+
+/**
+ * This object is responsible for all activities related with recovery of log segments from disk.
+ */
+object LogLoader extends Logging {
+ /**
+ * Load the log segments from the log files on disk, and return the components of the loaded log.
+ * Additionally, it also suitably updates the provided LeaderEpochFileCache and ProducerStateManager
+ * to reflect the contents of the loaded log.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is only
+ * called before all logs are loaded.
Review comment:
Thise comment was trying to convey that in the context of the calling thread, this function doesn't convert the IOException to KafkaStorageException. In the recovery path, we always asynchronously delete segments. The IOException conversion happens only within the background work that's scheduled for segment deletions. https://github.com/apache/kafka/blob/193a9dfc44fa2c0df163a8518a804c4579d37e36/core/src/main/scala/kafka/log/Log.scala#L2374-L2389
In case this is misleading, does the below improved comment feel better to you?
```
In the context of the calling thread, this function does not need to convert
IOException to KafkaStorageException because it is only called before all logs are loaded.
```
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r611433821
##########
File path: core/src/test/scala/unit/kafka/log/LogTestUtils.scala
##########
@@ -37,4 +47,246 @@ object LogTestUtils {
new LogSegment(ms, idx, timeIdx, txnIndex, offset, indexIntervalBytes, 0, time)
}
+
+ def createLogConfig(segmentMs: Long = Defaults.SegmentMs,
+ segmentBytes: Int = Defaults.SegmentSize,
+ retentionMs: Long = Defaults.RetentionMs,
+ retentionBytes: Long = Defaults.RetentionSize,
+ segmentJitterMs: Long = Defaults.SegmentJitterMs,
+ cleanupPolicy: String = Defaults.CleanupPolicy,
+ maxMessageBytes: Int = Defaults.MaxMessageSize,
+ indexIntervalBytes: Int = Defaults.IndexInterval,
+ segmentIndexBytes: Int = Defaults.MaxIndexSize,
+ messageFormatVersion: String = Defaults.MessageFormatVersion,
+ fileDeleteDelayMs: Long = Defaults.FileDeleteDelayMs): LogConfig = {
+ val logProps = new Properties()
+
+ logProps.put(LogConfig.SegmentMsProp, segmentMs: java.lang.Long)
+ logProps.put(LogConfig.SegmentBytesProp, segmentBytes: Integer)
+ logProps.put(LogConfig.RetentionMsProp, retentionMs: java.lang.Long)
+ logProps.put(LogConfig.RetentionBytesProp, retentionBytes: java.lang.Long)
+ logProps.put(LogConfig.SegmentJitterMsProp, segmentJitterMs: java.lang.Long)
+ logProps.put(LogConfig.CleanupPolicyProp, cleanupPolicy)
+ logProps.put(LogConfig.MaxMessageBytesProp, maxMessageBytes: Integer)
+ logProps.put(LogConfig.IndexIntervalBytesProp, indexIntervalBytes: Integer)
+ logProps.put(LogConfig.SegmentIndexBytesProp, segmentIndexBytes: Integer)
+ logProps.put(LogConfig.MessageFormatVersionProp, messageFormatVersion)
+ logProps.put(LogConfig.FileDeleteDelayMsProp, fileDeleteDelayMs: java.lang.Long)
+ LogConfig(logProps)
+ }
+
+ def createLog(dir: File,
+ config: LogConfig,
+ brokerTopicStats: BrokerTopicStats,
+ scheduler: Scheduler,
+ time: Time,
+ logStartOffset: Long = 0L,
+ recoveryPoint: Long = 0L,
+ maxProducerIdExpirationMs: Int = 60 * 60 * 1000,
+ producerIdExpirationCheckIntervalMs: Int = LogManager.ProducerIdExpirationCheckIntervalMs,
+ lastShutdownClean: Boolean = true,
+ topicId: Option[Uuid] = None): Log = {
+ Log(dir = dir,
+ config = config,
+ logStartOffset = logStartOffset,
+ recoveryPoint = recoveryPoint,
+ scheduler = scheduler,
+ brokerTopicStats = brokerTopicStats,
+ time = time,
+ maxProducerIdExpirationMs = maxProducerIdExpirationMs,
+ producerIdExpirationCheckIntervalMs = producerIdExpirationCheckIntervalMs,
+ logDirFailureChannel = new LogDirFailureChannel(10),
+ lastShutdownClean = lastShutdownClean,
+ topicId = topicId)
+ }
+
+ /**
+ * Check if the given log contains any segment with records that cause offset overflow.
+ * @param log Log to check
+ * @return true if log contains at least one segment with offset overflow; false otherwise
+ */
+ def hasOffsetOverflow(log: Log): Boolean = firstOverflowSegment(log).isDefined
+
+ def firstOverflowSegment(log: Log): Option[LogSegment] = {
+ def hasOverflow(baseOffset: Long, batch: RecordBatch): Boolean =
+ batch.lastOffset > baseOffset + Int.MaxValue || batch.baseOffset < baseOffset
+
+ for (segment <- log.logSegments) {
+ val overflowBatch = segment.log.batches.asScala.find(batch => hasOverflow(segment.baseOffset, batch))
+ if (overflowBatch.isDefined)
+ return Some(segment)
+ }
+ None
+ }
+
+ private def rawSegment(logDir: File, baseOffset: Long): FileRecords =
+ FileRecords.open(Log.logFile(logDir, baseOffset))
+
+ /**
+ * Initialize the given log directory with a set of segments, one of which will have an
+ * offset which overflows the segment
+ */
+ def initializeLogDirWithOverflowedSegment(logDir: File): Unit = {
+ def writeSampleBatches(baseOffset: Long, segment: FileRecords): Long = {
+ def record(offset: Long) = {
+ val data = offset.toString.getBytes
+ new SimpleRecord(data, data)
+ }
+
+ segment.append(MemoryRecords.withRecords(baseOffset, CompressionType.NONE, 0,
+ record(baseOffset)))
+ segment.append(MemoryRecords.withRecords(baseOffset + 1, CompressionType.NONE, 0,
+ record(baseOffset + 1),
+ record(baseOffset + 2)))
+ segment.append(MemoryRecords.withRecords(baseOffset + Int.MaxValue - 1, CompressionType.NONE, 0,
+ record(baseOffset + Int.MaxValue - 1)))
+ // Need to create the offset files explicitly to avoid triggering segment recovery to truncate segment.
+ Log.offsetIndexFile(logDir, baseOffset).createNewFile()
+ Log.timeIndexFile(logDir, baseOffset).createNewFile()
+ baseOffset + Int.MaxValue
+ }
+
+ def writeNormalSegment(baseOffset: Long): Long = {
+ val segment = rawSegment(logDir, baseOffset)
+ try writeSampleBatches(baseOffset, segment)
+ finally segment.close()
+ }
+
+ def writeOverflowSegment(baseOffset: Long): Long = {
+ val segment = rawSegment(logDir, baseOffset)
+ try {
+ val nextOffset = writeSampleBatches(baseOffset, segment)
+ writeSampleBatches(nextOffset, segment)
+ } finally segment.close()
+ }
+
+ // We create three segments, the second of which contains offsets which overflow
+ var nextOffset = 0L
+ nextOffset = writeNormalSegment(nextOffset)
+ nextOffset = writeOverflowSegment(nextOffset)
+ writeNormalSegment(nextOffset)
+ }
+
+ def allRecords(log: Log): List[Record] = {
Review comment:
Done. With `def verifyRecordsInLog` eliminated, I was able to eliminate this one too. Related comment: https://github.com/apache/kafka/pull/10478#discussion_r610969662.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] satishd commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
satishd commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r611789255
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -255,19 +261,21 @@ case object SnapshotGenerated extends LogStartOffsetIncrementReason {
@threadsafe
class Log(@volatile private var _dir: File,
@volatile var config: LogConfig,
+ val segments: LogSegments,
Review comment:
What I meant was we can have `Log` takes the argument as immutable `LogComponents` and it can initialize the vars inside `Log` with the respective fields from `LogComponents`. This will also set the right access of these vars by not giving write access by default.
```
class Log(@volatile private var _dir: File,
@volatile var config: LogConfig,
val segments: LogSegments,
val logComponents: LogComponents,
scheduler: Scheduler,
brokerTopicStats: BrokerTopicStats,
val time: Time,
val maxProducerIdExpirationMs: Int,
val producerIdExpirationCheckIntervalMs: Int,
val topicPartition: TopicPartition,
logDirFailureChannel: LogDirFailureChannel,
@volatile var topicId: Option[Uuid],
val keepPartitionMetadataFile: Boolean = true) extends Logging with KafkaMetricsGroup {
@volatile private var logStartOffset: Long = logComponents.logStartOffset
@volatile private var recoveryPoint: Long = logComponents.recoveryPoint
@volatile private var nextOffsetMetadata: LogOffsetMetadata = logComponents.nextOffsetMetadata
@volatile var leaderEpochCache: Option[LeaderEpochFileCache] = logComponents.leaderEpochCache
private val producerStateManager: ProducerStateManager = logComponents.producerStateManager
```
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613799479
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
Review comment:
In the recovery path, the `logDirFailureChannel` is used to asynchronously handle IOException within the `Log.deleteSegmentFiles()` calls. This was the existing behavior prior to this refactor as well. Should we change the existing behavior?https://github.com/apache/kafka/blob/193a9dfc44fa2c0df163a8518a804c4579d37e36/core/src/main/scala/kafka/log/Log.scala#L2374-L2384.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r611450200
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -2586,11 +1999,15 @@ object Log {
logDirFailureChannel: LogDirFailureChannel,
lastShutdownClean: Boolean = true,
topicId: Option[Uuid],
- keepPartitionMetadataFile: Boolean): Log = {
+ keepPartitionMetadataFile: Boolean = true): Log = {
+ // create the log directory if it doesn't exist
+ Files.createDirectories(dir.toPath)
val topicPartition = Log.parseTopicPartitionName(dir)
- val producerStateManager = new ProducerStateManager(topicPartition, dir, maxProducerIdExpirationMs)
- new Log(dir, config, logStartOffset, recoveryPoint, scheduler, brokerTopicStats, time, maxProducerIdExpirationMs,
- producerIdExpirationCheckIntervalMs, topicPartition, producerStateManager, logDirFailureChannel, lastShutdownClean, topicId, keepPartitionMetadataFile)
+ val logLoader = new LogLoader(dir, topicPartition, config, scheduler, time, logDirFailureChannel)
Review comment:
Great point. I tried this, but few problems I ran into were:
1. A test tries to intercept the recovery logic by overriding `def recoverLog` in `LogLoader`. Ex: `LogLoaderTest.testLogRecoveryIsCalledUponBrokerCrash`.
2. Some tests like to inject a custom `ProducerStateManager`, ex: `LogLoaderTest.testSkipTruncateAndReloadIfNewMessageFormatAndCleanShutdown`.
3. Some tests like to inject a custom `LogSegments`, ex: `LogLoaderTest. testProducerSnapshotsRecoveryAfterUncleanShutdownCurrentMessageFormat`.
I'm working on solving these issues, and will post a PR update soon.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r611450200
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -2586,11 +1999,15 @@ object Log {
logDirFailureChannel: LogDirFailureChannel,
lastShutdownClean: Boolean = true,
topicId: Option[Uuid],
- keepPartitionMetadataFile: Boolean): Log = {
+ keepPartitionMetadataFile: Boolean = true): Log = {
+ // create the log directory if it doesn't exist
+ Files.createDirectories(dir.toPath)
val topicPartition = Log.parseTopicPartitionName(dir)
- val producerStateManager = new ProducerStateManager(topicPartition, dir, maxProducerIdExpirationMs)
- new Log(dir, config, logStartOffset, recoveryPoint, scheduler, brokerTopicStats, time, maxProducerIdExpirationMs,
- producerIdExpirationCheckIntervalMs, topicPartition, producerStateManager, logDirFailureChannel, lastShutdownClean, topicId, keepPartitionMetadataFile)
+ val logLoader = new LogLoader(dir, topicPartition, config, scheduler, time, logDirFailureChannel)
Review comment:
I tried this, but the problems I ran into were:
1. A test tries to intercept the recovery logic by overriding `def recoverLog` in `LogLoader`. Ex: `LogLoaderTest.testLogRecoveryIsCalledUponBrokerCrash`.
2. Some tests like to inject a custom `ProducerStateManager`, ex: `LogLoaderTest.testSkipTruncateAndReloadIfNewMessageFormatAndCleanShutdown`.
3. Some tests like to inject a custom `LogSegments`, ex: `LogLoaderTest. testProducerSnapshotsRecoveryAfterUncleanShutdownCurrentMessageFormat`.
I couldn't simplify (1) because the test logic seems to be very specific and dependent on the details of the interception. When `LogLoader` is a class, such interception is possible easily. Otherwise if we make it an object, it seems trickier to intercept.
If we can simplify (1) to avoid the interception, then (2) and (3) can be simplified because we can always pass the objects into `load()` as parameters. But I couldn't find a better way for (1) so far.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613679576
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -2586,11 +1999,15 @@ object Log {
logDirFailureChannel: LogDirFailureChannel,
lastShutdownClean: Boolean = true,
topicId: Option[Uuid],
- keepPartitionMetadataFile: Boolean): Log = {
+ keepPartitionMetadataFile: Boolean = true): Log = {
+ // create the log directory if it doesn't exist
+ Files.createDirectories(dir.toPath)
val topicPartition = Log.parseTopicPartitionName(dir)
- val producerStateManager = new ProducerStateManager(topicPartition, dir, maxProducerIdExpirationMs)
- new Log(dir, config, logStartOffset, recoveryPoint, scheduler, brokerTopicStats, time, maxProducerIdExpirationMs,
- producerIdExpirationCheckIntervalMs, topicPartition, producerStateManager, logDirFailureChannel, lastShutdownClean, topicId, keepPartitionMetadataFile)
+ val logLoader = new LogLoader(dir, topicPartition, config, scheduler, time, logDirFailureChannel)
Review comment:
Done. I've updated the PR converting LogLoader into an object.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613804832
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
+ * directory failure
+ * @param hadCleanShutdown Boolean flag to indicate whether the associated log previously had a
+ * clean shutdown
+ * @param segments The LogSegments instance into which segments recovered from disk will be
+ * populated
+ * @param logStartOffsetCheckpoint The checkpoint of the log start offset
+ * @param recoveryPointCheckpoint The checkpoint of the offset at which to begin the recovery
+ * @param maxProducerIdExpirationMs The maximum amount of time to wait before a producer id is
+ * considered expired
+ * @param leaderEpochCache An optional LeaderEpochFileCache instance to be updated during recovery
+ * @param producerStateManager The ProducerStateManager instance to be updated during recovery
+ */
+case class LoadLogParams(dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ time: Time,
+ logDirFailureChannel: LogDirFailureChannel,
+ hadCleanShutdown: Boolean,
+ segments: LogSegments,
+ logStartOffsetCheckpoint: Long,
+ recoveryPointCheckpoint: Long,
+ maxProducerIdExpirationMs: Int,
+ leaderEpochCache: Option[LeaderEpochFileCache],
+ producerStateManager: ProducerStateManager) {
+ val logIdentifier: String = s"[LogLoader partition=$topicPartition, dir=${dir.getParent}]"
+}
+
+/**
+ * This object is responsible for all activities related with recovery of log segments from disk.
+ */
+object LogLoader extends Logging {
+ /**
+ * Load the log segments from the log files on disk, and return the components of the loaded log.
+ * Additionally, it also suitably updates the provided LeaderEpochFileCache and ProducerStateManager
+ * to reflect the contents of the loaded log.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is only
+ * called before all logs are loaded.
Review comment:
This comment was trying to convey that in the context of the calling thread, this function doesn't convert the `IOException` to `KafkaStorageException`. In the recovery path, we always asynchronously delete segments. The `IOException` conversion happens only within the background work that's scheduled for segment deletions. https://github.com/apache/kafka/blob/193a9dfc44fa2c0df163a8518a804c4579d37e36/core/src/main/scala/kafka/log/Log.scala#L2374-L2389
Since this is a bit misleading, I've improved the comment now to say:
```
In the context of the calling thread, this function does not need to convert
IOException to KafkaStorageException because it is only called before all logs are loaded.
```
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613795734
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -255,19 +261,21 @@ case object SnapshotGenerated extends LogStartOffsetIncrementReason {
@threadsafe
class Log(@volatile private var _dir: File,
@volatile var config: LogConfig,
+ val segments: LogSegments,
Review comment:
I took a stab at this. Things changed a bit after I did the refactoring that Jun suggested [here](https://github.com/apache/kafka/pull/10478#discussion_r610933966). In the latest version of the PR, `LogComponents` has been renamed to a class called `LoadedLogOffsets` which now only contains (logStartOffset, recoveryPoint, nextOffsetMetadata). It seemed reasonable to just pass these in as 3 separate parameters into `Log`, since the `LeaderEpochFileCache` and `ProducerStateManager` are no longer inside the `LoadedLogOffsets` object those have to anyway be passed in separately.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik edited a comment on pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik edited a comment on pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#issuecomment-817634015
@junrao @satishd: Thanks for the review! I've responded to your comments and addressed them in 8f068caa78a9998c14a18f4583eaaa8a90cba71c, except for [this comment](https://github.com/apache/kafka/pull/10478#discussion_r610933966) - I'm still working on this and will update the PR soon.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613804832
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
+ * directory failure
+ * @param hadCleanShutdown Boolean flag to indicate whether the associated log previously had a
+ * clean shutdown
+ * @param segments The LogSegments instance into which segments recovered from disk will be
+ * populated
+ * @param logStartOffsetCheckpoint The checkpoint of the log start offset
+ * @param recoveryPointCheckpoint The checkpoint of the offset at which to begin the recovery
+ * @param maxProducerIdExpirationMs The maximum amount of time to wait before a producer id is
+ * considered expired
+ * @param leaderEpochCache An optional LeaderEpochFileCache instance to be updated during recovery
+ * @param producerStateManager The ProducerStateManager instance to be updated during recovery
+ */
+case class LoadLogParams(dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ time: Time,
+ logDirFailureChannel: LogDirFailureChannel,
+ hadCleanShutdown: Boolean,
+ segments: LogSegments,
+ logStartOffsetCheckpoint: Long,
+ recoveryPointCheckpoint: Long,
+ maxProducerIdExpirationMs: Int,
+ leaderEpochCache: Option[LeaderEpochFileCache],
+ producerStateManager: ProducerStateManager) {
+ val logIdentifier: String = s"[LogLoader partition=$topicPartition, dir=${dir.getParent}]"
+}
+
+/**
+ * This object is responsible for all activities related with recovery of log segments from disk.
+ */
+object LogLoader extends Logging {
+ /**
+ * Load the log segments from the log files on disk, and return the components of the loaded log.
+ * Additionally, it also suitably updates the provided LeaderEpochFileCache and ProducerStateManager
+ * to reflect the contents of the loaded log.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is only
+ * called before all logs are loaded.
Review comment:
Thise comment was trying to convey that in the context of the calling thread, this function doesn't convert the `IOException` to `KafkaStorageException`. In the recovery path, we always asynchronously delete segments. The `IOException` conversion happens only within the background work that's scheduled for segment deletions. https://github.com/apache/kafka/blob/193a9dfc44fa2c0df163a8518a804c4579d37e36/core/src/main/scala/kafka/log/Log.scala#L2374-L2389
Since this is a bit misleading, I've improved the comment now to say:
```
In the context of the calling thread, this function does not need to convert
IOException to KafkaStorageException because it is only called before all logs are loaded.
```
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r611454958
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -255,19 +261,21 @@ case object SnapshotGenerated extends LogStartOffsetIncrementReason {
@threadsafe
class Log(@volatile private var _dir: File,
@volatile var config: LogConfig,
+ val segments: LogSegments,
Review comment:
I see your point. It's good thinking. But the main reason I didn't take this route was that some of the fields of `LogComponents` are immutable within the context of the `LogLoader`, but are mutable by `Log`. Examples: `logStartOffset`, `recoveryPoint` and `leaderEpochCache ` are defined as `var` only in `Log`. It seemed hard to explain if we made it a `var` inside `LogComponents` just because the `Log` class wants to mutate them.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r607168537
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,423 @@
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+class LogLoader(dir: File,
Review comment:
Documentation
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -2763,6 +2191,320 @@ object Log {
appendInfo.append(batch, firstOffsetMetadata)
}
+ def maybeCreateLeaderEpochCache(dir: File,
+ topicPartition: TopicPartition,
+ logDirFailureChannel: LogDirFailureChannel,
+ recordVersion: RecordVersion): Option[LeaderEpochFileCache] = {
+ val leaderEpochFile = LeaderEpochCheckpointFile.newFile(dir)
+
+ def newLeaderEpochFileCache(): LeaderEpochFileCache = {
+ val checkpointFile = new LeaderEpochCheckpointFile(leaderEpochFile, logDirFailureChannel)
+ new LeaderEpochFileCache(topicPartition, checkpointFile)
+ }
+
+ if (recordVersion.precedes(RecordVersion.V2)) {
+ val currentCache = if (leaderEpochFile.exists())
+ Some(newLeaderEpochFileCache())
+ else
+ None
+
+ if (currentCache.exists(_.nonEmpty))
+ warn(s"Deleting non-empty leader epoch cache due to incompatible message format $recordVersion")
+
+ Files.deleteIfExists(leaderEpochFile.toPath)
+ None
+ } else {
+ Some(newLeaderEpochFileCache())
+ }
+ }
+
+ /**
+ * Swap one or more new segment in place and delete one or more existing segments in a crash-safe manner. The old
+ * segments will be asynchronously deleted.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is either called before all logs are loaded
+ * or the caller will catch and handle IOException
+ *
+ * The sequence of operations is:
+ * <ol>
+ * <li> Cleaner creates one or more new segments with suffix .cleaned and invokes replaceSegments().
+ * If broker crashes at this point, the clean-and-swap operation is aborted and
+ * the .cleaned files are deleted on recovery in loadSegments().
+ * <li> New segments are renamed .swap. If the broker crashes before all segments were renamed to .swap, the
+ * clean-and-swap operation is aborted - .cleaned as well as .swap files are deleted on recovery in
+ * loadSegments(). We detect this situation by maintaining a specific order in which files are renamed from
+ * .cleaned to .swap. Basically, files are renamed in descending order of offsets. On recovery, all .swap files
+ * whose offset is greater than the minimum-offset .clean file are deleted.
+ * <li> If the broker crashes after all new segments were renamed to .swap, the operation is completed, the swap
+ * operation is resumed on recovery as described in the next step.
+ * <li> Old segment files are renamed to .deleted and asynchronous delete is scheduled.
+ * If the broker crashes, any .deleted files left behind are deleted on recovery in loadSegments().
+ * replaceSegments() is then invoked to complete the swap with newSegment recreated from
+ * the .swap file and oldSegments containing segments which were not renamed before the crash.
+ * <li> Swap segment(s) are renamed to replace the existing segments, completing this operation.
+ * If the broker crashes, any .deleted files which may be left behind are deleted
+ * on recovery in loadSegments().
+ * </ol>
+ *
+ * @param existingSegments The existing segments of the log
+ * @param newSegments The new log segment to add to the log
+ * @param oldSegments The old log segments to delete from the log
+ * @param isRecoveredSwapFile true if the new segment was created from a swap file during recovery after a crash
+ */
+ private[log] def replaceSegments(existingSegments: LogSegments,
+ newSegments: Seq[LogSegment],
+ oldSegments: Seq[LogSegment],
+ isRecoveredSwapFile: Boolean = false,
+ dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ logDirFailureChannel: LogDirFailureChannel,
+ producerStateManager: ProducerStateManager): Unit = {
+ val sortedNewSegments = newSegments.sortBy(_.baseOffset)
+ // Some old segments may have been removed from index and scheduled for async deletion after the caller reads segments
+ // but before this method is executed. We want to filter out those segments to avoid calling asyncDeleteSegment()
+ // multiple times for the same segment.
+ val sortedOldSegments = oldSegments.filter(seg => existingSegments.contains(seg.baseOffset)).sortBy(_.baseOffset)
+
+ // need to do this in two phases to be crash safe AND do the delete asynchronously
+ // if we crash in the middle of this we complete the swap in loadSegments()
+ if (!isRecoveredSwapFile)
+ sortedNewSegments.reverse.foreach(_.changeFileSuffixes(Log.CleanedFileSuffix, Log.SwapFileSuffix))
+ sortedNewSegments.reverse.foreach(existingSegments.add(_))
+ val newSegmentBaseOffsets = sortedNewSegments.map(_.baseOffset).toSet
+
+ // delete the old files
+ sortedOldSegments.foreach { seg =>
+ // remove the index entry
+ if (seg.baseOffset != sortedNewSegments.head.baseOffset)
+ existingSegments.remove(seg.baseOffset)
+ // delete segment files, but do not delete producer state for segment objects which are being replaced.
+ deleteSegmentFiles(
+ List(seg),
+ asyncDelete = true,
+ deleteProducerStateSnapshots = !newSegmentBaseOffsets.contains(seg.baseOffset),
+ dir,
+ topicPartition,
+ config,
+ scheduler,
+ logDirFailureChannel,
+ producerStateManager)
+ }
+ // okay we are safe now, remove the swap suffix
+ sortedNewSegments.foreach(_.changeFileSuffixes(Log.SwapFileSuffix, ""))
+ }
+
+ /**
+ * Perform physical deletion for the given file. Allows the file to be deleted asynchronously or synchronously.
+ *
+ * This method assumes that the file exists and the method is not thread-safe.
+ *
+ * This method does not need to convert IOException (thrown from changeFileSuffixes) to KafkaStorageException because
+ * it is either called before all logs are loaded or the caller will catch and handle IOException
+ *
+ * @throws IOException if the file can't be renamed and still exists
+ */
+ private[log] def deleteSegmentFiles(segments: Iterable[LogSegment],
+ asyncDelete: Boolean,
+ deleteProducerStateSnapshots: Boolean = true,
+ dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ logDirFailureChannel: LogDirFailureChannel,
+ producerStateManager: ProducerStateManager): Unit = {
+ segments.foreach(_.changeFileSuffixes("", Log.DeletedFileSuffix))
+
+ def deleteSegments(): Unit = {
+ info(s"Deleting segment files ${segments.mkString(",")}")
+ val parentDir = dir.getParent
+ maybeHandleIOException(logDirFailureChannel, parentDir, s"Error while deleting segments for $topicPartition in dir $parentDir") {
+ segments.foreach { segment =>
+ segment.deleteIfExists()
+ if (deleteProducerStateSnapshots)
+ producerStateManager.removeAndDeleteSnapshot(segment.baseOffset)
+ }
+ }
+ }
+
+ if (asyncDelete)
+ scheduler.schedule("delete-file", () => deleteSegments(), delay = config.fileDeleteDelayMs)
+ else
+ deleteSegments()
+ }
+
+ private def maybeHandleIOException[T](logDirFailureChannel: LogDirFailureChannel,
+ parentDir: String,
+ msg: => String)(fun: => T): T = {
+ if (logDirFailureChannel.hasOfflineLogDir(parentDir)) {
+ throw new KafkaStorageException(s"The log dir $parentDir is offline due to a previous IO exception.")
+ }
+ try {
+ fun
+ } catch {
+ case e: IOException =>
+ logDirFailureChannel.maybeAddOfflineLogDir(parentDir, msg, e)
+ throw new KafkaStorageException(msg, e)
+ }
+ }
+
+ // Rebuild producer state until lastOffset. This method may be called from the recovery code path, and thus must be
+ // free of all side-effects, i.e. it must not update any log-specific state.
+ private[log] def rebuildProducerState(segments: LogSegments,
Review comment:
Document the params
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -2763,6 +2191,320 @@ object Log {
appendInfo.append(batch, firstOffsetMetadata)
}
+ def maybeCreateLeaderEpochCache(dir: File,
+ topicPartition: TopicPartition,
+ logDirFailureChannel: LogDirFailureChannel,
+ recordVersion: RecordVersion): Option[LeaderEpochFileCache] = {
+ val leaderEpochFile = LeaderEpochCheckpointFile.newFile(dir)
+
+ def newLeaderEpochFileCache(): LeaderEpochFileCache = {
+ val checkpointFile = new LeaderEpochCheckpointFile(leaderEpochFile, logDirFailureChannel)
+ new LeaderEpochFileCache(topicPartition, checkpointFile)
+ }
+
+ if (recordVersion.precedes(RecordVersion.V2)) {
+ val currentCache = if (leaderEpochFile.exists())
+ Some(newLeaderEpochFileCache())
+ else
+ None
+
+ if (currentCache.exists(_.nonEmpty))
+ warn(s"Deleting non-empty leader epoch cache due to incompatible message format $recordVersion")
+
+ Files.deleteIfExists(leaderEpochFile.toPath)
+ None
+ } else {
+ Some(newLeaderEpochFileCache())
+ }
+ }
+
+ /**
+ * Swap one or more new segment in place and delete one or more existing segments in a crash-safe manner. The old
+ * segments will be asynchronously deleted.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is either called before all logs are loaded
+ * or the caller will catch and handle IOException
+ *
+ * The sequence of operations is:
+ * <ol>
+ * <li> Cleaner creates one or more new segments with suffix .cleaned and invokes replaceSegments().
+ * If broker crashes at this point, the clean-and-swap operation is aborted and
+ * the .cleaned files are deleted on recovery in loadSegments().
+ * <li> New segments are renamed .swap. If the broker crashes before all segments were renamed to .swap, the
+ * clean-and-swap operation is aborted - .cleaned as well as .swap files are deleted on recovery in
+ * loadSegments(). We detect this situation by maintaining a specific order in which files are renamed from
+ * .cleaned to .swap. Basically, files are renamed in descending order of offsets. On recovery, all .swap files
+ * whose offset is greater than the minimum-offset .clean file are deleted.
+ * <li> If the broker crashes after all new segments were renamed to .swap, the operation is completed, the swap
+ * operation is resumed on recovery as described in the next step.
+ * <li> Old segment files are renamed to .deleted and asynchronous delete is scheduled.
+ * If the broker crashes, any .deleted files left behind are deleted on recovery in loadSegments().
+ * replaceSegments() is then invoked to complete the swap with newSegment recreated from
+ * the .swap file and oldSegments containing segments which were not renamed before the crash.
+ * <li> Swap segment(s) are renamed to replace the existing segments, completing this operation.
+ * If the broker crashes, any .deleted files which may be left behind are deleted
+ * on recovery in loadSegments().
+ * </ol>
+ *
+ * @param existingSegments The existing segments of the log
+ * @param newSegments The new log segment to add to the log
+ * @param oldSegments The old log segments to delete from the log
+ * @param isRecoveredSwapFile true if the new segment was created from a swap file during recovery after a crash
+ */
+ private[log] def replaceSegments(existingSegments: LogSegments,
+ newSegments: Seq[LogSegment],
+ oldSegments: Seq[LogSegment],
+ isRecoveredSwapFile: Boolean = false,
+ dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ logDirFailureChannel: LogDirFailureChannel,
+ producerStateManager: ProducerStateManager): Unit = {
+ val sortedNewSegments = newSegments.sortBy(_.baseOffset)
+ // Some old segments may have been removed from index and scheduled for async deletion after the caller reads segments
+ // but before this method is executed. We want to filter out those segments to avoid calling asyncDeleteSegment()
+ // multiple times for the same segment.
+ val sortedOldSegments = oldSegments.filter(seg => existingSegments.contains(seg.baseOffset)).sortBy(_.baseOffset)
+
+ // need to do this in two phases to be crash safe AND do the delete asynchronously
+ // if we crash in the middle of this we complete the swap in loadSegments()
+ if (!isRecoveredSwapFile)
+ sortedNewSegments.reverse.foreach(_.changeFileSuffixes(Log.CleanedFileSuffix, Log.SwapFileSuffix))
+ sortedNewSegments.reverse.foreach(existingSegments.add(_))
+ val newSegmentBaseOffsets = sortedNewSegments.map(_.baseOffset).toSet
+
+ // delete the old files
+ sortedOldSegments.foreach { seg =>
+ // remove the index entry
+ if (seg.baseOffset != sortedNewSegments.head.baseOffset)
+ existingSegments.remove(seg.baseOffset)
+ // delete segment files, but do not delete producer state for segment objects which are being replaced.
+ deleteSegmentFiles(
+ List(seg),
+ asyncDelete = true,
+ deleteProducerStateSnapshots = !newSegmentBaseOffsets.contains(seg.baseOffset),
+ dir,
+ topicPartition,
+ config,
+ scheduler,
+ logDirFailureChannel,
+ producerStateManager)
+ }
+ // okay we are safe now, remove the swap suffix
+ sortedNewSegments.foreach(_.changeFileSuffixes(Log.SwapFileSuffix, ""))
+ }
+
+ /**
+ * Perform physical deletion for the given file. Allows the file to be deleted asynchronously or synchronously.
+ *
+ * This method assumes that the file exists and the method is not thread-safe.
+ *
+ * This method does not need to convert IOException (thrown from changeFileSuffixes) to KafkaStorageException because
+ * it is either called before all logs are loaded or the caller will catch and handle IOException
+ *
+ * @throws IOException if the file can't be renamed and still exists
+ */
+ private[log] def deleteSegmentFiles(segments: Iterable[LogSegment],
+ asyncDelete: Boolean,
+ deleteProducerStateSnapshots: Boolean = true,
+ dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ logDirFailureChannel: LogDirFailureChannel,
+ producerStateManager: ProducerStateManager): Unit = {
+ segments.foreach(_.changeFileSuffixes("", Log.DeletedFileSuffix))
+
+ def deleteSegments(): Unit = {
+ info(s"Deleting segment files ${segments.mkString(",")}")
+ val parentDir = dir.getParent
+ maybeHandleIOException(logDirFailureChannel, parentDir, s"Error while deleting segments for $topicPartition in dir $parentDir") {
+ segments.foreach { segment =>
+ segment.deleteIfExists()
+ if (deleteProducerStateSnapshots)
+ producerStateManager.removeAndDeleteSnapshot(segment.baseOffset)
+ }
+ }
+ }
+
+ if (asyncDelete)
+ scheduler.schedule("delete-file", () => deleteSegments(), delay = config.fileDeleteDelayMs)
+ else
+ deleteSegments()
+ }
+
+ private def maybeHandleIOException[T](logDirFailureChannel: LogDirFailureChannel,
+ parentDir: String,
+ msg: => String)(fun: => T): T = {
+ if (logDirFailureChannel.hasOfflineLogDir(parentDir)) {
+ throw new KafkaStorageException(s"The log dir $parentDir is offline due to a previous IO exception.")
+ }
+ try {
+ fun
+ } catch {
+ case e: IOException =>
+ logDirFailureChannel.maybeAddOfflineLogDir(parentDir, msg, e)
+ throw new KafkaStorageException(msg, e)
+ }
+ }
+
+ // Rebuild producer state until lastOffset. This method may be called from the recovery code path, and thus must be
+ // free of all side-effects, i.e. it must not update any log-specific state.
+ private[log] def rebuildProducerState(segments: LogSegments,
+ logStartOffset: Long,
+ lastOffset: Long,
+ recordVersion: RecordVersion,
+ time: Time,
+ reloadFromCleanShutdown: Boolean,
+ producerStateManager: ProducerStateManager): Unit = {
+ val allSegments = segments.values
+ val offsetsToSnapshot =
+ if (allSegments.nonEmpty) {
+ val nextLatestSegmentBaseOffset = segments.lowerSegment(allSegments.last.baseOffset).map(_.baseOffset)
+ Seq(nextLatestSegmentBaseOffset, Some(allSegments.last.baseOffset), Some(lastOffset))
+ } else {
+ Seq(Some(lastOffset))
+ }
+ info(s"Loading producer state till offset $lastOffset with message format version ${recordVersion.value}")
+
+ // We want to avoid unnecessary scanning of the log to build the producer state when the broker is being
+ // upgraded. The basic idea is to use the absence of producer snapshot files to detect the upgrade case,
+ // but we have to be careful not to assume too much in the presence of broker failures. The two most common
+ // upgrade cases in which we expect to find no snapshots are the following:
+ //
+ // 1. The broker has been upgraded, but the topic is still on the old message format.
+ // 2. The broker has been upgraded, the topic is on the new message format, and we had a clean shutdown.
+ //
+ // If we hit either of these cases, we skip producer state loading and write a new snapshot at the log end
+ // offset (see below). The next time the log is reloaded, we will load producer state using this snapshot
+ // (or later snapshots). Otherwise, if there is no snapshot file, then we have to rebuild producer state
+ // from the first segment.
+ if (recordVersion.value < RecordBatch.MAGIC_VALUE_V2 ||
+ (producerStateManager.latestSnapshotOffset.isEmpty && reloadFromCleanShutdown)) {
+ // To avoid an expensive scan through all of the segments, we take empty snapshots from the start of the
+ // last two segments and the last offset. This should avoid the full scan in the case that the log needs
+ // truncation.
+ offsetsToSnapshot.flatten.foreach { offset =>
+ producerStateManager.updateMapEndOffset(offset)
+ producerStateManager.takeSnapshot()
+ }
+ } else {
+ info(s"Reloading from producer snapshot and rebuilding producer state from offset $lastOffset")
+ val isEmptyBeforeTruncation = producerStateManager.isEmpty && producerStateManager.mapEndOffset >= lastOffset
+ val producerStateLoadStart = time.milliseconds()
+ producerStateManager.truncateAndReload(logStartOffset, lastOffset, time.milliseconds())
+ val segmentRecoveryStart = time.milliseconds()
+
+ // Only do the potentially expensive reloading if the last snapshot offset is lower than the log end
+ // offset (which would be the case on first startup) and there were active producers prior to truncation
+ // (which could be the case if truncating after initial loading). If there weren't, then truncating
+ // shouldn't change that fact (although it could cause a producerId to expire earlier than expected),
+ // and we can skip the loading. This is an optimization for users which are not yet using
+ // idempotent/transactional features yet.
+ if (lastOffset > producerStateManager.mapEndOffset && !isEmptyBeforeTruncation) {
+ val segmentOfLastOffset = segments.floorSegment(lastOffset)
+
+ segments.values(producerStateManager.mapEndOffset, lastOffset).foreach { segment =>
+ val startOffset = Utils.max(segment.baseOffset, producerStateManager.mapEndOffset, logStartOffset)
+ producerStateManager.updateMapEndOffset(startOffset)
+
+ if (offsetsToSnapshot.contains(Some(segment.baseOffset)))
+ producerStateManager.takeSnapshot()
+
+ val maxPosition = if (segmentOfLastOffset.contains(segment)) {
+ Option(segment.translateOffset(lastOffset))
+ .map(_.position)
+ .getOrElse(segment.size)
+ } else {
+ segment.size
+ }
+
+ val fetchDataInfo = segment.read(startOffset,
+ maxSize = Int.MaxValue,
+ maxPosition = maxPosition,
+ minOneMessage = false)
+ if (fetchDataInfo != null)
+ loadProducersFromRecords(producerStateManager, fetchDataInfo.records)
+ }
+ }
+ producerStateManager.updateMapEndOffset(lastOffset)
+ producerStateManager.takeSnapshot()
+ info(s"Producer state recovery took ${producerStateLoadStart - segmentRecoveryStart}ms for snapshot load " +
+ s"and ${time.milliseconds() - segmentRecoveryStart}ms for segment recovery from offset $lastOffset")
+ }
+ }
+
+ /**
+ * Split a segment into one or more segments such that there is no offset overflow in any of them. The
+ * resulting segments will contain the exact same messages that are present in the input segment. On successful
+ * completion of this method, the input segment will be deleted and will be replaced by the resulting new segments.
+ * See replaceSegments for recovery logic, in case the broker dies in the middle of this operation.
+ * <p>Note that this method assumes we have already determined that the segment passed in contains records that cause
+ * offset overflow.</p>
+ * <p>The split logic overloads the use of .clean files that LogCleaner typically uses to make the process of replacing
+ * the input segment with multiple new segments atomic and recoverable in the event of a crash. See replaceSegments
+ * and completeSwapOperations for the implementation to make this operation recoverable on crashes.</p>
+ * @param segment Segment to split
+ * @return List of new segments that replace the input segment
+ */
+ private[log] def splitOverflowedSegment(segment: LogSegment,
Review comment:
Document the params
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -2763,6 +2191,320 @@ object Log {
appendInfo.append(batch, firstOffsetMetadata)
}
+ def maybeCreateLeaderEpochCache(dir: File,
+ topicPartition: TopicPartition,
+ logDirFailureChannel: LogDirFailureChannel,
+ recordVersion: RecordVersion): Option[LeaderEpochFileCache] = {
+ val leaderEpochFile = LeaderEpochCheckpointFile.newFile(dir)
+
+ def newLeaderEpochFileCache(): LeaderEpochFileCache = {
+ val checkpointFile = new LeaderEpochCheckpointFile(leaderEpochFile, logDirFailureChannel)
+ new LeaderEpochFileCache(topicPartition, checkpointFile)
+ }
+
+ if (recordVersion.precedes(RecordVersion.V2)) {
+ val currentCache = if (leaderEpochFile.exists())
+ Some(newLeaderEpochFileCache())
+ else
+ None
+
+ if (currentCache.exists(_.nonEmpty))
+ warn(s"Deleting non-empty leader epoch cache due to incompatible message format $recordVersion")
+
+ Files.deleteIfExists(leaderEpochFile.toPath)
+ None
+ } else {
+ Some(newLeaderEpochFileCache())
+ }
+ }
+
+ /**
+ * Swap one or more new segment in place and delete one or more existing segments in a crash-safe manner. The old
+ * segments will be asynchronously deleted.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is either called before all logs are loaded
+ * or the caller will catch and handle IOException
+ *
+ * The sequence of operations is:
+ * <ol>
+ * <li> Cleaner creates one or more new segments with suffix .cleaned and invokes replaceSegments().
+ * If broker crashes at this point, the clean-and-swap operation is aborted and
+ * the .cleaned files are deleted on recovery in loadSegments().
+ * <li> New segments are renamed .swap. If the broker crashes before all segments were renamed to .swap, the
+ * clean-and-swap operation is aborted - .cleaned as well as .swap files are deleted on recovery in
+ * loadSegments(). We detect this situation by maintaining a specific order in which files are renamed from
+ * .cleaned to .swap. Basically, files are renamed in descending order of offsets. On recovery, all .swap files
+ * whose offset is greater than the minimum-offset .clean file are deleted.
+ * <li> If the broker crashes after all new segments were renamed to .swap, the operation is completed, the swap
+ * operation is resumed on recovery as described in the next step.
+ * <li> Old segment files are renamed to .deleted and asynchronous delete is scheduled.
+ * If the broker crashes, any .deleted files left behind are deleted on recovery in loadSegments().
+ * replaceSegments() is then invoked to complete the swap with newSegment recreated from
+ * the .swap file and oldSegments containing segments which were not renamed before the crash.
+ * <li> Swap segment(s) are renamed to replace the existing segments, completing this operation.
+ * If the broker crashes, any .deleted files which may be left behind are deleted
+ * on recovery in loadSegments().
+ * </ol>
+ *
+ * @param existingSegments The existing segments of the log
+ * @param newSegments The new log segment to add to the log
+ * @param oldSegments The old log segments to delete from the log
+ * @param isRecoveredSwapFile true if the new segment was created from a swap file during recovery after a crash
+ */
+ private[log] def replaceSegments(existingSegments: LogSegments,
Review comment:
Document the params
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -2566,10 +1979,18 @@ object Log {
logDirFailureChannel: LogDirFailureChannel,
lastShutdownClean: Boolean = true,
keepPartitionMetadataFile: Boolean = true): Log = {
+ // create the log directory if it doesn't exist
+ Files.createDirectories(dir.toPath)
val topicPartition = Log.parseTopicPartitionName(dir)
+ val leaderEpochCache = Log.maybeCreateLeaderEpochCache(dir, topicPartition, logDirFailureChannel,
+ config.messageFormatVersion.recordVersion)
val producerStateManager = new ProducerStateManager(topicPartition, dir, maxProducerIdExpirationMs)
- new Log(dir, config, logStartOffset, recoveryPoint, scheduler, brokerTopicStats, time, maxProducerIdExpirationMs,
- producerIdExpirationCheckIntervalMs, topicPartition, producerStateManager, logDirFailureChannel, lastShutdownClean, keepPartitionMetadataFile)
+ val logLoader = new LogLoader(dir, config, logStartOffset, recoveryPoint, scheduler, time, maxProducerIdExpirationMs,
Review comment:
Should the LogLoader also instantiate leaderEpochCache and producerStateManager?
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -2763,6 +2191,320 @@ object Log {
appendInfo.append(batch, firstOffsetMetadata)
}
+ def maybeCreateLeaderEpochCache(dir: File,
+ topicPartition: TopicPartition,
+ logDirFailureChannel: LogDirFailureChannel,
+ recordVersion: RecordVersion): Option[LeaderEpochFileCache] = {
+ val leaderEpochFile = LeaderEpochCheckpointFile.newFile(dir)
+
+ def newLeaderEpochFileCache(): LeaderEpochFileCache = {
+ val checkpointFile = new LeaderEpochCheckpointFile(leaderEpochFile, logDirFailureChannel)
+ new LeaderEpochFileCache(topicPartition, checkpointFile)
+ }
+
+ if (recordVersion.precedes(RecordVersion.V2)) {
+ val currentCache = if (leaderEpochFile.exists())
+ Some(newLeaderEpochFileCache())
+ else
+ None
+
+ if (currentCache.exists(_.nonEmpty))
+ warn(s"Deleting non-empty leader epoch cache due to incompatible message format $recordVersion")
+
+ Files.deleteIfExists(leaderEpochFile.toPath)
+ None
+ } else {
+ Some(newLeaderEpochFileCache())
+ }
+ }
+
+ /**
+ * Swap one or more new segment in place and delete one or more existing segments in a crash-safe manner. The old
+ * segments will be asynchronously deleted.
+ *
+ * This method does not need to convert IOException to KafkaStorageException because it is either called before all logs are loaded
+ * or the caller will catch and handle IOException
+ *
+ * The sequence of operations is:
+ * <ol>
+ * <li> Cleaner creates one or more new segments with suffix .cleaned and invokes replaceSegments().
+ * If broker crashes at this point, the clean-and-swap operation is aborted and
+ * the .cleaned files are deleted on recovery in loadSegments().
+ * <li> New segments are renamed .swap. If the broker crashes before all segments were renamed to .swap, the
+ * clean-and-swap operation is aborted - .cleaned as well as .swap files are deleted on recovery in
+ * loadSegments(). We detect this situation by maintaining a specific order in which files are renamed from
+ * .cleaned to .swap. Basically, files are renamed in descending order of offsets. On recovery, all .swap files
+ * whose offset is greater than the minimum-offset .clean file are deleted.
+ * <li> If the broker crashes after all new segments were renamed to .swap, the operation is completed, the swap
+ * operation is resumed on recovery as described in the next step.
+ * <li> Old segment files are renamed to .deleted and asynchronous delete is scheduled.
+ * If the broker crashes, any .deleted files left behind are deleted on recovery in loadSegments().
+ * replaceSegments() is then invoked to complete the swap with newSegment recreated from
+ * the .swap file and oldSegments containing segments which were not renamed before the crash.
+ * <li> Swap segment(s) are renamed to replace the existing segments, completing this operation.
+ * If the broker crashes, any .deleted files which may be left behind are deleted
+ * on recovery in loadSegments().
+ * </ol>
+ *
+ * @param existingSegments The existing segments of the log
+ * @param newSegments The new log segment to add to the log
+ * @param oldSegments The old log segments to delete from the log
+ * @param isRecoveredSwapFile true if the new segment was created from a swap file during recovery after a crash
+ */
+ private[log] def replaceSegments(existingSegments: LogSegments,
+ newSegments: Seq[LogSegment],
+ oldSegments: Seq[LogSegment],
+ isRecoveredSwapFile: Boolean = false,
+ dir: File,
+ topicPartition: TopicPartition,
+ config: LogConfig,
+ scheduler: Scheduler,
+ logDirFailureChannel: LogDirFailureChannel,
+ producerStateManager: ProducerStateManager): Unit = {
+ val sortedNewSegments = newSegments.sortBy(_.baseOffset)
+ // Some old segments may have been removed from index and scheduled for async deletion after the caller reads segments
+ // but before this method is executed. We want to filter out those segments to avoid calling asyncDeleteSegment()
+ // multiple times for the same segment.
+ val sortedOldSegments = oldSegments.filter(seg => existingSegments.contains(seg.baseOffset)).sortBy(_.baseOffset)
+
+ // need to do this in two phases to be crash safe AND do the delete asynchronously
+ // if we crash in the middle of this we complete the swap in loadSegments()
+ if (!isRecoveredSwapFile)
+ sortedNewSegments.reverse.foreach(_.changeFileSuffixes(Log.CleanedFileSuffix, Log.SwapFileSuffix))
+ sortedNewSegments.reverse.foreach(existingSegments.add(_))
+ val newSegmentBaseOffsets = sortedNewSegments.map(_.baseOffset).toSet
+
+ // delete the old files
+ sortedOldSegments.foreach { seg =>
+ // remove the index entry
+ if (seg.baseOffset != sortedNewSegments.head.baseOffset)
+ existingSegments.remove(seg.baseOffset)
+ // delete segment files, but do not delete producer state for segment objects which are being replaced.
+ deleteSegmentFiles(
+ List(seg),
+ asyncDelete = true,
+ deleteProducerStateSnapshots = !newSegmentBaseOffsets.contains(seg.baseOffset),
+ dir,
+ topicPartition,
+ config,
+ scheduler,
+ logDirFailureChannel,
+ producerStateManager)
+ }
+ // okay we are safe now, remove the swap suffix
+ sortedNewSegments.foreach(_.changeFileSuffixes(Log.SwapFileSuffix, ""))
+ }
+
+ /**
+ * Perform physical deletion for the given file. Allows the file to be deleted asynchronously or synchronously.
+ *
+ * This method assumes that the file exists and the method is not thread-safe.
+ *
+ * This method does not need to convert IOException (thrown from changeFileSuffixes) to KafkaStorageException because
+ * it is either called before all logs are loaded or the caller will catch and handle IOException
+ *
+ * @throws IOException if the file can't be renamed and still exists
+ */
+ private[log] def deleteSegmentFiles(segments: Iterable[LogSegment],
Review comment:
Document the params
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r611454958
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -255,19 +261,21 @@ case object SnapshotGenerated extends LogStartOffsetIncrementReason {
@threadsafe
class Log(@volatile private var _dir: File,
@volatile var config: LogConfig,
+ val segments: LogSegments,
Review comment:
I see your point. It's good thinking. But the main reason I didn't take this route was that some of the fields of `LogComponents` are immutable within the context of the `LogLoader`, but are mutable by `Log`. Examples: `logStartOffset`, `recoveryPoint` and `leaderEpochCache ` are defined as `var` only in `Log`. It seemed hard to explain if we made it a `var` inside `LogComponents` just because the `Log` class wants to mutate them in the future.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r611433367
##########
File path: core/src/test/scala/unit/kafka/log/LogTestUtils.scala
##########
@@ -37,4 +47,246 @@ object LogTestUtils {
new LogSegment(ms, idx, timeIdx, txnIndex, offset, indexIntervalBytes, 0, time)
}
+
+ def createLogConfig(segmentMs: Long = Defaults.SegmentMs,
+ segmentBytes: Int = Defaults.SegmentSize,
+ retentionMs: Long = Defaults.RetentionMs,
+ retentionBytes: Long = Defaults.RetentionSize,
+ segmentJitterMs: Long = Defaults.SegmentJitterMs,
+ cleanupPolicy: String = Defaults.CleanupPolicy,
+ maxMessageBytes: Int = Defaults.MaxMessageSize,
+ indexIntervalBytes: Int = Defaults.IndexInterval,
+ segmentIndexBytes: Int = Defaults.MaxIndexSize,
+ messageFormatVersion: String = Defaults.MessageFormatVersion,
+ fileDeleteDelayMs: Long = Defaults.FileDeleteDelayMs): LogConfig = {
+ val logProps = new Properties()
+
+ logProps.put(LogConfig.SegmentMsProp, segmentMs: java.lang.Long)
+ logProps.put(LogConfig.SegmentBytesProp, segmentBytes: Integer)
+ logProps.put(LogConfig.RetentionMsProp, retentionMs: java.lang.Long)
+ logProps.put(LogConfig.RetentionBytesProp, retentionBytes: java.lang.Long)
+ logProps.put(LogConfig.SegmentJitterMsProp, segmentJitterMs: java.lang.Long)
+ logProps.put(LogConfig.CleanupPolicyProp, cleanupPolicy)
+ logProps.put(LogConfig.MaxMessageBytesProp, maxMessageBytes: Integer)
+ logProps.put(LogConfig.IndexIntervalBytesProp, indexIntervalBytes: Integer)
+ logProps.put(LogConfig.SegmentIndexBytesProp, segmentIndexBytes: Integer)
+ logProps.put(LogConfig.MessageFormatVersionProp, messageFormatVersion)
+ logProps.put(LogConfig.FileDeleteDelayMsProp, fileDeleteDelayMs: java.lang.Long)
+ LogConfig(logProps)
+ }
+
+ def createLog(dir: File,
+ config: LogConfig,
+ brokerTopicStats: BrokerTopicStats,
+ scheduler: Scheduler,
+ time: Time,
+ logStartOffset: Long = 0L,
+ recoveryPoint: Long = 0L,
+ maxProducerIdExpirationMs: Int = 60 * 60 * 1000,
+ producerIdExpirationCheckIntervalMs: Int = LogManager.ProducerIdExpirationCheckIntervalMs,
+ lastShutdownClean: Boolean = true,
+ topicId: Option[Uuid] = None): Log = {
+ Log(dir = dir,
+ config = config,
+ logStartOffset = logStartOffset,
+ recoveryPoint = recoveryPoint,
+ scheduler = scheduler,
+ brokerTopicStats = brokerTopicStats,
+ time = time,
+ maxProducerIdExpirationMs = maxProducerIdExpirationMs,
+ producerIdExpirationCheckIntervalMs = producerIdExpirationCheckIntervalMs,
+ logDirFailureChannel = new LogDirFailureChannel(10),
+ lastShutdownClean = lastShutdownClean,
+ topicId = topicId)
+ }
+
+ /**
+ * Check if the given log contains any segment with records that cause offset overflow.
+ * @param log Log to check
+ * @return true if log contains at least one segment with offset overflow; false otherwise
+ */
+ def hasOffsetOverflow(log: Log): Boolean = firstOverflowSegment(log).isDefined
+
+ def firstOverflowSegment(log: Log): Option[LogSegment] = {
+ def hasOverflow(baseOffset: Long, batch: RecordBatch): Boolean =
+ batch.lastOffset > baseOffset + Int.MaxValue || batch.baseOffset < baseOffset
+
+ for (segment <- log.logSegments) {
+ val overflowBatch = segment.log.batches.asScala.find(batch => hasOverflow(segment.baseOffset, batch))
+ if (overflowBatch.isDefined)
+ return Some(segment)
+ }
+ None
+ }
+
+ private def rawSegment(logDir: File, baseOffset: Long): FileRecords =
+ FileRecords.open(Log.logFile(logDir, baseOffset))
+
+ /**
+ * Initialize the given log directory with a set of segments, one of which will have an
+ * offset which overflows the segment
+ */
+ def initializeLogDirWithOverflowedSegment(logDir: File): Unit = {
+ def writeSampleBatches(baseOffset: Long, segment: FileRecords): Long = {
+ def record(offset: Long) = {
+ val data = offset.toString.getBytes
+ new SimpleRecord(data, data)
+ }
+
+ segment.append(MemoryRecords.withRecords(baseOffset, CompressionType.NONE, 0,
+ record(baseOffset)))
+ segment.append(MemoryRecords.withRecords(baseOffset + 1, CompressionType.NONE, 0,
+ record(baseOffset + 1),
+ record(baseOffset + 2)))
+ segment.append(MemoryRecords.withRecords(baseOffset + Int.MaxValue - 1, CompressionType.NONE, 0,
+ record(baseOffset + Int.MaxValue - 1)))
+ // Need to create the offset files explicitly to avoid triggering segment recovery to truncate segment.
+ Log.offsetIndexFile(logDir, baseOffset).createNewFile()
+ Log.timeIndexFile(logDir, baseOffset).createNewFile()
+ baseOffset + Int.MaxValue
+ }
+
+ def writeNormalSegment(baseOffset: Long): Long = {
+ val segment = rawSegment(logDir, baseOffset)
+ try writeSampleBatches(baseOffset, segment)
+ finally segment.close()
+ }
+
+ def writeOverflowSegment(baseOffset: Long): Long = {
+ val segment = rawSegment(logDir, baseOffset)
+ try {
+ val nextOffset = writeSampleBatches(baseOffset, segment)
+ writeSampleBatches(nextOffset, segment)
+ } finally segment.close()
+ }
+
+ // We create three segments, the second of which contains offsets which overflow
+ var nextOffset = 0L
+ nextOffset = writeNormalSegment(nextOffset)
+ nextOffset = writeOverflowSegment(nextOffset)
+ writeNormalSegment(nextOffset)
+ }
+
+ def allRecords(log: Log): List[Record] = {
+ val recordsFound = ListBuffer[Record]()
+ for (logSegment <- log.logSegments) {
+ for (batch <- logSegment.log.batches.asScala) {
+ recordsFound ++= batch.iterator().asScala
+ }
+ }
+ recordsFound.toList
+ }
+
+ def verifyRecordsInLog(log: Log, expectedRecords: List[Record]): Unit = {
Review comment:
Good catch. It was unused and I've eliminated it.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao merged pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
junrao merged pull request #10478:
URL: https://github.com/apache/kafka/pull/10478
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613799479
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -0,0 +1,526 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log
+
+import java.io.{File, IOException}
+import java.nio.file.{Files, NoSuchFileException}
+
+import kafka.common.LogSegmentOffsetOverflowException
+import kafka.log.Log.{CleanedFileSuffix, DeletedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile, offsetFromFileName}
+import kafka.server.{LogDirFailureChannel, LogOffsetMetadata}
+import kafka.server.epoch.LeaderEpochFileCache
+import kafka.utils.{CoreUtils, Logging, Scheduler}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.InvalidOffsetException
+import org.apache.kafka.common.utils.Time
+
+import scala.collection.{Seq, Set, mutable}
+
+case class LoadedLogOffsets(logStartOffset: Long,
+ recoveryPoint: Long,
+ nextOffsetMetadata: LogOffsetMetadata)
+
+/**
+ * @param dir The directory from which log segments need to be loaded
+ * @param topicPartition The topic partition associated with the log being loaded
+ * @param config The configuration settings for the log being loaded
+ * @param scheduler The thread pool scheduler used for background actions
+ * @param time The time instance used for checking the clock
+ * @param logDirFailureChannel The LogDirFailureChannel instance to asynchronously handle log
Review comment:
In the recovery path, the `logDirFailureChannel` is used to asynchronously handle `IOException` within the `Log.deleteSegmentFiles()` calls. This was the existing behavior prior to this refactor as well. Should we change the existing behavior?https://github.com/apache/kafka/blob/193a9dfc44fa2c0df163a8518a804c4579d37e36/core/src/main/scala/kafka/log/Log.scala#L2377
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #10478: KAFKA-12553: Refactor recovery logic to introduce LogLoader
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #10478:
URL: https://github.com/apache/kafka/pull/10478#discussion_r613788936
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -222,25 +222,31 @@ case object SnapshotGenerated extends LogStartOffsetIncrementReason {
*
* @param _dir The directory in which log segments are created.
* @param config The log configuration settings
+ * @param segments The log segments, these may be non-empty when recovered from disk
Review comment:
Done. Fixed it.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org