From 72a267c73643270c6ede5e72def6afb32ccfee67 Mon Sep 17 00:00:00 2001
From: Luke Chen <showuon@gmail.com>
Date: Wed, 3 Sep 2025 17:34:46 +0900
Subject: [PATCH] fix 2 issues

---
 .../main/java/kafka/server/RemoteClusterMetadataManager.java | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/core/src/main/java/kafka/server/RemoteClusterMetadataManager.java b/core/src/main/java/kafka/server/RemoteClusterMetadataManager.java
index 25f0f31a761ee..d22f02a270640 100644
--- a/core/src/main/java/kafka/server/RemoteClusterMetadataManager.java
+++ b/core/src/main/java/kafka/server/RemoteClusterMetadataManager.java
@@ -140,7 +140,7 @@ private void handleReadOnlyLeadersChanges(Map<TopicPartition, LocalReplicaChange
                         metrics,
                         time,
                         brokerEndpoint.id(),
-                        "broker-" + nodeId + "-remote-cluster-metadata-manager-" + k,
+                        "broker-" + nodeId + "-remote-cluster-metadata-manager-" + k.replace(":", "-"),
                         logContext
                     );
                 });
@@ -192,7 +192,8 @@ public void refreshRemoteMetadata() {
                         partitionLeaders.put(partitionMetadata.topicPartition, remoteClusterNodes.get(remoteBootstrapServers).get(partitionMetadata.leaderId.get()));
                     });
 
-                    if (metadataImage.topics().getTopic(topicMetadata.topicId()).partitions().size() < partitionLeaders.size()) {
+                    if (metadataImage.topics().getTopic(topicMetadata.topicId()) != null &&
+                            metadataImage.topics().getTopic(topicMetadata.topicId()).partitions().size() < partitionLeaders.size()) {
                         createPartitionsTopics.add(new CreatePartitionsRequestData.CreatePartitionsTopic()
                             .setName(topicMetadata.topic())
                             .setCount(partitionLeaders.size())