jolshan commented on a change in pull request #11170: URL: https://github.com/apache/kafka/pull/11170#discussion_r707742261
########## File path: core/src/main/scala/kafka/server/ReplicaManager.scala ########## @@ -1755,6 +1765,78 @@ class ReplicaManager(val config: KafkaConfig, partitionsToMakeFollower } + private def updateTopicIdForFollowers(controllerId: Int, + controllerEpoch: Int, + partitionStates: Map[Partition, LeaderAndIsrPartitionState], + correlationId: Int, + responseMap: mutable.Map[TopicPartition, Errors], + topicIds: String => Option[Uuid]) : Set[Partition] = { + val traceLoggingEnabled = stateChangeLogger.isTraceEnabled + // Do we need this? + partitionStates.forKeyValue { (partition, partitionState) => + if (traceLoggingEnabled) + stateChangeLogger.trace(s"Handling LeaderAndIsr request correlationId $correlationId from controller $controllerId " + + s"epoch $controllerEpoch starting the become-follower transition for partition ${partition.topicPartition} with leader " + + s"${partitionState.leader}") + responseMap.put(partition.topicPartition, Errors.NONE) + } + + val partitionsToUpdateFollower: mutable.Set[Partition] = mutable.Set() + try { + partitionStates.forKeyValue { (partition, partitionState) => + val newLeaderBrokerId = partitionState.leader + if (metadataCache.hasAliveBroker(newLeaderBrokerId)) { + // Only change partition state when the leader is available + partitionsToUpdateFollower += partition + } else { + // The leader broker should always be present in the metadata cache. + // If not, we should record the error message and abort the transition process for this partition + stateChangeLogger.error(s"Received LeaderAndIsrRequest with correlation id $correlationId from " + + s"controller $controllerId epoch $controllerEpoch for partition ${partition.topicPartition} " + + s"(last update controller epoch ${partitionState.controllerEpoch}) " + + s"but cannot become follower since the new leader $newLeaderBrokerId is unavailable.") + } + } + + if (isShuttingDown.get()) { + if (traceLoggingEnabled) { + partitionsToUpdateFollower.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} with leader ${partitionStates(partition).leader} " + + "since it is shutting down") + } + } + } else { + val partitionsToMakeFollowerWithLeaderAndOffset = partitionsToUpdateFollower.map { partition => + val leaderNode = partition.leaderReplicaIdOpt.flatMap(leaderId => metadataCache. + getAliveBrokerNode(leaderId, config.interBrokerListenerName)).getOrElse(Node.noNode()) + val leader = new BrokerEndPoint(leaderNode.id(), leaderNode.host(), leaderNode.port()) + val log = partition.localLogOrException + val fetchOffset = initialFetchOffset(log) + partition.topicPartition -> InitialFetchState(topicIds(partition.topic), leader, partition.getLeaderEpoch, fetchOffset) + }.toMap + + replicaFetcherManager.addFetcherForPartitions(partitionsToMakeFollowerWithLeaderAndOffset) Review comment: There are a few ways to do this and I'll be interested to see which is the best. It is a little tricky to go from topic partition -> fetcher -> fetch state where we can update the topic ID. One way is to use `find` like `fetchState` does. Another is to look up the fetcher in the map like `addFetcherForPartitions` does. Finally, we could try to just add topic IDs to all the partitions in the fetcher by iterating through, but that might not be very efficient -- 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