jolshan commented on a change in pull request #11126:
URL: https://github.com/apache/kafka/pull/11126#discussion_r677854946
##########
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:
Ah yeah. That would work. I felt it was a bit odd too, but didn't think
of this alternative. This placement is better.
--
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
