public CompletableFuture<RecordMetadata> publishMessage(RemoteLogMetadata remoteLogMetadata) {
        CompletableFuture<RecordMetadata> future = new CompletableFuture<>();

        TopicIdPartition topicIdPartition = remoteLogMetadata.topicIdPartition();
        int metadataPartitionNum = topicPartitioner.metadataPartition(topicIdPartition);
        log.debug("Publishing metadata message of partition:[{}] into metadata topic partition:[{}] with payload: [{}]",
                  topicIdPartition, metadataPartitionNum, remoteLogMetadata);
        if (metadataPartitionNum >= rlmmConfig.metadataTopicPartitionsCount()) {
            // This should never occur as long as metadata partitions always remain the same.
            throw new KafkaException("Chosen partition no " + metadataPartitionNum +
                                             " must be less than the partition count: " + rlmmConfig.metadataTopicPartitionsCount());
        }

        try {
            Callback callback = (metadata, exception) -> {
                if (exception != null) {
                    future.completeExceptionally(exception);
                } else {
                    future.complete(metadata);
                }
            };
            producer.send(new ProducerRecord<>(rlmmConfig.remoteLogMetadataTopicName(), metadataPartitionNum, null,
                                               serde.serialize(remoteLogMetadata)), callback);
        } catch (Exception ex) {
            future.completeExceptionally(ex);
        }

        return future;
    }
Returns {@link CompletableFuture} which will complete only after publishing of the given {@code remoteLogMetadata}
is considered complete.

@param remoteLogMetadata RemoteLogMetadata to be published
@return
storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ProducerManager.java