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
