code: def _destroy(self):
      # The idea cleanup strategy is to consume all message in topics,
      # and let pulsar cluster to collect the used topics.
  
      LOGGER.debug("[pulsar.cleanup]start to cleanup...")
  
      # 1. remove subscription
      response = self._pulsar_manager.unsubscribe_namespace_all_topics(
          tenant=self._tenant,
          namespace=self._session_id,
          subscription_name=DEFAULT_SUBSCRIPTION_NAME,
      )
      if response.ok:
          LOGGER.debug("successfully unsubscribe all topics")
      else:
          LOGGER.error(response.text)
  
      # 2. reset retention policy
      response = self._pulsar_manager.set_retention(
          self._tenant,
          self._session_id,
          retention_time_in_minutes=0,
          retention_size_in_MB=0,
      )
      if response.ok:
          LOGGER.debug("successfully reset all retention policy")
      else:
          LOGGER.error(response.text)
  
      # 3. remove cluster from namespace
      response = self._pulsar_manager.set_clusters_to_namespace(
          self._tenant, self._session_id, [self._cluster]
      )
      if response.ok:
          LOGGER.debug("successfully reset all replicated cluster")
      else:
          LOGGER.error(response.text)
  
imports: import logging
  from typing import List, Optional
  from fate.arch.federation.api import PartyMeta
  from fate.arch.federation.message_queue import MessageQueueBasedFederation
  from ._mq_channel import DEFAULT_CLUSTER, DEFAULT_SUBSCRIPTION_NAME, DEFAULT_TENANT, MQChannel
  from ._pulsar_manager import PulsarManager
label_desc: Use image node_id 4 for calling the PulsarFederation obj's underlying member method code with example usage: obj._destroy() without return types