code: def consume(self):
      self._get_or_create_consumer()
  
      try:
          LOGGER.debug(
              "receive topic: {}".format(self._consumer_receive.topic())
          )
          receive_timeout = self._consumer_config.get(
              "receive_timeout_millis", None
          )
          if receive_timeout is not None:
              LOGGER.debug(f"receive timeout millis {receive_timeout}")
          message = self._consumer_receive.receive(
              timeout_millis=receive_timeout
          )
          return message
      except Exception:
          self._consumer_receive.seek(pulsar.MessageId.earliest)
          raise TimeoutError(
              "meet receive timeout, try to reset the cursor"
          )
  
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 3 for calling the MQChannel obj's underlying member method code with example usage: obj.consume() and returns: message