dajac commented on a change in pull request #11170: URL: https://github.com/apache/kafka/pull/11170#discussion_r710837733
########## File path: core/src/main/scala/kafka/server/AbstractFetcherManager.scala ########## @@ -163,6 +163,16 @@ abstract class AbstractFetcherManager[T <: AbstractFetcherThread](val name: Stri info(s"Added fetcher to broker ${fetcherThread.sourceBroker.id} for partitions $initialOffsetAndEpochs") } + def addTopicIdsToFetcherThread(partitionsToUpdate: Set[(TopicPartition, Int)], topicIds: String => Option[Uuid]): Unit = { Review comment: 1) Should we called this one `maybeUpdateTopicIds`? Using `add` feels a bit awkward for the topic id field. Also, the `ToFetcherThread` is not necessary in my opinion because we already know that the fetcher manager is all about managing fetcher threads. 2) Could we use a `Map` instead of `Set`? `Map` feels a bit more natural for a key-value pair mapping. 3) Could we add a small scaladoc to explain the parameters? ########## File path: core/src/test/scala/unit/kafka/server/AbstractFetcherManagerTest.scala ########## @@ -20,7 +20,7 @@ import com.yammer.metrics.core.Gauge import kafka.cluster.BrokerEndPoint import kafka.metrics.KafkaYammerMetrics import kafka.utils.TestUtils -import org.apache.kafka.common.TopicPartition +import org.apache.kafka.common.{TopicPartition, Uuid} Review comment: Should we add few unit tests for the newly added methods in this suite and in `AbstractFetcherThreadTest` ? ########## File path: core/src/main/scala/kafka/server/AbstractFetcherManager.scala ########## @@ -163,6 +163,16 @@ abstract class AbstractFetcherManager[T <: AbstractFetcherThread](val name: Stri info(s"Added fetcher to broker ${fetcherThread.sourceBroker.id} for partitions $initialOffsetAndEpochs") } + def addTopicIdsToFetcherThread(partitionsToUpdate: Set[(TopicPartition, Int)], topicIds: String => Option[Uuid]): Unit = { + lock synchronized { + val partitionsPerFetcher = partitionsToUpdate.groupMap(partitionsToUpdate => BrokerIdAndFetcherId(partitionsToUpdate._2, getFetcherId(partitionsToUpdate._1)))(_._1) Review comment: We usually avoid using `._1` or `._2` and would rather prefer to deconstruct the tuple in order to name the argument: `groupMap { case (tp, leader) => ....`. Does `groupMap` work with all the versions of Scala that we use? ########## File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala ########## @@ -491,6 +491,21 @@ abstract class AbstractFetcherThread(name: String, } finally partitionMapLock.unlock() } + def maybeAddTopicIdsToThread(partitions: Set[TopicPartition], topicIds: String => Option[Uuid]) = { Review comment: If we rename `addTopicIdsToFetcherThread`, we should also rename this one to follow the same naming. ########## File path: core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala ########## @@ -281,14 +281,14 @@ class ReplicaAlterLogDirsThread(name: String, val fetchRequestOpt = if (requestMap.isEmpty) { None } else { - val version: Short = if (topicIds.containsKey(tp.topic())) + val version: Short = if (fetchState.topicId.isEmpty) 12 else ApiKeys.FETCH.latestVersion // Set maxWait and minBytes to 0 because the response should return immediately if // the future log has caught up with the current log of the partition val requestBuilder = FetchRequest.Builder.forReplica(version, replicaId, 0, 0, requestMap, - topicIds).setMaxBytes(maxBytes) + Collections.singletonMap(tp.topic(), fetchState.topicId.getOrElse(Uuid.ZERO_UUID))).setMaxBytes(maxBytes) Review comment: nit: `topic()` -> `topic`. ########## File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala ########## @@ -491,6 +491,21 @@ abstract class AbstractFetcherThread(name: String, } finally partitionMapLock.unlock() } + def maybeAddTopicIdsToThread(partitions: Set[TopicPartition], topicIds: String => Option[Uuid]) = { + partitionMapLock.lockInterruptibly() + try { + partitions.foreach { tp => + val currentState = partitionStates.stateValue(tp) + if (currentState != null) { + val updatedState = currentState.updateTopicId(topicIds(tp.topic)) + partitionStates.updateAndMoveToEnd(tp, updatedState) Review comment: I think that we should use `update` here instead of `updateAndMoveToEnd`. `updateAndMoveToEnd` will basically deprioritize the partition and I don't think that we want to do this when we only update its topic id. ########## File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala ########## @@ -2810,4 +2811,132 @@ class ReplicaManagerTest { Replicas.NONE, Replicas.NONE, 2, 123, 456)))), replicaManager.calculateDeltaChanges(TEST_DELTA)) } + + @Test + def testPartitionFetchStateUpdatesWithTopicIdAdded(): Unit = { + val aliveBrokersIds = Seq(0, 1) + val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time), + brokerId = 0, aliveBrokersIds) + try { + val tp = new TopicPartition(topic, 0) + val replicas = aliveBrokersIds.toList.map(Int.box).asJava + + def leaderAndIsrRequest(epoch: Int, topicIds: util.Map[String, Uuid]): LeaderAndIsrRequest = Review comment: You use this method in the two tests. Should we pull it out into an helper method? Also, it might be worth checking if we already have such helper in the class. ########## File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala ########## @@ -2810,4 +2811,132 @@ class ReplicaManagerTest { Replicas.NONE, Replicas.NONE, 2, 123, 456)))), replicaManager.calculateDeltaChanges(TEST_DELTA)) } + + @Test + def testPartitionFetchStateUpdatesWithTopicIdAdded(): Unit = { + val aliveBrokersIds = Seq(0, 1) + val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time), + brokerId = 0, aliveBrokersIds) + try { + val tp = new TopicPartition(topic, 0) + val replicas = aliveBrokersIds.toList.map(Int.box).asJava + + def leaderAndIsrRequest(epoch: Int, topicIds: util.Map[String, Uuid]): LeaderAndIsrRequest = + new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, + Seq(new LeaderAndIsrPartitionState() + .setTopicName(topic) + .setPartitionIndex(0) + .setControllerEpoch(0) + .setLeader(1) + .setLeaderEpoch(epoch) + .setIsr(replicas) + .setZkVersion(0) + .setReplicas(replicas) + .setIsNew(true)).asJava, + topicIds, + Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build() + + val leaderAndIsrRequest1 = leaderAndIsrRequest(0, Collections.emptyMap()) + val leaderAndIsrResponse = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest1, (_, _) => ()) Review comment: nit: As `leaderAndIsrRequest1` is only used one, I would be tempted to inline it directly. You can break the line if it becomes too long. ########## File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala ########## @@ -2810,4 +2811,132 @@ class ReplicaManagerTest { Replicas.NONE, Replicas.NONE, 2, 123, 456)))), replicaManager.calculateDeltaChanges(TEST_DELTA)) } + + @Test + def testPartitionFetchStateUpdatesWithTopicIdAdded(): Unit = { + val aliveBrokersIds = Seq(0, 1) + val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time), + brokerId = 0, aliveBrokersIds) + try { + val tp = new TopicPartition(topic, 0) + val replicas = aliveBrokersIds.toList.map(Int.box).asJava + + def leaderAndIsrRequest(epoch: Int, topicIds: util.Map[String, Uuid]): LeaderAndIsrRequest = + new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, + Seq(new LeaderAndIsrPartitionState() + .setTopicName(topic) + .setPartitionIndex(0) + .setControllerEpoch(0) + .setLeader(1) + .setLeaderEpoch(epoch) + .setIsr(replicas) + .setZkVersion(0) + .setReplicas(replicas) + .setIsNew(true)).asJava, + topicIds, + Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build() + + val leaderAndIsrRequest1 = leaderAndIsrRequest(0, Collections.emptyMap()) + val leaderAndIsrResponse = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest1, (_, _) => ()) + assertEquals(Errors.NONE, leaderAndIsrResponse.error) + + val fetchState = replicaManager.replicaFetcherManager.getFetcher(tp).flatMap(fetcher => fetcher.fetchState(tp)) + assertTrue(fetchState.isDefined) + assertEquals(None, fetchState.get.topicId) Review comment: nit: This assertion is used multiple times in the newly added tests. How about extracting it into a small helper method? e.g. `assertFetcherHasTopicId(manager, tp, expectedTopicId)` or something like this. ########## File path: core/src/main/scala/kafka/server/ReplicaManager.scala ########## @@ -1755,6 +1762,40 @@ class ReplicaManager(val config: KafkaConfig, partitionsToMakeFollower } + private def updateTopicIdForFollowers(controllerId: Int, + controllerEpoch: Int, + partitionStates: Set[Partition], + correlationId: Int, + topicIds: String => Option[Uuid]): Unit = { + val traceLoggingEnabled = stateChangeLogger.isTraceEnabled + + try { + if (isShuttingDown.get()) { + if (traceLoggingEnabled) { + partitionStates.foreach { partition => + stateChangeLogger.trace(s"Skipped the update topic ID step of the become-follower state " + + s"change with correlation id $correlationId from controller $controllerId epoch $controllerEpoch for " + + s"partition ${partition.topicPartition} since it is shutting down") + } + } + } else { + val partitionsToUpdateFollowerWithLeader = partitionStates.map { partition => + val leaderNode = partition.leaderReplicaIdOpt.flatMap(leaderId => metadataCache. + getAliveBrokerNode(leaderId, config.interBrokerListenerName)).getOrElse(Node.noNode()) + val leaderId = leaderNode.id Review comment: So, you take the leader id from the partition, then you lookup the node from the metadata cache, finally you get the leader id from the node. I guess you could use the leader id directly. :) Initially, I thought that you would put the `metadataCache.hasAliveBroker(partitionState.leader)` check here instead of doing it at L1407. It does not matter much but doing it here would be a bit more aligned with the logic in the `makeFollowers` method. ########## File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala ########## @@ -491,6 +491,21 @@ abstract class AbstractFetcherThread(name: String, } finally partitionMapLock.unlock() } + def maybeAddTopicIdsToThread(partitions: Set[TopicPartition], topicIds: String => Option[Uuid]) = { + partitionMapLock.lockInterruptibly() + try { + partitions.foreach { tp => + val currentState = partitionStates.stateValue(tp) + if (currentState != null) { + val updatedState = currentState.updateTopicId(topicIds(tp.topic)) + partitionStates.updateAndMoveToEnd(tp, updatedState) + } + } + Review comment: nit: We could remove this empty line. ########## File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala ########## @@ -2810,4 +2811,132 @@ class ReplicaManagerTest { Replicas.NONE, Replicas.NONE, 2, 123, 456)))), replicaManager.calculateDeltaChanges(TEST_DELTA)) } + + @Test + def testPartitionFetchStateUpdatesWithTopicIdAdded(): Unit = { + val aliveBrokersIds = Seq(0, 1) + val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time), + brokerId = 0, aliveBrokersIds) + try { + val tp = new TopicPartition(topic, 0) + val replicas = aliveBrokersIds.toList.map(Int.box).asJava + + def leaderAndIsrRequest(epoch: Int, topicIds: util.Map[String, Uuid]): LeaderAndIsrRequest = + new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, + Seq(new LeaderAndIsrPartitionState() + .setTopicName(topic) + .setPartitionIndex(0) + .setControllerEpoch(0) + .setLeader(1) + .setLeaderEpoch(epoch) + .setIsr(replicas) + .setZkVersion(0) + .setReplicas(replicas) + .setIsNew(true)).asJava, + topicIds, + Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build() + + val leaderAndIsrRequest1 = leaderAndIsrRequest(0, Collections.emptyMap()) + val leaderAndIsrResponse = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest1, (_, _) => ()) + assertEquals(Errors.NONE, leaderAndIsrResponse.error) + + val fetchState = replicaManager.replicaFetcherManager.getFetcher(tp).flatMap(fetcher => fetcher.fetchState(tp)) + assertTrue(fetchState.isDefined) + assertEquals(None, fetchState.get.topicId) + + val leaderAndIsrRequest2 = leaderAndIsrRequest(0, topicIds.asJava) + val leaderAndIsrResponse2 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest2, (_, _) => ()) + assertEquals(Errors.NONE, leaderAndIsrResponse2.error) + + val fetchState2 = replicaManager.replicaFetcherManager.getFetcher(tp).flatMap(fetcher => fetcher.fetchState(tp)) + assertTrue(fetchState2.isDefined) + assertEquals(Some(topicId), fetchState2.get.topicId) + + } finally { + replicaManager.shutdown(checkpointHW = false) + } + } + + @Test + def testReplicaAlterLogDirsWithAndWithoutIds(): Unit = { Review comment: Does this test work? I have the impression that we never update the `ReplicaAlterLogDirsManager`. ########## File path: core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala ########## @@ -76,7 +76,8 @@ class ReplicaAlterLogDirsThread(name: String, var partitionData: Seq[(TopicPartition, FetchData)] = null val request = fetchRequest.build() - val (topicIds, topicNames) = replicaMgr.metadataCache.topicIdInfo() + val topicIds = request.data().topics().asScala.map(topic => (topic.topic, topic.topicId)).toMap + val topicNames = topicIds.map(_.swap) Review comment: 1) Is it fine to rely on the request to construct the topic to topicId mapping? I suppose that it is OK here because the request was populated with both, right? If this is correct, is it worth adding a small comment here? 2) nit: We could omit few parenthesis here (e.g. `data`, `topics`). 3) Is it worth using two mutable Maps which would be populated while iterating over the `topics`? -- 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