public void onLeadershipChange(Set<Partition> partitionsBecomeLeader,
                                   Set<Partition> partitionsBecomeFollower,
                                   Map<String, Uuid> topicIds) {
        LOGGER.debug("Received leadership changes for leaders: {} and followers: {}", partitionsBecomeLeader, partitionsBecomeFollower);

        if (rlmConfig.isRemoteStorageSystemEnabled() && !isRemoteLogManagerConfigured()) {
            throw new KafkaException("RemoteLogManager is not configured when remote storage system is enabled");
        }

        Set<TopicIdPartition> leaderPartitions = filterPartitions(partitionsBecomeLeader)
                .map(p -> new TopicIdPartition(topicIds.get(p.topic()), p.topicPartition())).collect(Collectors.toSet());

        Set<TopicIdPartition> followerPartitions = filterPartitions(partitionsBecomeFollower)
                .map(p -> new TopicIdPartition(topicIds.get(p.topic()), p.topicPartition())).collect(Collectors.toSet());

        if (!leaderPartitions.isEmpty() || !followerPartitions.isEmpty()) {
            LOGGER.debug("Effective topic partitions after filtering compact and internal topics, leaders: {} and followers: {}",
                    leaderPartitions, followerPartitions);

            leaderPartitions.forEach(this::cacheTopicPartitionIds);
            followerPartitions.forEach(this::cacheTopicPartitionIds);

            remoteLogMetadataManager.onPartitionLeadershipChanges(leaderPartitions, followerPartitions);
            followerPartitions.forEach(this::doHandleFollowerPartition);

            // If this node was the previous leader for the partition, then the RLMTask might be running in the
            // background thread and might emit metrics. So, removing the metrics after marking this node as follower.
            followerPartitions.forEach(this::removeRemoteTopicPartitionMetrics);

            leaderPartitions.forEach(this::doHandleLeaderPartition);
        }
    }
Callback to receive any leadership changes for the topic partitions assigned to this broker. If there are no
existing tasks for a given topic partition then it will assign new leader or follower task else it will convert the
task to respective target state(leader or follower).

@param partitionsBecomeLeader   partitions that have become leaders on this broker.
@param partitionsBecomeFollower partitions that have become followers on this broker.
@param topicIds                 topic name to topic id mappings.
core/src/main/java/kafka/log/remote/RemoteLogManager.java