@Override
    public void configure(Map<String, ?> configs) {
        Objects.requireNonNull(configs, "configs can not be null.");

        lock.writeLock().lock();
        try {
            if (configured) {
                log.info("Skipping configure as it is already configured.");
                return;
            }

            log.info("Started configuring topic-based RLMM with configs: {}", configs);

            rlmmConfig = new TopicBasedRemoteLogMetadataManagerConfig(configs);
            rlmTopicPartitioner = remoteLogMetadataTopicPartitionerFunction.apply(rlmmConfig.metadataTopicPartitionsCount());
            remotePartitionMetadataStore = remoteLogMetadataManagerSupplier.get();
            configured = true;
            log.info("Successfully configured topic-based RLMM with config: {}", rlmmConfig);

            // Scheduling the initialization producer/consumer managers in a separate thread. Required resources may
            // not yet be available now. This thread makes sure that it is retried at regular intervals until it is
            // successful.
            initializationThread = KafkaThread.nonDaemon("RLMMInitializationThread", this::initializeResources);
            initializationThread.start();
        } finally {
            lock.writeLock().unlock();
        }
    }
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