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())