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, these mocks
will return a non-empty `FetchResponse` as essential safeguards:
- If `initEpoch` is accidentally set equal to `newEpoch`, the response will
be handled, and the leader epoch will be bumped as the leader epoch of
fetchResponsethe test to fail.
- If `AbstractFetcherThread#L334` is accidentally removed, the response will
be handled, and the leader epoch will be bumped as the leader epoch of
fetchResponsethe test to fail.
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
