private NewTopic createRemoteLogMetadataTopicRequest() {
        Map<String, String> topicConfigs = new HashMap<>();
        topicConfigs.put(TopicConfig.RETENTION_MS_CONFIG, Long.toString(rlmmConfig.metadataTopicRetentionMs()));
        topicConfigs.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE);
        topicConfigs.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "false");
        return new NewTopic(rlmmConfig.remoteLogMetadataTopicName(),
                            rlmmConfig.metadataTopicPartitionsCount(),
                            rlmmConfig.metadataTopicReplicationFactor()).configs(topicConfigs);
    }
Returns {@link CompletableFuture} which will complete only after publishing of the given {@code remoteLogMetadata} into
the remote log metadata topic and the internal consumer is caught up until the produced record's offset.

@param topicIdPartition partition of the given remoteLogMetadata.
@param remoteLogMetadata RemoteLogMetadata to be stored.
@return a future with acknowledge and potentially waiting also for consumer to catch up.
This ensures cache is synchronized with backing topic.
@throws RemoteStorageException if there are any storage errors occur.
storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManager.java