MINOR: Serialize state change logs for handling LeaderAndIsr and Stop…

…Replica requests (apache#8493) This patch moves the state change logger logs for handling a LeaderAndIsr/StopReplica request inside the replicaStateChangeLock in order to serialize the logs. This helps to tell apart per-partition actions of concurrent LAIR/StopReplica requests in cases where requests pile up waiting on the lock. Reviewer: Jun Rao <[email protected]>
sayandigital · Apr 15, 2020 · bd42734 · bd42734
1 parent 0a50973
commit bd42734
Showing 1 changed file with 20 additions and 20 deletions.
diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -362,16 +362,16 @@ class ReplicaManager(val config: KafkaConfig,
                    brokerEpoch: Long,
                    partitionStates: Map[TopicPartition, StopReplicaPartitionState]
                   ): (mutable.Map[TopicPartition, Errors], Errors) = {
-    stateChangeLogger.info(s"Handling StopReplica request correlationId $correlationId from controller " +
-      s"$controllerId for ${partitionStates.size} partitions")
-    if (stateChangeLogger.isTraceEnabled)
-      partitionStates.foreach { case (topicPartition, partitionState) =>
-        stateChangeLogger.trace(s"Received StopReplica request $partitionState " +
-          s"correlation id $correlationId from controller $controllerId " +
-          s"epoch $controllerEpoch for partition $topicPartition")
-      }
-
     replicaStateChangeLock synchronized {
+      stateChangeLogger.info(s"Handling StopReplica request correlationId $correlationId from controller " +
+        s"$controllerId for ${partitionStates.size} partitions")
+      if (stateChangeLogger.isTraceEnabled)
+        partitionStates.foreach { case (topicPartition, partitionState) =>
+          stateChangeLogger.trace(s"Received StopReplica request $partitionState " +
+            s"correlation id $correlationId from controller $controllerId " +
+            s"epoch $controllerEpoch for partition $topicPartition")
+        }
+
       val responseMap = new collection.mutable.HashMap[TopicPartition, Errors]
       if (controllerEpoch < this.controllerEpoch) {
         stateChangeLogger.warn(s"Ignoring StopReplica request from " +
@@ -1237,18 +1237,18 @@ class ReplicaManager(val config: KafkaConfig,
   def becomeLeaderOrFollower(correlationId: Int,
                              leaderAndIsrRequest: LeaderAndIsrRequest,
                              onLeadershipChange: (Iterable[Partition], Iterable[Partition]) => Unit): LeaderAndIsrResponse = {
-    val controllerId = leaderAndIsrRequest.controllerId
-    val requestPartitionStates = leaderAndIsrRequest.partitionStates.asScala
-    stateChangeLogger.info(s"Handling LeaderAndIsr request correlationId $correlationId from controller " +
-      s"$controllerId for ${requestPartitionStates.size} partitions")
-    if (stateChangeLogger.isTraceEnabled)
-      requestPartitionStates.foreach { partitionState =>
-        stateChangeLogger.trace(s"Received LeaderAndIsr request $partitionState " +
-          s"correlation id $correlationId from controller $controllerId " +
-          s"epoch ${leaderAndIsrRequest.controllerEpoch}")
-      }
-
     replicaStateChangeLock synchronized {
+      val controllerId = leaderAndIsrRequest.controllerId
+      val requestPartitionStates = leaderAndIsrRequest.partitionStates.asScala
+      stateChangeLogger.info(s"Handling LeaderAndIsr request correlationId $correlationId from controller " +
+        s"$controllerId for ${requestPartitionStates.size} partitions")
+      if (stateChangeLogger.isTraceEnabled)
+        requestPartitionStates.foreach { partitionState =>
+          stateChangeLogger.trace(s"Received LeaderAndIsr request $partitionState " +
+            s"correlation id $correlationId from controller $controllerId " +
+            s"epoch ${leaderAndIsrRequest.controllerEpoch}")
+        }
+
       if (leaderAndIsrRequest.controllerEpoch < controllerEpoch) {
         stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from controller $controllerId with " +
           s"correlation id $correlationId since its controller epoch ${leaderAndIsrRequest.controllerEpoch} is old. " +