Add log if process fetch takes more than replicaLagTimeMaxMs

linkedin · Aug 16, 2023 · 50e372f · 50e372f
1 parent 73cdd9b
commit 50e372f
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 3 deletions.
diff --git a/core/src/main/scala/kafka/server/AbstractAsyncFetcher.scala b/core/src/main/scala/kafka/server/AbstractAsyncFetcher.scala
@@ -88,7 +88,8 @@ abstract class AbstractAsyncFetcher(name: String,
                                      failedPartitions: FailedPartitions,
                                      fetchBackOffMs: Int = 0,
                                      val brokerTopicStats: BrokerTopicStats,  //BrokerTopicStats's lifecycle managed by ReplicaManager
-                                     fetcherEventBus: FetcherEventBus) extends FetcherEventProcessor with Logging {
+                                     fetcherEventBus: FetcherEventBus,
+                                     replicaMaxLagMs: Long) extends FetcherEventProcessor with Logging {
 
   type FetchData = FetchResponseData.PartitionData
   type EpochData = OffsetForLeaderEpochRequestData.OffsetForLeaderPartition
@@ -166,10 +167,21 @@ abstract class AbstractAsyncFetcher(name: String,
         trace(s"There are no active partitions. Back off for $fetchBackOffMs ms before sending a fetch request")
         true
       case Some(ReplicaFetch(sessionPartitions, fetchRequest)) =>
-        processFetchRequest(sessionPartitions, fetchRequest)
+        processFetchRequestAndMaybeLog(sessionPartitions, fetchRequest)
     }
   }
 
+  private def processFetchRequestAndMaybeLog(sessionPartitions: util.Map[TopicPartition, FetchRequest.PartitionData],
+    fetchRequest: FetchRequest.Builder): Boolean = {
+    val startTime = System.currentTimeMillis()
+    val result = processFetchRequest(sessionPartitions, fetchRequest)
+    val processFetchRequestTotalTime = System.currentTimeMillis() - startTime
+    if (processFetchRequestTotalTime >= replicaMaxLagMs) {
+      warn(s"Process fetch request took $processFetchRequestTotalTime ms. Fetch request: $fetchRequest")
+    }
+    result
+  }
+
   // deal with partitions with errors, potentially due to leadership changes
   private def handlePartitionsWithErrors(partitions: Iterable[TopicPartition], methodName: String): Unit = {
     if (partitions.nonEmpty) {

diff --git a/core/src/main/scala/kafka/server/AsyncReplicaFetcher.scala b/core/src/main/scala/kafka/server/AsyncReplicaFetcher.scala
@@ -66,7 +66,8 @@ class AsyncReplicaFetcher(name: String,
                                 failedPartitions,
                                 fetchBackOffMs = brokerConfig.replicaFetchBackoffMs,
                                 replicaMgr.brokerTopicStats,
-                                fetcherEventBus = fetcherEventBus) {
+                                fetcherEventBus = fetcherEventBus,
+                                replicaMaxLagMs = brokerConfig.replicaLagTimeMaxMs) {
 
   private val replicaId = brokerConfig.brokerId
   private val logContext = new LogContext(s"[ReplicaFetcher replicaId=$replicaId, leaderId=${sourceBroker.id}, " +