private void removeRemoteTopicPartitionMetrics(TopicIdPartition topicIdPartition) {
        String topic = topicIdPartition.topic();
        if (!brokerTopicStats.isTopicStatsExisted(topicIdPartition.topic())) {
            // The topic metrics are already removed, removing this topic key from broker-level metrics
            brokerTopicStats.removeBrokerLevelRemoteCopyLagBytes(topic);
            brokerTopicStats.removeBrokerLevelRemoteCopyLagSegments(topic);
            brokerTopicStats.removeBrokerLevelRemoteDeleteLagBytes(topic);
            brokerTopicStats.removeBrokerLevelRemoteDeleteLagSegments(topic);
            brokerTopicStats.removeBrokerLevelRemoteLogMetadataCount(topic);
            brokerTopicStats.removeBrokerLevelRemoteLogSizeComputationTime(topic);
            brokerTopicStats.removeBrokerLevelRemoteLogSizeBytes(topic);
        } else {
            int partition = topicIdPartition.partition();
            // remove the partition metric values and update the broker-level metrics
            brokerTopicStats.removeRemoteCopyLagBytes(topic, partition);
            brokerTopicStats.removeRemoteCopyLagSegments(topic, partition);
            brokerTopicStats.removeRemoteDeleteLagBytes(topic, partition);
            brokerTopicStats.removeRemoteDeleteLagSegments(topic, partition);
            brokerTopicStats.removeRemoteLogMetadataCount(topic, partition);
            brokerTopicStats.removeRemoteLogSizeComputationTime(topic, partition);
            brokerTopicStats.removeRemoteLogSizeBytes(topic, partition);
        }
    }
Closes and releases all the resources like RemoterStorageManager and RemoteLogMetadataManager.
core/src/main/java/kafka/log/remote/RemoteLogManager.java