soondenana commented on a change in pull request #9054:
URL: https://github.com/apache/kafka/pull/9054#discussion_r459603473
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -1151,6 +1144,12 @@ class LogManager(logDirs: Seq[File],
}
}
}
+
+ private def removeLogAndMetrics(logs: Pool[TopicPartition, Log], tp:
TopicPartition): Log = {
+ val removedLog = logs.remove(tp)
+ if (removedLog != null) removedLog.removeLogMetrics()
+ removedLog
Review comment:
nit: Lets return Option(removedLog) to ease null checking by clients.
Seems like the same object gets returned by `asyncDelete` but is only used
in one place in test code, so many want to change the return value of that too.
The less "null" we have the better.
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -199,27 +199,22 @@ class LogManager(logDirs: Seq[File],
if (cleaner != null)
cleaner.handleLogDirFailure(dir)
- val offlineCurrentTopicPartitions = currentLogs.collect {
- case (tp, log) if log.parentDir == dir => tp
- }
- offlineCurrentTopicPartitions.foreach { topicPartition => {
- val removedLog = currentLogs.remove(topicPartition)
- if (removedLog != null) {
- removedLog.closeHandlers()
- removedLog.removeLogMetrics()
+ def removeOfflineLogs(logs: Pool[TopicPartition, Log]):
Iterable[TopicPartition] = {
Review comment:
Thanks for deduping this code.
----------------------------------------------------------------
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:
us...@infra.apache.org
