hachikuji commented on a change in pull request #10252: URL: https://github.com/apache/kafka/pull/10252#discussion_r586724166
########## File path: core/src/main/scala/kafka/server/metadata/MetadataPartitions.scala ########## @@ -172,23 +203,36 @@ class MetadataPartitionsBuilder(val brokerId: Int, if (partition.isReplicaFor(brokerId)) { _localChanged.add(partition) } else if (prevPartition != null && prevPartition.isReplicaFor(brokerId)) { - _localRemoved.add(prevPartition) + maybeAddToLocalRemoved(prevPartition) } newNameMap.put(partition.topicName, newPartitionMap) } + private def maybeAddToLocalRemoved(partition: MetadataPartition): Unit= { + val currentTopicId = newReverseIdMap.get(partition.topicName) + val prevImageContainsTopic = if (currentTopicId != null) { + prevPartitions.topicIdToName(currentTopicId).isDefined + } else { + prevPartitions.allTopicNames().contains(partition.topicName) + } + + if (prevImageContainsTopic) { + _localRemoved.add(partition) + } + } + def remove(topicName: String, partitionId: Int): Unit = { val prevPartitionMap = newNameMap.get(topicName) if (prevPartitionMap != null) { if (changed.contains(prevPartitionMap)) { val prevPartition = prevPartitionMap.remove(partitionId) if (prevPartition.isReplicaFor(brokerId)) { Review comment: Yeah, good point. Looking at how it is used in `RaftMetadataCache.updateMetadata`, I don't see how we can assume the partition previously existed. There are really no ordering guarantees that we can get for that case. I find it pretty confusing that we try to let this class seamlessly handle metadata updates from both the zk and raft controllers. I would really rather get rid of `RaftMetadataCache.updateMetadata` altogether. ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org