Yunyung commented on code in PR #19966:
URL: https://github.com/apache/kafka/pull/19966#discussion_r2147754923
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -1707,20 +1521,10 @@ class ReplicaManagerTest {
// Make this replica the leader
val leaderEpoch = 1
- val leaderAndIsrRequest = new LeaderAndIsrRequest.Builder(0, 0,
brokerEpoch,
- Seq(new LeaderAndIsrRequest.PartitionState()
- .setTopicName(topic)
- .setPartitionIndex(0)
- .setControllerEpoch(0)
- .setLeader(0)
- .setLeaderEpoch(leaderEpoch)
- .setIsr(brokerList)
- .setPartitionEpoch(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 localId = 0
Review Comment:
Nit: ditto
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -3164,9 +2924,9 @@ class ReplicaManagerTest {
fetchInfos: Seq[(TopicIdPartition, PartitionData)],
responseCallback: Seq[(TopicIdPartition, FetchPartitionData)] => Unit,
requestVersion: Short = ApiKeys.FETCH.latestVersion,
- maxWaitMs: Long = 0,
- minBytes: Int = 1,
- maxBytes: Int = 1024 * 1024,
+ maxWaitMs: Long,
+ minBytes: Int,
+ maxBytes: Int,
Review Comment:
Why make this change?
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -970,29 +964,19 @@ class ReplicaManagerTest {
@Test
def testFetchBeyondHighWatermark(): Unit = {
+ val localId = 0
val rm = setupReplicaManagerWithMockedPurgatories(new MockTimer(time),
aliveBrokerIds = Seq(0, 1, 2))
try {
- val brokerList = Seq[Integer](0, 1, 2).asJava
val partition = rm.createPartition(new TopicPartition(topic, 0))
partition.createLogIfNotExists(isNew = false, isFutureReplica = false,
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 = topicsCreateDelta(localId, isStartIdLeader = true,
topicName = topic, topicId = topicIds(topic))
Review Comment:
This does not align with the original behavior (3 brokers).
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -1038,25 +1022,12 @@ class ReplicaManagerTest {
try {
val tp = new TopicPartition(topic, 0)
val tidp = new TopicIdPartition(topicId, tp)
- val replicas = aliveBrokersIds.toList.map(Int.box).asJava
// Broker 0 becomes leader of the partition
- val leaderAndIsrPartitionState = new LeaderAndIsrRequest.PartitionState()
- .setTopicName(topic)
- .setPartitionIndex(0)
- .setControllerEpoch(0)
- .setLeader(0)
- .setLeaderEpoch(leaderEpoch)
- .setIsr(replicas)
- .setPartitionEpoch(0)
- .setReplicas(replicas)
- .setIsNew(true)
- val leaderAndIsrRequest = new LeaderAndIsrRequest.Builder(0, 0,
brokerEpoch,
- Seq(leaderAndIsrPartitionState).asJava,
- Collections.singletonMap(topic, topicId),
- Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
- val leaderAndIsrResponse = replicaManager.becomeLeaderOrFollower(0,
leaderAndIsrRequest, (_, _) => ())
- assertEquals(Errors.NONE, leaderAndIsrResponse.error)
+ val localId = 0
Review Comment:
Nit: Move this to the beginning of the method for consistent code style.
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -1917,22 +1699,10 @@ class ReplicaManagerTest {
val tidp0 = new TopicIdPartition(topicId, tp0)
val offsetCheckpoints = new
LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints.asJava)
replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false,
isFutureReplica = false, offsetCheckpoints, None)
- val partition0Replicas = Seq[Integer](0, 1).asJava
- val becomeLeaderRequest = new LeaderAndIsrRequest.Builder(0, 0,
brokerEpoch,
- Seq(new LeaderAndIsrRequest.PartitionState()
- .setTopicName(tp0.topic)
- .setPartitionIndex(tp0.partition)
- .setControllerEpoch(0)
- .setLeader(0)
- .setLeaderEpoch(1)
- .setIsr(partition0Replicas)
- .setPartitionEpoch(0)
- .setReplicas(partition0Replicas)
- .setIsNew(true)).asJava,
- topicIds.asJava,
- Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
- replicaManager.becomeLeaderOrFollower(1, becomeLeaderRequest, (_, _) =>
())
+ val leaderDelta = topicsCreateDelta(localId, isStartIdLeader = true,
topicName = topic, topicId = topicIds(topic))
Review Comment:
LeaderEpoch = 1
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -1265,36 +1116,14 @@ class ReplicaManagerTest {
val offsetCheckpoints = new
LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints.asJava)
replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false,
isFutureReplica = false, offsetCheckpoints, None)
replicaManager.createPartition(tp1).createLogIfNotExists(isNew = false,
isFutureReplica = false, offsetCheckpoints, None)
- val partition0Replicas = Seq[Integer](0, 1).asJava
- val partition1Replicas = Seq[Integer](0, 2).asJava
Review Comment:
Keep these so you can reuse them below.
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2833,6 +2592,7 @@ class ReplicaManagerTest {
allLogs.put(topicPartitionObj, mockLog)
when(mockLogMgr.allLogs).thenReturn(allLogs.values.asScala)
when(mockLogMgr.isLogDirOnline(anyString)).thenReturn(true)
+ when(mockLogMgr.directoryId(anyString)).thenReturn(None)
Review Comment:
why make this change?
--
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
