code: def _get_or_create_producer(self):
      if self._check_producer_alive() != True:
          # if self._producer_conn is None:
          try:
              self._producer_conn = pulsar.Client(
                  service_url="pulsar://{}:{}".format(
                      self._host, self._port
                  ),
                  operation_timeout_seconds=30,
              )
          except Exception:
              self._producer_conn = None
  
          # alway used current client to fetch producer
          try:
              self._producer_send = self._producer_conn.create_producer(
                  TOPIC_PREFIX.format(
                      self._tenant, self._namespace, self._send_topic
                  ),
                  producer_name=UNIQUE_PRODUCER_NAME,
                  send_timeout_millis=60000,
                  max_pending_messages=500,
                  compression_type=pulsar.CompressionType.LZ4,
                  **self._producer_config,
              )
          except Exception as e:
              LOGGER.debug(
                  f"catch exception {e} in creating pulsar producer"
              )
              self._producer_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 7 for calling the MQChannel obj's underlying member method code with example usage: obj._get_or_create_producer() without return types