private void deleteRemoteLogPartition(TopicIdPartition partition) throws RemoteStorageException, ExecutionException, InterruptedException {
        List<RemoteLogSegmentMetadata> metadataList = new ArrayList<>();
        remoteLogMetadataManager.listRemoteLogSegments(partition).forEachRemaining(metadataList::add);

        List<RemoteLogSegmentMetadataUpdate> deleteSegmentStartedEvents = metadataList.stream()
                .map(metadata ->
                        new RemoteLogSegmentMetadataUpdate(metadata.remoteLogSegmentId(), time.milliseconds(),
                                metadata.customMetadata(), RemoteLogSegmentState.DELETE_SEGMENT_STARTED, brokerId))
                .collect(Collectors.toList());
        publishEvents(deleteSegmentStartedEvents).get();

        // KAFKA-15313: Delete remote log segments partition asynchronously when a partition is deleted.
        Collection<Uuid> deletedSegmentIds = new ArrayList<>();
        for (RemoteLogSegmentMetadata metadata: metadataList) {
            deletedSegmentIds.add(metadata.remoteLogSegmentId().id());
            remoteLogStorageManager.deleteLogSegmentData(metadata);
        }
        indexCache.removeAll(deletedSegmentIds);

        List<RemoteLogSegmentMetadataUpdate> deleteSegmentFinishedEvents = metadataList.stream()
                .map(metadata ->
                        new RemoteLogSegmentMetadataUpdate(metadata.remoteLogSegmentId(), time.milliseconds(),
                                metadata.customMetadata(), RemoteLogSegmentState.DELETE_SEGMENT_FINISHED, brokerId))
                .collect(Collectors.toList());
        publishEvents(deleteSegmentFinishedEvents).get();
    }
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