KAFKA-19122: updateClusterMetadata receives multiples PartitionInfo

[brandboat]

This patch resolves the following issues in MetadataCache#toCluster:

• Avoids duplicate Node entries when a broker has multiple endpoints.
• Fixes a bug where fenced brokers result in NPE.
• Ensures missing topic IDs are properly populated in cluster metadata.
• Delete unused test code snippet.

[m1a2st]

Thanks @brandboat for this patch, left a question.

[m1a2st]

Why shouldn’t we filter out the fenced broker here?

[brandboat]

Fixes a bug where fenced brokers result in NPE.

Like I mentioned in the PR description, this result in a NullPointerException.
Filtering out the fenced broker while Partition itself still includes that fenced broker in replicas can trigger the NPE as we already filter them out...

[brandboat]

I'm not quite sure why we filtered out fenced brokers in the first place—was there a reason for doing so?

[m1a2st]

I'm not sure, but I'm a bit confused — shouldn't the partition leader broker not be a fenced broker? In KIP-841 has following invariants

• a fenced or in-controlled-shutdown replica is not eligible to be in the ISR; and
• a fenced or in-controlled-shutdown replica is not eligible to become leader.

Maybe I misunderstanding something.

[brandboat]

Pardon me, I meant "replicas", not leader.

Maybe an example will help clarify things:
Let’s say we have 3 brokers — broker0, broker1, and broker2 — and a topic called my-topic. Partition 0 of this topic has 3 replicas, one on each broker. Now suppose broker2 unexpectedly shuts down.

Here’s what happens next:

1. DynamicTopicClusterQuotaPublisher#onMetadataUpdate gets triggered
2. That leads to MetadataCache#toCluster being called
3. In toCluster, nodesById gets constructed without the fenced broker (broker2 is filtered out)
4. Then MetadataCache#toArray uses this nodesById, but in the TopicsImage the PartitionRegistration (my-topic partition-0) still has replicas [0, 1, 2]. Since broker2 isn’t in nodesById, we get a NullPointerException, kaboom!