void maybeWaitForPartitionAssignments() throws InterruptedException {
        // Snapshots of the metadata-partition and user-topic-partition are used to reduce the scope of the
        // synchronization block.
        // 1) LEADER_AND_ISR and STOP_REPLICA requests adds / removes the user-topic-partitions from the request
        //    handler threads. Those threads should not be blocked for a long time, therefore scope of the
        //    synchronization block is reduced to bare minimum.
        // 2) Note that the consumer#position, consumer#seekToBeginning, consumer#seekToEnd and the other consumer APIs
        //    response times are un-predictable. Those should not be kept in the synchronization block.
        final Set<Integer> metadataPartitionSnapshot = new HashSet<>();
        final Set<UserTopicIdPartition> assignedUserTopicIdPartitionsSnapshot = new HashSet<>();
        synchronized (assignPartitionsLock) {
            while (!isClosed && assignedUserTopicIdPartitions.isEmpty()) {
                log.debug("Waiting for remote log metadata partitions to be assigned");
                assignPartitionsLock.wait();
            }
            if (!isClosed && hasAssignmentChanged) {
                assignedUserTopicIdPartitions.values().forEach(utp -> {
                    metadataPartitionSnapshot.add(utp.metadataPartition);
                    assignedUserTopicIdPartitionsSnapshot.add(utp);
                });
                hasAssignmentChanged = false;
            }
        }
        if (!metadataPartitionSnapshot.isEmpty()) {
            final Set<TopicPartition> remoteLogPartitions = toRemoteLogPartitions(metadataPartitionSnapshot);
            consumer.assign(remoteLogPartitions);
            this.assignedMetadataPartitions = Collections.unmodifiableSet(metadataPartitionSnapshot);
            // for newly assigned user-partitions, read from the beginning of the corresponding metadata partition
            final Set<TopicPartition> seekToBeginOffsetPartitions = assignedUserTopicIdPartitionsSnapshot
                    .stream()
                    .filter(utp -> !utp.isAssigned)
                    .map(utp -> utp.metadataPartition)
                    // When reset to beginning is happening, we also need to reset the last read offset
                    // Otherwise if the next reassignment request for the same metadata partition comes in
                    // before the record of already assigned topic has been read, then the reset will happen again to the last read offset
                    .peek(readOffsetsByMetadataPartition::remove)
                    .map(ConsumerTask::toRemoteLogPartition)
                    .collect(Collectors.toSet());
            consumer.seekToBeginning(seekToBeginOffsetPartitions);
            // for other metadata partitions, read from the offset where the processing left last time.
            remoteLogPartitions.stream()
                .filter(tp -> !seekToBeginOffsetPartitions.contains(tp) &&
                    readOffsetsByMetadataPartition.containsKey(tp.partition()))
                .forEach(tp -> consumer.seek(tp, readOffsetsByMetadataPartition.get(tp.partition())));
            Set<TopicIdPartition> processedAssignmentPartitions = new HashSet<>();
            // mark all the user-topic-partitions as assigned to the consumer.
            assignedUserTopicIdPartitionsSnapshot.forEach(utp -> {
                if (!utp.isAssigned) {
                    // Note that there can be a race between `remove` and `add` partition assignment. Calling the
                    // `maybeLoadPartition` here again to be sure that the partition gets loaded on the handler.
                    remotePartitionMetadataEventHandler.maybeLoadPartition(utp.topicIdPartition);
                    utp.isAssigned = true;
                }
                processedAssignmentPartitions.add(utp.topicIdPartition);
            });
            processedAssignmentOfUserTopicIdPartitions = new HashSet<>(processedAssignmentPartitions);
            clearResourcesForUnassignedUserTopicPartitions(processedAssignmentPartitions);
            isAllUserTopicPartitionsInitialized = false;
            uninitializedAt = 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