chia7712 commented on code in PR #19966:
URL: https://github.com/apache/kafka/pull/19966#discussion_r2148790261
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -979,20 +973,20 @@ class ReplicaManagerTest {
new LazyOffsetCheckpoints(rm.highWatermarkCheckpoints.asJava), None)
// Make this replica the leader.
- val leaderAndIsrRequest1 = new LeaderAndIsrRequest.Builder(0, 0,
brokerEpoch,
- Seq(new LeaderAndIsrRequest.PartitionState()
- .setTopicName(topic)
- .setPartitionIndex(0)
- .setControllerEpoch(0)
- .setLeader(0)
- .setLeaderEpoch(0)
- .setIsr(brokerList)
- .setPartitionEpoch(0)
- .setReplicas(brokerList)
- .setIsNew(false)).asJava,
- topicIds.asJava,
- Set(new Node(0, "host1", 0), new Node(1, "host2", 1), new Node(2,
"host2", 2)).asJava).build()
- rm.becomeLeaderOrFollower(0, leaderAndIsrRequest1, (_, _) => ())
+ val leaderDelta = new TopicsDelta(TopicsImage.EMPTY)
+ leaderDelta.replay(new TopicRecord().setName(topic).setTopicId(topicId))
+ leaderDelta.replay(new PartitionRecord()
+ .setPartitionId(0)
+ .setTopicId(topicId)
+ .setReplicas(brokerList)
+ .setIsr(brokerList)
+ .setRemovingReplicas(Collections.emptyList())
Review Comment:
we don't need to set `Collections.emptyList()` as it's default value is
empty list. please fix it for all `PartitionRecord`
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -1181,68 +1149,6 @@ class ReplicaManagerTest {
val fetch1 = successfulFetch.headOption.filter(_._1 ==
inconsistentTidp).map(_._2)
assertTrue(fetch1.isDefined)
assertEquals(Errors.INCONSISTENT_TOPIC_ID, fetch1.get.error)
-
- // Simulate where the fetch request did not use topic IDs
- // Fetch messages simulating an ID in the log.
- // We should not see topic ID errors.
- val zeroTidp = new TopicIdPartition(Uuid.ZERO_UUID, tidp.topicPartition)
- fetchPartitions(
- replicaManager,
- replicaId = 1,
- fetchInfos = Seq(zeroTidp -> validFetchPartitionData),
- responseCallback = callback
- )
- val fetch2 = successfulFetch.headOption.filter(_._1 ==
zeroTidp).map(_._2)
- assertTrue(fetch2.isDefined)
- assertEquals(Errors.NONE, fetch2.get.error)
-
- // Next create a topic without a topic ID written in the log.
- val tp2 = new TopicPartition("noIdTopic", 0)
- val tidp2 = new TopicIdPartition(Uuid.randomUuid(), tp2)
-
- // Broker 0 becomes leader of the partition
- val leaderAndIsrPartitionState2 = new
LeaderAndIsrRequest.PartitionState()
- .setTopicName("noIdTopic")
- .setPartitionIndex(0)
- .setControllerEpoch(0)
- .setLeader(0)
- .setLeaderEpoch(leaderEpoch)
- .setIsr(replicas)
- .setPartitionEpoch(0)
- .setReplicas(replicas)
- .setIsNew(true)
- val leaderAndIsrRequest2 = new LeaderAndIsrRequest.Builder(0, 0,
brokerEpoch,
- Seq(leaderAndIsrPartitionState2).asJava,
- Collections.emptyMap(),
- Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
- val leaderAndIsrResponse2 = replicaManager.becomeLeaderOrFollower(0,
leaderAndIsrRequest2, (_, _) => ())
- assertEquals(Errors.NONE, leaderAndIsrResponse2.error)
-
- assertEquals(None, replicaManager.getPartitionOrException(tp2).topicId)
-
- // Fetch messages simulating the request containing a topic ID. We
should not have an error.
Review Comment:
I think this test case has value, as the fetch request could have no topic ID
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -520,29 +520,18 @@ class ReplicaManagerTest {
}
private[this] def testFencedErrorCausedByBecomeLeader(loopEpochChange: Int):
Unit = {
Review Comment:
Could you please use `@ValueSource` instead of `@Test`?
--
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
