chia7712 commented on code in PR #15136:
URL: https://github.com/apache/kafka/pull/15136#discussion_r1569178614
##########
core/src/main/scala/kafka/log/LogManager.scala:
##########
@@ -1189,50 +1216,63 @@ class LogManager(logDirs: Seq[File],
val sourceLog = currentLogs.get(topicPartition)
val destLog = futureLogs.get(topicPartition)
- info(s"Attempting to replace current log $sourceLog with $destLog for
$topicPartition")
if (sourceLog == null)
throw new KafkaStorageException(s"The current replica for
$topicPartition is offline")
if (destLog == null)
throw new KafkaStorageException(s"The future replica for
$topicPartition is offline")
- destLog.renameDir(UnifiedLog.logDirName(topicPartition),
shouldReinitialize = true)
- // the metrics tags still contain "future", so we have to remove it.
- // we will add metrics back after sourceLog remove the metrics
- destLog.removeLogMetrics()
- destLog.updateHighWatermark(sourceLog.highWatermark)
+ replaceCurrentWithFutureLog(Option(sourceLog), destLog,
updateHighWatermark = true)
+ }
+ }
- // Now that future replica has been successfully renamed to be the
current replica
- // Update the cached map and log cleaner as appropriate.
- futureLogs.remove(topicPartition)
- currentLogs.put(topicPartition, destLog)
- if (cleaner != null) {
- cleaner.alterCheckpointDir(topicPartition, sourceLog.parentDirFile,
destLog.parentDirFile)
- resumeCleaning(topicPartition)
+ def replaceCurrentWithFutureLog(sourceLog: Option[UnifiedLog], destLog:
UnifiedLog, updateHighWatermark: Boolean = false): Unit = {
+ val topicPartition = destLog.topicPartition
+ info(s"Attempting to replace current log $sourceLog with $destLog for
$topicPartition")
Review Comment:
`sourceLog` could be empty now, so maybe we need to revise the log message
##########
core/src/main/scala/kafka/log/LogManager.scala:
##########
@@ -1178,6 +1178,33 @@ class LogManager(logDirs: Seq[File],
}
}
+ def recoverAbandonedFutureLogs(brokerId: Int, newTopicsImage: TopicsImage):
Unit = {
+ val abandonedFutureLogs = findAbandonedFutureLogs(brokerId, newTopicsImage)
+ abandonedFutureLogs.foreach { case (futureLog, currentLog) =>
+ val tp = futureLog.topicPartition
+ if (cleaner != null) {
Review Comment:
we can replace this `abortAndPauseCleaning(tp)`
##########
core/src/main/scala/kafka/log/LogManager.scala:
##########
@@ -1178,6 +1178,33 @@ class LogManager(logDirs: Seq[File],
}
}
+ def recoverAbandonedFutureLogs(brokerId: Int, newTopicsImage: TopicsImage):
Unit = {
+ val abandonedFutureLogs = findAbandonedFutureLogs(brokerId, newTopicsImage)
+ abandonedFutureLogs.foreach { case (futureLog, currentLog) =>
+ val tp = futureLog.topicPartition
+ if (cleaner != null) {
+ cleaner.abortAndPauseCleaning(tp)
+ }
+
+ replaceCurrentWithFutureLog(currentLog, futureLog)
+
+ info(s"Successfully renamed abandoned future log for $tp")
Review Comment:
line#1189 will print `info(s"The current replica is successfully replaced
with the future replica for $topicPartition")` and that is not much related to
fact, right?
##########
core/src/main/scala/kafka/log/LogManager.scala:
##########
@@ -1178,6 +1178,33 @@ class LogManager(logDirs: Seq[File],
}
}
+ def recoverAbandonedFutureLogs(brokerId: Int, newTopicsImage: TopicsImage):
Unit = {
+ val abandonedFutureLogs = findAbandonedFutureLogs(brokerId, newTopicsImage)
+ abandonedFutureLogs.foreach { case (futureLog, currentLog) =>
+ val tp = futureLog.topicPartition
+ if (cleaner != null) {
Review Comment:
I guess `cleaner.abortAndPauseCleaning` is added because we call
`resumeCleaning` later, and it will cause error if we don't call
`abortAndPauseCleaning` here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
