dajac commented on code in PR #12783:
URL: https://github.com/apache/kafka/pull/12783#discussion_r1006836805
##########
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)
Review Comment:
nit: In the case where `appendInfo` is `None`, it is a tad awkward that we
set records here.
--
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
