private boolean isPartitionsCountSameAsConfigured(Admin adminClient,
                                                      String topicName) throws InterruptedException, ExecutionException {
        log.debug("Getting topic details to check for partition count and replication factor.");
        TopicDescription topicDescription = adminClient.describeTopics(Collections.singleton(topicName))
                                                       .topicNameValues().get(topicName).get();
        int expectedPartitions = rlmmConfig.metadataTopicPartitionsCount();
        int topicPartitionsSize = topicDescription.partitions().size();

        if (topicPartitionsSize != expectedPartitions) {
            log.error("Existing topic partition count [{}] is not same as the expected partition count [{}]",
                      topicPartitionsSize, expectedPartitions);
            return false;
        }

        return true;
    }
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