bozhao12 commented on a change in pull request #11965:
URL: https://github.com/apache/kafka/pull/11965#discussion_r839401608
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
##########
@@ -1300,6 +1301,59 @@ class ReplicaManagerTest {
TestUtils.assertNoNonDaemonThreads(this.getClass.getName)
}
+ @Test
+ def testFetchFromFollowerShouldNotRunPreferLeaderSelect(): Unit = {
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new
MockTimer(time),
+ propsModifier = props => props.put(KafkaConfig.ReplicaSelectorClassProp,
"kafka.test.MockReplicaSelector"))
+ try {
+ val leaderBrokerId = 0
+ val followerBrokerId = 1
+ val brokerList = Seq[Integer](leaderBrokerId, followerBrokerId).asJava
+ val topicId = Uuid.randomUuid()
+ val tp0 = new TopicPartition(topic, 0)
+ val tidp0 = new TopicIdPartition(topicId, tp0)
+ initializeLogAndTopicId(replicaManager, tp0, topicId)
+ when(replicaManager.metadataCache.getPartitionReplicaEndpoints(
+ tp0,
+ new ListenerName("default")
+ )).thenReturn(Map(
+ leaderBrokerId -> new Node(leaderBrokerId, "host1", 9092, "rack-a"),
+ followerBrokerId -> new Node(followerBrokerId, "host2", 9092, "rack-a")
+ ).toMap)
+
+ // Make this replica the follower
+ val leaderAndIsrRequest = new
LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0,
brokerEpoch,
+ Seq(new LeaderAndIsrPartitionState()
+ .setTopicName(topic)
+ .setPartitionIndex(0)
+ .setControllerEpoch(0)
+ .setLeader(1)
+ .setLeaderEpoch(1)
+ .setIsr(brokerList)
+ .setZkVersion(0)
+ .setReplicas(brokerList)
+ .setIsNew(false)).asJava,
+ Collections.singletonMap(topic, topicId),
+ Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
+ replicaManager.becomeLeaderOrFollower(1, leaderAndIsrRequest, (_, _) =>
())
+
+ val metadata = new DefaultClientMetadata("rack-a", "client-id",
+ InetAddress.getLocalHost, KafkaPrincipal.ANONYMOUS, "default")
+
+ val consumerResult = fetchAsConsumer(replicaManager, tidp0,
+ new PartitionData(Uuid.ZERO_UUID, 0, 0, 100000,
+ Optional.empty()), clientMetadata = Some(metadata))
+
+ // Fetch from follower succeeds
+ assertTrue(consumerResult.isFired)
+
+ // Expect not run the preferred read replica selection
+ assertEquals(0,
replicaManager.replicaSelectorOpt.get.asInstanceOf[MockReplicaSelector].getTriggerElectionCount)
+
+ } finally replicaManager.shutdown(checkpointHW = false)
+ }
+
+
Review comment:
@dajac All comments has addresses. Thanks.
--
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
