log time taken to handle LeaderAndIsr request
Reviewers: Lucas Bradstreet Jun Rao
@@ -1230,6 +1230,7 @@ class ReplicaManager(val config: KafkaConfig,
def becomeLeaderOrFollower(correlationId: Int,
leaderAndIsrRequest: LeaderAndIsrRequest,
onLeadershipChange: (Iterable[Partition], Iterable[Partition]) => Unit): LeaderAndIsrResponse = {
+ val startMs = time.milliseconds()
replicaStateChangeLock synchronized {
val controllerId = leaderAndIsrRequest.controllerId
val requestPartitionStates = leaderAndIsrRequest.partitionStates.asScala
@@ -1242,6 +1243,7 @@ class ReplicaManager(val config: KafkaConfig,
s"epoch ${leaderAndIsrRequest.controllerEpoch}")
}
+ val response = {
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. " +
@@ -1368,6 +1370,12 @@ class ReplicaManager(val config: KafkaConfig,
.setPartitionErrors(responsePartitions.asJava))
}
}
+ val endMs = time.milliseconds()
+ val elapsedMs = endMs - startMs
+ stateChangeLogger.info(s"Finished LeaderAndIsr request in ${elapsedMs}ms correlationId $correlationId from controller " +
+ s"$controllerId for ${requestPartitionStates.size} partitions")
+ response
+ }
}
private def maybeAddLogDirFetchers(partitions: Set[Partition],
