code: def _create_topic_by_replication_mode(self, party, topic_suffix):
      send_topic_name = f"{self._party.role}-{self._party.party_id}-{party.role}-{party.party_id}-{topic_suffix}"
      receive_topic_name = f"{party.role}-{party.party_id}-{self._party.role}-{self._party.party_id}-{topic_suffix}"
  
      # topic_pair is a pair of topic for sending and receiving message respectively
      topic_pair = _TopicPair(
          tenant=self._tenant,
          namespace=self._session_id,
          send=send_topic_name,
          receive=receive_topic_name,
      )
  
      if party.party_id == self._party.party_id:
          LOGGER.debug(
              "connecting to local broker, skipping cluster creation"
          )
      else:
          # init pulsar cluster
          cluster = self._pulsar_manager.get_cluster(
              party.party_id
          ).json()
          if (
              cluster.get("brokerServiceUrl", "") == ""
              and cluster.get("brokerServiceUrlTls", "") == ""
          ):
              LOGGER.debug(
                  "pulsar cluster with name %s does not exist or broker url is empty, creating...",
                  party.party_id,
              )
  
              remote_party = self._mq.route_table.get(
                  int(party.party_id), None
              )
  
              # handle party does not exist in route table first
              if remote_party is None:
                  domain = self._mq.route_table.get("default").get(
                      "domain"
                  )
                  host = f"{party.party_id}.{domain}"
                  port = self._mq.route_table.get("default").get(
                      "brokerPort", "6650"
                  )
                  sslPort = self._mq.route_table.get("default").get(
                      "brokerSslPort", "6651"
                  )
                  proxy = self._mq.route_table.get("default").get(
                      "proxy", ""
                  )
              # fetch party info from the route table
              else:
                  host = self._mq.route_table.get(
                      int(party.party_id)
                  ).get("host")
                  port = self._mq.route_table.get(
                      int(party.party_id)
                  ).get("port", "6650")
                  sslPort = self._mq.route_table.get(
                      int(party.party_id)
                  ).get("sslPort", "6651")
                  proxy = self._mq.route_table.get(
                      int(party.party_id)
                  ).get("proxy", "")
  
              broker_url = f"pulsar://{host}:{port}"
              broker_url_tls = f"pulsar+ssl://{host}:{sslPort}"
              if proxy != "":
                  proxy = f"pulsar+ssl://{proxy}"
  
              if self._pulsar_manager.create_cluster(
                  cluster_name=party.party_id,
                  broker_url=broker_url,
                  broker_url_tls=broker_url_tls,
                  proxy_url=proxy,
              ).ok:
                  LOGGER.debug(
                      "pulsar cluster with name: %s, broker_url: %s created",
                      party.party_id,
                      broker_url,
                  )
              elif self._pulsar_manager.update_cluster(
                  cluster_name=party.party_id,
                  broker_url=broker_url,
                  broker_url_tls=broker_url_tls,
                  proxy_url=proxy,
              ).ok:
                  LOGGER.debug(
                      "pulsar cluster with name: %s, broker_url: %s updated",
                      party.party_id,
                      broker_url,
                  )
              else:
                  error_message = (
                      "unable to create pulsar cluster: %s".format(
                          party.party_id
                      )
                  )
                  LOGGER.error(error_message)
                  # just leave this alone.
                  raise Exception(error_message)
  
          # update tenant
          tenant_info = self._pulsar_manager.get_tenant(
              self._tenant
          ).json()
          if party.party_id not in tenant_info["allowedClusters"]:
              tenant_info["allowedClusters"].append(party.party_id)
              if self._pulsar_manager.update_tenant(
                  self._tenant,
                  tenant_info.get("admins", []),
                  tenant_info.get(
                      "allowedClusters",
                  ),
              ).ok:
                  LOGGER.debug(
                      "successfully update tenant with cluster: %s",
                      party.party_id,
                  )
              else:
                  raise Exception("unable to update tenant")
  
      # TODO: remove this for the loop
      # init pulsar namespace
      namespaces = self._pulsar_manager.get_namespace(
          self._tenant
      ).json()
      # create namespace
      if f"{self._tenant}/{self._session_id}" not in namespaces:
          # append target cluster to the pulsar namespace
          clusters = [self._cluster]
          if (
              party.party_id != self._party.party_id
              and party.party_id not in clusters
          ):
              clusters.append(party.party_id)
  
          policy = {"replication_clusters": clusters}
  
          code = self._pulsar_manager.create_namespace(
              self._tenant, self._session_id, policies=policy
          ).status_code
          # according to https://pulsar.apache.org/admin-rest-api/?version=2.7.0&apiversion=v2#operation/getPolicies
          # return 409 if existed
          # return 204 if ok
          if code == 204 or code == 409:
              LOGGER.debug(
                  "successfully create pulsar namespace: %s",
                  self._session_id,
              )
          else:
              raise Exception(
                  "unable to create pulsar namespace with status code: {}".format(
                      code
                  )
              )
  
          # set message ttl for the namespace
          response = self._pulsar_manager.set_retention(
              self._tenant,
              self._session_id,
              retention_time_in_minutes=int(self._topic_ttl),
              retention_size_in_MB=-1,
          )
  
          LOGGER.debug(response.text)
          if response.ok:
              LOGGER.debug(
                  "successfully set message ttl to namespace: {} about {} mintues".format(
                      self._session_id, self._topic_ttl
                  )
              )
          else:
              LOGGER.debug("failed to set message ttl to namespace")
      # update party to namespace
      else:
          if party.party_id != self._party.party_id:
              clusters = (
                  self._pulsar_manager.get_cluster_from_namespace(
                      self._tenant, self._session_id
                  ).json()
              )
              if party.party_id not in clusters:
                  clusters.append(party.party_id)
                  if self._pulsar_manager.set_clusters_to_namespace(
                      self._tenant, self._session_id, clusters
                  ).ok:
                      LOGGER.debug(
                          "successfully set clusters: {}  to pulsar namespace: {}".format(
                              clusters, self._session_id
                          )
                      )
                  else:
                      raise Exception(
                          "unable to update clusters: {} to pulsar namespaces: {}".format(
                              clusters, self._session_id
                          )
                      )
  
      return topic_pair
  
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 7 for calling the PulsarFederation obj's underlying member method code with example usage: obj._create_topic_by_replication_mode(party, topic_suffix) and returns: topic_pair