junrao commented on code in PR #20082:
URL: https://github.com/apache/kafka/pull/20082#discussion_r2211715425
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1150,6 +1155,7 @@ class ReplicaManager(val config: KafkaConfig,
if (partition.futureReplicaDirChanged(destinationDir)) {
replicaAlterLogDirsManager.removeFetcherForPartitions(Set(topicPartition))
partition.removeFutureLocalReplica()
+ logManager.resumeCleaning(topicPartition)
Review Comment:
In theory, it's possible that immediately after the
`partition.futureReplicaDirChanged` check, the future partition catches up and
is replaced as the current partition. During that process,
`logManager.resumeCleaning()` has already been called. Calling it a second time
here will lead to `IllegalStateException`. We could potentially check for the
future replica again after
`replicaAlterLogDirsManager.removeFetcherForPartitions()`. At that point, if
the future replica still exist, it's guaranteed to be there afterward.
##########
core/src/main/scala/kafka/log/LogManager.scala:
##########
@@ -77,7 +77,8 @@ class LogManager(logDirs: Seq[File],
logDirFailureChannel: LogDirFailureChannel,
time: Time,
remoteStorageSystemEnable: Boolean,
- val initialTaskDelayMs: Long) extends Logging {
+ val initialTaskDelayMs: Long,
+ cleanerFactory: (CleanerConfig, util.List[File],
ConcurrentMap[TopicPartition, UnifiedLog], LogDirFailureChannel, Time) =>
LogCleaner = (cleanerConfig, files, map, logDirFailureChannel, time) => new
LogCleaner(cleanerConfig, files, map, logDirFailureChannel, time)) extends
Logging {
Review Comment:
Quite a long time. Could we format it better?
--
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
