@Override
    public long remoteLogSize(TopicIdPartition topicIdPartition, int leaderEpoch) throws RemoteStorageException {
        long remoteLogSize = 0L;
        // This is a simple-to-understand but not the most optimal solution.
        // The TopicBasedRemoteLogMetadataManager's remote metadata store is file-based. During design discussions
        // at https://lists.apache.org/thread/kxd6fffq02thbpd0p5y4mfbs062g7jr6
        // we reached a consensus that sequential iteration over files on the local file system is performant enough.
        // Should this stop being the case, the remote log size could be calculated by incrementing/decrementing
        // counters during API calls for a more performant implementation.
        Iterator<RemoteLogSegmentMetadata> remoteLogSegmentMetadataIterator = remotePartitionMetadataStore.listRemoteLogSegments(topicIdPartition, leaderEpoch);
        while (remoteLogSegmentMetadataIterator.hasNext()) {
            RemoteLogSegmentMetadata remoteLogSegmentMetadata = remoteLogSegmentMetadataIterator.next();
            remoteLogSize += remoteLogSegmentMetadata.segmentSizeInBytes();
        }
        return remoteLogSize;
    }
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