MINOR: Remove redundant checkpoint thread started field in ReplicaManager
We have two fields `highWatermarkCheckPointThreadStarted` and `hwThreadInitialized` which appear to be serving the same purpose. This patch gets rid of `hwThreadInitialized`.
Reviewers: Guozhang Wang
@@ -201,7 +201,6 @@ class ReplicaManager(val config: KafkaConfig,
@volatile var highWatermarkCheckpoints = logManager.liveLogDirs.map(dir =>
(dir.getAbsolutePath, new OffsetCheckpointFile(new File(dir, ReplicaManager.HighWatermarkFilename), logDirFailureChannel))).toMap
- private var hwThreadInitialized = false
this.logIdent = s"[ReplicaManager broker=$localBrokerId] "
private val stateChangeLogger = new StateChangeLogger(localBrokerId, inControllerContext = false, None)
@@ -265,7 +264,7 @@ class ReplicaManager(val config: KafkaConfig,
def underReplicatedPartitionCount: Int = leaderPartitionsIterator.count(_.isUnderReplicated)
- def startHighWaterMarksCheckPointThread() = {
+ def startHighWatermarkCheckPointThread() = {
if (highWatermarkCheckPointThreadStarted.compareAndSet(false, true))
scheduler.schedule("highwatermark-checkpoint", checkpointHighWatermarks _, period = config.replicaHighWatermarkCheckpointIntervalMs, unit = TimeUnit.MILLISECONDS)
}
@@ -1136,13 +1135,9 @@ class ReplicaManager(val config: KafkaConfig,
markPartitionOffline(topicPartition)
}
-
// we initialize highwatermark thread after the first leaderisrrequest. This ensures that all the partitions
// have been completely populated before starting the checkpointing there by avoiding weird race conditions
- if (!hwThreadInitialized) {
- startHighWaterMarksCheckPointThread()
- hwThreadInitialized = true
- }
+ startHighWatermarkCheckPointThread()
val futureReplicasAndInitialOffset = new mutable.HashMap[TopicPartition, InitialFetchState]
for (partition <- newPartitions) {
