frankvicky commented on code in PR #19223:
URL: https://github.com/apache/kafka/pull/19223#discussion_r2008619688
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1153,4 +1153,29 @@ class AbstractFetcherThreadTest {
assertTrue(fetcher.fetchState(unknownPartition).isEmpty)
}
-}
+ @Test
+ def testIgnoreFetchResponseWhenLeaderEpochChanged(): Unit = {
+ val newEpoch = 1
+ val initEpoch = 0
+
+ val partition = new TopicPartition("topic", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint,
mockTierStateMachine)
+ val replicaState = PartitionState(leaderEpoch = newEpoch)
+ fetcher.setReplicaState(partition, replicaState)
+ val initFetchState = initialFetchState(topicIds.get(partition.topic), 0L,
leaderEpoch = newEpoch)
+ fetcher.addPartitions(Map(partition -> initFetchState))
+
+ val batch = mkBatch(baseOffset = 0L, leaderEpoch = 0, new
SimpleRecord("a".getBytes))
+ val leaderState = PartitionState(Seq(batch), leaderEpoch = initEpoch,
highWatermark = 1L)
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+
fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
Review Comment:
I think only L1173 is redundant.
Technically, these lines don't affect the test result if `initEpoch !=
newEpoch`.
But since these lines implement the mock mechanism of fetch, they provide
essential safeguards:
- If `initEpoch` is accidentally set equal to `newEpoch`, these mock lines
will trigger specific logic that causes the test to fail.
- If `AbstractFetcherThread#L334` is accidentally removed, the mock will
return an empty response, which will cause test assertion failures.
Given that, we should keep these lines. WDYT?
--
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
