jeffkbkim commented on code in PR #12783:
URL: https://github.com/apache/kafka/pull/12783#discussion_r1007204668
##########
core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala:
##########
@@ -1100,6 +1100,66 @@ class ReplicaFetcherThreadTest {
assertEquals(Collections.singletonList(tid1p0),
fetchRequestBuilder2.removed())
}
+ @Test
+ def testFetchResponseWithPartitionData(): Unit = {
+ val appendInfo: LogAppendInfo = mock(classOf[LogAppendInfo])
+ verifyLocalFetchCompletionAfterReplication(Some(appendInfo))
+ }
+
+ @Test
+ def testFetchResponseWithNoPartitionData(): Unit = {
+ verifyLocalFetchCompletionAfterReplication(appendInfo = None)
+ }
+
+ private def verifyLocalFetchCompletionAfterReplication(appendInfo:
Option[LogAppendInfo]): Unit = {
+ val props = TestUtils.createBrokerConfig(1, "localhost:1234")
+ val config = KafkaConfig.fromProps(props)
+
+ val mockBlockingSend: BlockingSend = mock(classOf[BlockingSend])
+ when(mockBlockingSend.brokerEndPoint()).thenReturn(brokerEndPoint)
+
+ val log: UnifiedLog = mock(classOf[UnifiedLog])
+
+ 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: thread.FetchData = new FetchResponseData.PartitionData()
+ .setRecords(records)
+
+ thread.processPartitionData(tp0, 0, partitionData.setPartitionIndex(0))
+ thread.processPartitionData(tp1, 0, partitionData.setPartitionIndex(1))
+ thread.doWork()
+ appendInfo match {
+ case Some(_) =>
+ val argument: ArgumentCaptor[Seq[TopicPartition]] =
ArgumentCaptor.forClass(classOf[Seq[TopicPartition]])
+ verify(replicaManager,
times(1)).completeDelayedFetchRequests(argument.capture())
+ assertEquals(Seq(tp0, tp1), argument.getValue)
+ case None =>
+ verify(replicaManager,
times(0)).completeDelayedFetchRequests(any[Seq[TopicPartition]])
Review Comment:
i'm not sure i follow. replicaManager.completeDelayedFetchRequests is not
called when appendInfo is empty. i will add a check for both cases to assert
that the underlying buffer is empty after thread.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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
