dajac commented on a change in pull request #11942:
URL: https://github.com/apache/kafka/pull/11942#discussion_r836134933
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
##########
@@ -1300,6 +1300,62 @@ class ReplicaManagerTest {
TestUtils.assertNoNonDaemonThreads(this.getClass.getName)
}
+ @Test
+ def testHasPreferredReplica(): Unit = {
Review comment:
nit: Would
`testFetchShouldReturnImmediatelyWhenPreferredReadReplicaIsDefined` be a better
name?
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
##########
@@ -1300,6 +1300,62 @@ class ReplicaManagerTest {
TestUtils.assertNoNonDaemonThreads(this.getClass.getName)
}
+ @Test
+ def testHasPreferredReplica(): Unit = {
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new
MockTimer(time),
+ propsModifier = props => props.put(KafkaConfig.ReplicaSelectorClassProp,
"org.apache.kafka.common.replica.RackAwareReplicaSelector"))
+
+ 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(
+ any[TopicPartition], any[ListenerName])).
+ thenReturn(Map(leaderBrokerId -> new Node(leaderBrokerId, "host1",
9092, "rack-a"),
+ followerBrokerId -> new Node(followerBrokerId, "host2", 9092,
"rack-b")).toMap)
+
+ // Make this replica the leader
+ val leaderAndIsrRequest2 = new
LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0,
brokerEpoch,
Review comment:
nit: `leaderAndIsrRequest2` -> `leaderAndIsrRequest`?
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
##########
@@ -1300,6 +1300,62 @@ class ReplicaManagerTest {
TestUtils.assertNoNonDaemonThreads(this.getClass.getName)
}
+ @Test
+ def testHasPreferredReplica(): Unit = {
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new
MockTimer(time),
+ propsModifier = props => props.put(KafkaConfig.ReplicaSelectorClassProp,
"org.apache.kafka.common.replica.RackAwareReplicaSelector"))
+
+ 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(
+ any[TopicPartition], any[ListenerName])).
+ thenReturn(Map(leaderBrokerId -> new Node(leaderBrokerId, "host1",
9092, "rack-a"),
+ followerBrokerId -> new Node(followerBrokerId, "host2", 9092,
"rack-b")).toMap)
+
+ // Make this replica the leader
+ val leaderAndIsrRequest2 = new
LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0,
brokerEpoch,
+ Seq(new LeaderAndIsrPartitionState()
+ .setTopicName(topic)
+ .setPartitionIndex(0)
+ .setControllerEpoch(0)
+ .setLeader(0)
+ .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, leaderAndIsrRequest2, (_, _) =>
())
+ // Avoid the replica selector ignore the follower replica if it not have
the data that need to fetch
+
replicaManager.getPartitionOrException(tp0).updateFollowerFetchState(followerBrokerId,
new LogOffsetMetadata(0), 0, 0, 0)
+
+ val metadata = new DefaultClientMetadata("rack-b", "client-id",
+ InetAddress.getByName("localhost"), KafkaPrincipal.ANONYMOUS,
"default")
+
+ //If a preferred read replica is selected, the fetch response will
return immediately，even with minBytes and timeout configured
Review comment:
* nit: Could we add a space after `//` to be consistent?
* nit: I would rephrase the comment as follow: `If a preferred read replica
is selected, the fetch response returns immediately, even if min bytes and
timeout conditions are not met.`. What do you think?
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
##########
@@ -1300,6 +1300,62 @@ class ReplicaManagerTest {
TestUtils.assertNoNonDaemonThreads(this.getClass.getName)
}
+ @Test
+ def testHasPreferredReplica(): Unit = {
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new
MockTimer(time),
+ propsModifier = props => props.put(KafkaConfig.ReplicaSelectorClassProp,
"org.apache.kafka.common.replica.RackAwareReplicaSelector"))
+
+ 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(
+ any[TopicPartition], any[ListenerName])).
+ thenReturn(Map(leaderBrokerId -> new Node(leaderBrokerId, "host1",
9092, "rack-a"),
+ followerBrokerId -> new Node(followerBrokerId, "host2", 9092,
"rack-b")).toMap)
Review comment:
nit: I find this block hard to read. I usually tend to format
`when`/`then` as follow:
```scala
when(replicaManager.metadataCache.getPartitionReplicaEndpoints(
any[TopicPartition],
any[ListenerName])
).thenReturn(Map(
leaderBrokerId -> new Node(leaderBrokerId, "host1", 9092, "rack-a"),
followerBrokerId -> new Node(followerBrokerId, "host2", 9092, "rack-b")
).toMap)
```
Could we also replace the two `any`s with the expected values?
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
##########
@@ -1300,6 +1300,62 @@ class ReplicaManagerTest {
TestUtils.assertNoNonDaemonThreads(this.getClass.getName)
}
+ @Test
+ def testHasPreferredReplica(): Unit = {
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new
MockTimer(time),
+ propsModifier = props => props.put(KafkaConfig.ReplicaSelectorClassProp,
"org.apache.kafka.common.replica.RackAwareReplicaSelector"))
+
+ 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(
+ any[TopicPartition], any[ListenerName])).
+ thenReturn(Map(leaderBrokerId -> new Node(leaderBrokerId, "host1",
9092, "rack-a"),
+ followerBrokerId -> new Node(followerBrokerId, "host2", 9092,
"rack-b")).toMap)
+
+ // Make this replica the leader
+ val leaderAndIsrRequest2 = new
LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0,
brokerEpoch,
+ Seq(new LeaderAndIsrPartitionState()
+ .setTopicName(topic)
+ .setPartitionIndex(0)
+ .setControllerEpoch(0)
+ .setLeader(0)
+ .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, leaderAndIsrRequest2, (_, _) =>
())
+ // Avoid the replica selector ignore the follower replica if it not have
the data that need to fetch
+
replicaManager.getPartitionOrException(tp0).updateFollowerFetchState(followerBrokerId,
new LogOffsetMetadata(0), 0, 0, 0)
+
+ val metadata = new DefaultClientMetadata("rack-b", "client-id",
+ InetAddress.getByName("localhost"), KafkaPrincipal.ANONYMOUS,
"default")
Review comment:
nit: `InetAddress.getByName("localhost")` -> `InetAddress.getLocalHost`?
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
##########
@@ -1300,6 +1300,62 @@ class ReplicaManagerTest {
TestUtils.assertNoNonDaemonThreads(this.getClass.getName)
}
+ @Test
+ def testHasPreferredReplica(): Unit = {
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new
MockTimer(time),
+ propsModifier = props => props.put(KafkaConfig.ReplicaSelectorClassProp,
"org.apache.kafka.common.replica.RackAwareReplicaSelector"))
+
+ 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(
+ any[TopicPartition], any[ListenerName])).
+ thenReturn(Map(leaderBrokerId -> new Node(leaderBrokerId, "host1",
9092, "rack-a"),
+ followerBrokerId -> new Node(followerBrokerId, "host2", 9092,
"rack-b")).toMap)
+
+ // Make this replica the leader
+ val leaderAndIsrRequest2 = new
LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0,
brokerEpoch,
+ Seq(new LeaderAndIsrPartitionState()
+ .setTopicName(topic)
+ .setPartitionIndex(0)
+ .setControllerEpoch(0)
+ .setLeader(0)
+ .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, leaderAndIsrRequest2, (_, _) =>
())
+ // Avoid the replica selector ignore the follower replica if it not have
the data that need to fetch
+
replicaManager.getPartitionOrException(tp0).updateFollowerFetchState(followerBrokerId,
new LogOffsetMetadata(0), 0, 0, 0)
+
+ val metadata = new DefaultClientMetadata("rack-b", "client-id",
+ InetAddress.getByName("localhost"), KafkaPrincipal.ANONYMOUS,
"default")
+
+ //If a preferred read replica is selected, the fetch response will
return immediately，even with minBytes and timeout configured
+ val consumerResult = fetchAsConsumer(replicaManager, tidp0,
+ new PartitionData(Uuid.ZERO_UUID, 0, 0, 100000, Optional.empty()),
minBytes = 1,
+ clientMetadata = Some(metadata), timeout = 5000)
+
+ // Fetch from leader succeeds
+ assertTrue(consumerResult.isFired)
+
+ // No delayed fetch was inserted
+ assertEquals(0, replicaManager.delayedFetchPurgatory.watched)
+
+ // Returns a preferred replica,not None
Review comment:
nit: I think that we can remove `,not None` 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
