public void stopPartitions(Set<StopPartition> stopPartitions,
                               BiConsumer<TopicPartition, Throwable> errorHandler) {
        LOGGER.debug("Stop partitions: {}", stopPartitions);
        for (StopPartition stopPartition: stopPartitions) {
            TopicPartition tp = stopPartition.topicPartition();
            try {
                if (topicIdByPartitionMap.containsKey(tp)) {
                    TopicIdPartition tpId = new TopicIdPartition(topicIdByPartitionMap.get(tp), tp);
                    leaderCopyRLMTasks.computeIfPresent(tpId, (topicIdPartition, task) -> {
                        LOGGER.info("Cancelling the copy RLM task for tpId: {}", tpId);
                        task.cancel();
                        return null;
                    });
                    leaderExpirationRLMTasks.computeIfPresent(tpId, (topicIdPartition, task) -> {
                        LOGGER.info("Cancelling the expiration RLM task for tpId: {}", tpId);
                        task.cancel();
                        return null;
                    });
                    followerRLMTasks.computeIfPresent(tpId, (topicIdPartition, task) -> {
                        LOGGER.info("Cancelling the follower RLM task for tpId: {}", tpId);
                        task.cancel();
                        return null;
                    });

                    removeRemoteTopicPartitionMetrics(tpId);

                    if (stopPartition.deleteRemoteLog()) {
                        LOGGER.info("Deleting the remote log segments task for partition: {}", tpId);
                        deleteRemoteLogPartition(tpId);
                    }
                } else {
                    LOGGER.warn("StopPartition call is not expected for partition: {}", tp);
                }
            } catch (Exception ex) {
                errorHandler.accept(tp, ex);
                LOGGER.error("Error while stopping the partition: {}", stopPartition, ex);
            }
        }
        // Note `deleteLocalLog` will always be true when `deleteRemoteLog` is true but not the other way around.
        Set<TopicIdPartition> deleteLocalPartitions = stopPartitions.stream()
                .filter(sp -> sp.deleteLocalLog() && topicIdByPartitionMap.containsKey(sp.topicPartition()))
                .map(sp -> new TopicIdPartition(topicIdByPartitionMap.get(sp.topicPartition()), sp.topicPartition()))
                .collect(Collectors.toSet());
        if (!deleteLocalPartitions.isEmpty()) {
            // NOTE: In ZK mode, this#stopPartitions method is called when Replica state changes to Offline and
            // ReplicaDeletionStarted
            remoteLogMetadataManager.onStopPartitions(deleteLocalPartitions);
            deleteLocalPartitions.forEach(tpId -> topicIdByPartitionMap.remove(tpId.topicPartition()));
        }
    }
Stop the remote-log-manager task for the given partitions. And, calls the
{@link RemoteLogMetadataManager#onStopPartitions(Set)} when {@link StopPartition#deleteLocalLog()} is true.
Deletes the partitions from the remote storage when {@link StopPartition#deleteRemoteLog()} is true.

@param stopPartitions topic partitions that needs to be stopped.
@param errorHandler   callback to handle any errors while stopping the partitions.
core/src/main/java/kafka/log/remote/RemoteLogManager.java