(kafka) branch trunk updated: KAFKA-16013: Add metric for expiration rate of delayed remote fetch (#15014)

[sh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

showuon pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 52496dcd385 KAFKA-16013: Add metric for expiration rate of delayed remote fetch (#15014)
52496dcd385 is described below

commit 52496dcd385acc6fcbae93396c47f90c0cf5cfb6
Author: Nikhil Ramakrishnan <ni...@amazon.com>
AuthorDate: Fri Dec 15 11:21:39 2023 +0000

    KAFKA-16013: Add metric for expiration rate of delayed remote fetch (#15014)
    
    Add metric for the number of expired remote fetches per second, and corresponding unit test to verify that the metric is marked on expiration.
    
    kafka.server:type=DelayedRemoteFetchMetrics,name=ExpiresPerSec
    
    Reviewers: Luke Chen <sh...@gmail.com>, Satish Duggana <sa...@apache.org>, Kamal Chandraprakash<ka...@gmail.com>, Christo Lolov <lo...@amazon.com>
---
 .../scala/kafka/server/DelayedRemoteFetch.scala    | 15 +++++-
 .../kafka/server/DelayedRemoteFetchTest.scala      | 59 ++++++++++++++++++++--
 2 files changed, 69 insertions(+), 5 deletions(-)

diff --git a/core/src/main/scala/kafka/server/DelayedRemoteFetch.scala b/core/src/main/scala/kafka/server/DelayedRemoteFetch.scala
index 5fcf851fe83..143201f3d83 100644
--- a/core/src/main/scala/kafka/server/DelayedRemoteFetch.scala
+++ b/core/src/main/scala/kafka/server/DelayedRemoteFetch.scala
@@ -20,9 +20,10 @@ package kafka.server
 import org.apache.kafka.common.TopicIdPartition
 import org.apache.kafka.common.errors._
 import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.server.metrics.KafkaMetricsGroup
 import org.apache.kafka.storage.internals.log.{FetchParams, FetchPartitionData, LogOffsetMetadata, RemoteLogReadResult, RemoteStorageFetchInfo}
 
-import java.util.concurrent.{CompletableFuture, Future}
+import java.util.concurrent.{CompletableFuture, Future, TimeUnit}
 import java.util.{Optional, OptionalInt, OptionalLong}
 import scala.collection._
 
@@ -79,7 +80,12 @@ class DelayedRemoteFetch(remoteFetchTask: Future[Void],
   override def onExpiration(): Unit = {
     // cancel the remote storage read task, if it has not been executed yet
     val cancelled = remoteFetchTask.cancel(true)
-    if (!cancelled) debug(s"Remote fetch task for for RemoteStorageFetchInfo: $remoteFetchInfo could not be cancelled and its isDone value is ${remoteFetchTask.isDone}")
+    if (!cancelled) debug(s"Remote fetch task for RemoteStorageFetchInfo: $remoteFetchInfo could not be cancelled and its isDone value is ${remoteFetchTask.isDone}")
+
+    if (fetchParams.isFromFollower)
+      warn(s"The follower should not invoke remote fetch. Fetch params are: $fetchParams")
+    else
+      DelayedRemoteFetchMetrics.expiredRequestMeter.mark()
   }
 
   /**
@@ -114,3 +120,8 @@ class DelayedRemoteFetch(remoteFetchTask: Future[Void],
     responseCallback(fetchPartitionData)
   }
 }
+
+object DelayedRemoteFetchMetrics {
+  private val metricsGroup = new KafkaMetricsGroup(DelayedRemoteFetchMetrics.getClass)
+  val expiredRequestMeter = metricsGroup.newMeter("ExpiresPerSec", "requests", TimeUnit.SECONDS)
+}
diff --git a/core/src/test/scala/integration/kafka/server/DelayedRemoteFetchTest.scala b/core/src/test/scala/integration/kafka/server/DelayedRemoteFetchTest.scala
index 007154f46de..c943258da8f 100644
--- a/core/src/test/scala/integration/kafka/server/DelayedRemoteFetchTest.scala
+++ b/core/src/test/scala/integration/kafka/server/DelayedRemoteFetchTest.scala
@@ -22,15 +22,16 @@ import org.apache.kafka.common.protocol.{ApiKeys, Errors}
 import org.apache.kafka.common.record.MemoryRecords
 import org.apache.kafka.common.requests.FetchRequest
 import org.apache.kafka.common.{TopicIdPartition, Uuid}
+import org.apache.kafka.server.metrics.KafkaYammerMetrics
 import org.apache.kafka.storage.internals.log._
 import org.junit.jupiter.api.Assertions._
 import org.junit.jupiter.api.Test
-import org.mockito.Mockito.{mock, when}
+import org.mockito.Mockito.{mock, verify, when}
 
 import java.util.Optional
-import java.util.concurrent.CompletableFuture
-
+import java.util.concurrent.{CompletableFuture, Future}
 import scala.collection._
+import scala.jdk.CollectionConverters._
 
 class DelayedRemoteFetchTest {
   private val maxBytes = 1024
@@ -143,6 +144,58 @@ class DelayedRemoteFetchTest {
     assertEquals(Errors.FENCED_LEADER_EPOCH, fetchResultOpt.get.error)
   }
 
+  @Test
+  def testRequestExpiry(): Unit = {
+    var actualTopicPartition: Option[TopicIdPartition] = None
+    var fetchResultOpt: Option[FetchPartitionData] = None
+
+    def callback(responses: Seq[(TopicIdPartition, FetchPartitionData)]): Unit = {
+      assertEquals(1, responses.size)
+      actualTopicPartition = Some(responses.head._1)
+      fetchResultOpt = Some(responses.head._2)
+    }
+
+    val replicaId = -1
+    val highWatermark = 100
+    val leaderLogStartOffset = 10
+
+    val remoteFetchTask = mock(classOf[Future[Void]])
+    val future: CompletableFuture[RemoteLogReadResult] = new CompletableFuture[RemoteLogReadResult]()
+    val fetchInfo: RemoteStorageFetchInfo = new RemoteStorageFetchInfo(0, false, topicIdPartition.topicPartition(), null, null, false)
+    val fetchParams = buildFollowerFetchParams(replicaId, maxWaitMs = 500)
+    val logReadInfo = buildReadResult(Errors.NONE, highWatermark, leaderLogStartOffset)
+
+    val delayedRemoteFetch = new DelayedRemoteFetch(remoteFetchTask, future, fetchInfo, Seq(topicIdPartition -> fetchStatus), fetchParams,
+      Seq(topicIdPartition -> logReadInfo), replicaManager, callback)
+
+    when(replicaManager.getPartitionOrException(topicIdPartition.topicPartition))
+      .thenReturn(mock(classOf[Partition]))
+
+    // Verify that the ExpiresPerSec metric is zero before fetching
+    val metrics = KafkaYammerMetrics.defaultRegistry.allMetrics
+    assertEquals(0, metrics.keySet.asScala.count(_.getMBeanName == "kafka.server:type=DelayedRemoteFetchMetrics,name=ExpiresPerSec"))
+
+    // Force the delayed remote fetch to expire
+    delayedRemoteFetch.run()
+
+    // Check that the task was cancelled and force-completed
+    verify(remoteFetchTask).cancel(true)
+    assertTrue(delayedRemoteFetch.isCompleted)
+
+    // Check that the ExpiresPerSec metric was incremented
+    assertEquals(1, metrics.keySet.asScala.count(_.getMBeanName == "kafka.server:type=DelayedRemoteFetchMetrics,name=ExpiresPerSec"))
+
+    // Fetch results should still include local read results
+    assertTrue(actualTopicPartition.isDefined)
+    assertEquals(topicIdPartition, actualTopicPartition.get)
+    assertTrue(fetchResultOpt.isDefined)
+
+    val fetchResult = fetchResultOpt.get
+    assertEquals(Errors.NONE, fetchResult.error)
+    assertEquals(highWatermark, fetchResult.highWatermark)
+    assertEquals(leaderLogStartOffset, fetchResult.logStartOffset)
+  }
+
   private def buildFollowerFetchParams(replicaId: Int,
                                        maxWaitMs: Int): FetchParams = {
     new FetchParams(