code: def __init__(
      self,
      host,
      port,
      tenant,
      namespace,
      send_topic,
      receive_topic,
      src_party_id,
      src_role,
      dst_party_id,
      dst_role,
      credential=None,
      extra_args: dict = None,
  ):
      # "host:port" is used to connect the pulsar broker
      self._host = host
      self._port = port
      self._tenant = tenant
      self._namespace = namespace
      self._send_topic = send_topic
      self._receive_topic = receive_topic
      self._credential = credential
      self._src_party_id = src_party_id
      self._src_role = src_role
      self._dst_party_id = dst_party_id
      self._dst_role = dst_role
      self._extra_args = extra_args
  
      # "_channel" is the subscriptor for the topic
      self._producer_send = None
      self._producer_conn = None
  
      self._consumer_receive = None
      self._consumer_conn = None
  
      self._sequence_id = None
  
      # these are pulsar message id
      self._latest_confirmed = None
      self._first_confirmed = None
  
      self._subscription_config = {}
      if self._extra_args.get("subscription") is not None:
          self._subscription_config.update(
              self._extra_args["subscription"]
          )
  
      self._producer_config = {}
      if self._extra_args.get("producer") is not None:
          self._producer_config.update(self._extra_args["producer"])
  
      self._consumer_config = {}
      if self._extra_args.get("consumer") is not None:
          self._consumer_config.update(self._extra_args["consumer"])
  
file_path: FATE/python/fate/arch/federation/backends/pulsar/_mq_channel.py
imports: import logging
  import pulsar
  from fate.arch.federation.message_queue import nretry
label_desc: Use image node_id 1 to create a new MQChannel object from inherited base classes: object with example: obj = MQChannel(host, port, tenant, namespace, send_topic, receive_topic, src_party_id, src_role, dst_party_id, dst_role, credential, extra_args)