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
