code: def _get_or_create_consumer(self):
      if not self._check_consumer_alive():
          try:
              self._consumer_conn = pulsar.Client(
                  service_url="pulsar://{}:{}".format(
                      self._host, self._port
                  ),
                  operation_timeout_seconds=30,
              )
          except Exception:
              self._consumer_conn = None
  
          try:
              self._consumer_receive = self._consumer_conn.subscribe(
                  TOPIC_PREFIX.format(
                      self._tenant, self._namespace, self._receive_topic
                  ),
                  subscription_name=DEFAULT_SUBSCRIPTION_NAME,
                  consumer_name=UNIQUE_CONSUMER_NAME,
                  initial_position=pulsar.InitialPosition.Earliest,
                  replicate_subscription_state_enabled=True,
                  **self._subscription_config,
              )
  
              # set cursor to latest confirmed
              if self._latest_confirmed is not None:
                  self._consumer_receive.seek(self._latest_confirmed)
  
          except Exception as e:
              LOGGER.debug(
                  f"catch exception {e} in creating pulsar consumer"
              )
              self._consumer_conn.close()
              self._consumer_conn = None
  
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 8 for calling the MQChannel obj's underlying member method code with example usage: obj._get_or_create_consumer() without return types