soarez commented on code in PR #15136:
URL: https://github.com/apache/kafka/pull/15136#discussion_r1549709937
##########
core/src/main/scala/kafka/log/LogManager.scala:
##########
@@ -1173,6 +1173,35 @@ class LogManager(logDirs: Seq[File],
}
}
+ def recoverAbandonedFutureLogs(brokerId: Int, newTopicsImage: TopicsImage):
Unit = {
+ val abandonedFutureLogs = findAbandonedFutureLogs(brokerId, newTopicsImage)
+ abandonedFutureLogs.foreach { log =>
+ val tp = log.topicPartition
+
+ log.renameDir(UnifiedLog.logDirName(tp), shouldReinitialize = true)
+ log.removeLogMetrics()
+ futureLogs.remove(tp)
+
+ currentLogs.put(tp, log)
+ log.newMetrics()
+
+ info(s"Successfully renamed abandoned future log for $tp")
+ }
+ }
+
+ private def findAbandonedFutureLogs(brokerId: Int, newTopicsImage:
TopicsImage): Iterable[UnifiedLog] = {
+ futureLogs.values.flatMap { log =>
+ val topicId = log.topicId.getOrElse {
+ throw new RuntimeException(s"The log dir $log does not have a topic
ID, " +
+ "which is not allowed when running in KRaft mode.")
+ }
+ val partitionId = log.topicPartition.partition()
+ Option(newTopicsImage.getPartition(topicId, partitionId))
+ .filter(pr =>
directoryId(log.parentDir).contains(pr.directory(brokerId)))
+ .map(_ => log)
Review Comment:
If we didn't know if the future log was caught up or not, then I'd prefer
(a), but at this point I can't conceive of a different scenario – other than a
failure during replica promotion – that would cause the future log to be in the
directory assigned in the metadata. So I agree that the two logs likely will be
either caught up or very close. So I agree it makes more sense to do (b) -
promote the future log and delete the main one.
We can still run into trouble if the directory with the main replica is
offline. At some point that will cause a crash if the directory ever comes back
online. But there's nothing we can do about that here. Maybe future work could
improve how the broker handles loading conflicting logs.
--
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
