[GitHub] [kafka] showuon commented on a diff in pull request #12783: KAFKA-14334: complete delayed purgatory after replication

[GitBox <gi...@apache.org>]

showuon commented on code in PR #12783:
URL: https://github.com/apache/kafka/pull/12783#discussion_r1013697955


##########
core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala:
##########
@@ -1100,6 +1102,66 @@ class ReplicaFetcherThreadTest {
     assertEquals(Collections.singletonList(tid1p0), fetchRequestBuilder2.removed())
   }
 
+  @ParameterizedTest
+  @ValueSource(booleans = Array(true, false))
+  def testLocalFetchCompletionIfHighWatermarkUpdated(highWatermarkUpdated: Boolean): Unit = {
+    val props = TestUtils.createBrokerConfig(1, "localhost:1234")
+    val config = KafkaConfig.fromProps(props)
+    val highWatermarkReceivedFromLeader = 100L
+
+    val mockBlockingSend: BlockingSend = mock(classOf[BlockingSend])
+    when(mockBlockingSend.brokerEndPoint()).thenReturn(brokerEndPoint)
+
+    var maybeNewHighWatermark: Option[Long] = None
+    if (highWatermarkUpdated) {
+      maybeNewHighWatermark = Some(highWatermarkReceivedFromLeader)
+    }
+    val log: UnifiedLog = mock(classOf[UnifiedLog])
+    when(log.maybeUpdateHighWatermark(highWatermarkReceivedFromLeader))
+      .thenReturn(maybeNewHighWatermark)
+
+    val appendInfo: Option[LogAppendInfo] = Some(mock(classOf[LogAppendInfo]))
+
+    val partition: Partition = mock(classOf[Partition])
+    when(partition.localLogOrException).thenReturn(log)
+    when(partition.appendRecordsToFollowerOrFutureReplica(any[MemoryRecords], any[Boolean])).thenReturn(appendInfo)
+
+    val replicaManager: ReplicaManager = mock(classOf[ReplicaManager])
+    when(replicaManager.getPartitionOrException(any[TopicPartition])).thenReturn(partition)
+    val brokerTopicStats = new BrokerTopicStats
+    when(replicaManager.brokerTopicStats).thenReturn(brokerTopicStats)
+
+    val replicaQuota: ReplicaQuota = mock(classOf[ReplicaQuota])
+
+    val thread = createReplicaFetcherThread(
+      name = "replica-fetcher",
+      fetcherId = 0,
+      brokerConfig = config,
+      failedPartitions = failedPartitions,
+      replicaMgr = replicaManager,
+      quota = replicaQuota,
+      leaderEndpointBlockingSend = mockBlockingSend)
+
+    val tp0 = new TopicPartition("testTopic", 0)
+    val tp1 = new TopicPartition("testTopic", 1)
+    val records = MemoryRecords.withRecords(CompressionType.NONE,
+      new SimpleRecord(1000, "foo".getBytes(StandardCharsets.UTF_8)))
+    val partitionData = new FetchResponseData.PartitionData()
+      .setRecords(records)
+      .setHighWatermark(highWatermarkReceivedFromLeader)
+
+    thread.processPartitionData(tp0, 0, partitionData.setPartitionIndex(0))
+    thread.processPartitionData(tp1, 0, partitionData.setPartitionIndex(1))
+    thread.doWork()
+
+    if (highWatermarkUpdated) {
+      verify(replicaManager, times(1)).completeDelayedFetchRequests(Seq(tp0, tp1))

Review Comment:
   nit: Could we verify `completeDelayedFetchRequets` is not called before invoking `doWork` ?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org