jolshan commented on a change in pull request #11170:
URL: https://github.com/apache/kafka/pull/11170#discussion_r709357795
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -1755,6 +1762,66 @@ class ReplicaManager(val config: KafkaConfig,
partitionsToMakeFollower
}
+ private def updateTopicIdForFollowers(controllerId: Int,
+ controllerEpoch: Int,
+ partitionStates: Map[Partition,
LeaderAndIsrPartitionState],
+ correlationId: Int,
+ topicIds: String => Option[Uuid]) :
Set[Partition] = {
+ val traceLoggingEnabled = stateChangeLogger.isTraceEnabled
+
+ val partitionsToUpdateFollower = mutable.Set.empty[Partition]
+ 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 partitionsToUpdateFollowerWithLeader =
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())
+ (partition.topicPartition, BrokerAndFetcherId(leader,
replicaFetcherManager.getFetcherId(partition.topicPartition)))
+ }
+
replicaFetcherManager.addTopicIdsToFetcherThread(partitionsToUpdateFollowerWithLeader,
topicIds)
Review comment:
Do you think we also don't need the check for the leader in the metadata
cache? ie:
```
val newLeaderBrokerId = partitionState.leader
if (metadataCache.hasAliveBroker(newLeaderBrokerId)) {
// Only change partition state when the leader is available
partitionsToUpdateFollower += partition
```
If we do keep it, I can change the comment to be less confusing.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
