private void maybeFetchStartAndEndOffsets() {
        // If the leader for a `__remote_log_metadata` partition is not available, then the call to `ListOffsets`
        // will fail after the default timeout of 1 min. Added a delay between the retries to prevent the thread from
        // aggressively fetching the list offsets. During this time, the recently reassigned user-topic-partitions
        // won't be marked as initialized.
        if (hasLastOffsetsFetchFailed && lastFailedFetchOffsetsTimestamp + offsetFetchRetryIntervalMs < time.milliseconds()) {
            fetchStartAndEndOffsets();
        }
    }
This class is responsible for consuming messages from remote log metadata topic ({@link TopicBasedRemoteLogMetadataManagerConfig#REMOTE_LOG_METADATA_TOPIC_NAME})
partitions and maintain the state of the remote log segment metadata. It gives an API to add or remove
for what topic partition's metadata should be consumed by this instance using
{{@link #addAssignmentsForPartitions(Set)}} and {@link #removeAssignmentsForPartitions(Set)} respectively.
<p>
When a broker is started, controller sends topic partitions that this broker is leader or follower for and the
partitions to be deleted. This class receives those notifications with
{@link #addAssignmentsForPartitions(Set)} and {@link #removeAssignmentsForPartitions(Set)} assigns consumer for the
respective remote log metadata partitions by using {@link RemoteLogMetadataTopicPartitioner#metadataPartition(TopicIdPartition)}.
Any leadership changes later are called through the same API. We will remove the partitions that are deleted from
this broker which are received through {@link #removeAssignmentsForPartitions(Set)}.
<p>
After receiving these events it invokes {@link RemotePartitionMetadataEventHandler#handleRemoteLogSegmentMetadata(RemoteLogSegmentMetadata)},
which maintains in-memory representation of the state of {@link RemoteLogSegmentMetadata}.
storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java