hachikuji commented on a change in pull request #11171:
URL: https://github.com/apache/kafka/pull/11171#discussion_r683680618
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -553,6 +553,17 @@ class Log(@volatile var logStartOffset: Long,
/** Only used for ZK clusters when we update and start using topic IDs on
existing topics */
def assignTopicId(topicId: Uuid): Unit = {
+ // defensively check that any newly assign topic ID matches any that is
already set
+ _topicId.foreach { current =>
+ if (!current.equals(topicId)) {
+ // we should never get here as the topic IDs should have been checked
in becomeLeaderOrFollower
Review comment:
nit: I think it's ok to leave this out. The point of adding the check is
to reduce the coupling with ReplicaManager
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -553,6 +553,17 @@ class Log(@volatile var logStartOffset: Long,
/** Only used for ZK clusters when we update and start using topic IDs on
existing topics */
def assignTopicId(topicId: Uuid): Unit = {
+ // defensively check that any newly assign topic ID matches any that is
already set
+ _topicId.foreach { current =>
+ if (!current.equals(topicId)) {
+ // we should never get here as the topic IDs should have been checked
in becomeLeaderOrFollower
+ throw new InconsistentTopicIdException(s"Tried to assign topic ID
$topicId to log for topic partition $topicPartition," +
+ s"but log already contained topic ID $current")
+ }
+ // Topic ID already assigned so we can return
+ return
Review comment:
I was thinking how we could avoid this return. How about something like
this:
```scala
_topicId match {
case Some(currentId) =>
if (!currentId.equals(topicId)) {
throw new InconsistentTopicIdException(s"Tried to assign topic ID
$topicId to log for topic partition $topicPartition," +
s"but log already contained topic ID $current")
}
case None =>
if (keepPartitionMetadataFile) {
_topicId = Some(topicId)
if (!partitionMetadataFile.exists()) {
partitionMetadataFile.record(topicId)
scheduler.schedule("flush-metadata-file", maybeFlushMetadataFile)
}
}
}
```
--
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
