hachikuji commented on a change in pull request #11126: URL: https://github.com/apache/kafka/pull/11126#discussion_r677853531
########## File path: core/src/main/scala/kafka/server/ReplicaManager.scala ########## @@ -1367,12 +1367,25 @@ class ReplicaManager(val config: KafkaConfig, val currentLeaderEpoch = partition.getLeaderEpoch val requestLeaderEpoch = partitionState.leaderEpoch val requestTopicId = topicIdFromRequest(topicPartition.topic) + val logTopicId = partition.topicId - if (!hasConsistentTopicId(requestTopicId, partition.topicId)) { - stateChangeLogger.error(s"Topic ID in memory: ${partition.topicId.get} does not" + + // When running a ZK controller and upgrading to topic IDs we may receive a request with leader epoch + // that is equal to the current leader epoch. In this case, we want to assign topic ID to the log. + def isUpgradingToTopicIdWithExistingLog: Boolean = { + requestLeaderEpoch == currentLeaderEpoch && Review comment: Might be just me, but I find it a little awkward to see the epoch check nested here given the other checks below. Would it be reasonable instead to move this to after the inequality checks? ```scala else if (requestLeaderEpoch < currentLeaderEpoch) { ... } else { val error = requestTopicId match { case Some(topicId) if logTopicId.isEmpty => // The controller may send LeaderAndIsr to update topicId without bumping the epoch log.assignTopicId(topicId) stateChangeLogger.info("Updating topicId for $log to $topicId from LeaderAndIsr request....") Errors.NONE case _ => stateChangeLogger.info(s"Ignoring LeaderAndIsr request from " + s"controller $controllerId with correlation id $correlationId " + s"epoch $controllerEpoch for partition $topicPartition since its associated " + s"leader epoch $requestLeaderEpoch matches the current leader epoch") Errors.STALE_CONTROLLER_EPOCH } responseMap.put(topicPartition, error) } ``` -- 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