code: def _partition_receive(
      self,
      index,
      name,
      tag,
      src_party_id,
      src_role,
      dst_party_id,
      dst_role,
      topic_infos,
      mq,
      conf: dict,
  ):
      _, _, topic_pair = topic_infos[index]
      channel_info = self._get_channel(
          topic_pair=topic_pair,
          src_party_id=src_party_id,
          src_role=src_role,
          dst_party_id=dst_party_id,
          dst_role=dst_role,
          mq=mq,
          conf=conf,
      )
  
      message_key_cache = set()
      count = 0
      partition_size = -1
      all_data = []
  
      while True:
          try:
              for id, properties, body in self._get_consume_message(
                  channel_info
              ):
                  LOGGER.debug(
                      f"[federation._partition_receive] properties: {properties}."
                  )
                  if (
                      properties["message_id"] != name
                      or properties["correlation_id"] != tag
                  ):
                      # todo: fix this
                      self._consume_ack(channel_info, id)
                      LOGGER.debug(
                          f"[federation._partition_receive]: require {name}.{tag}, got {properties['message_id']}.{properties['correlation_id']}"
                      )
                      continue
  
                  if properties["content_type"] == "application/json":
                      header = json.loads(properties["headers"])
                      message_key = header["message_key"]
                      if message_key in message_key_cache:
                          LOGGER.debug(
                              f"[federation._partition_receive] message_key : {message_key} is duplicated"
                          )
                          self._consume_ack(channel_info, id)
                          continue
  
                      message_key_cache.add(message_key)
  
                      if header["partition_size"] >= 0:
                          partition_size = header["partition_size"]
  
                      data = json.loads(body.decode())
                      data_iter = (
                          (
                              bytes.fromhex(el["k"]),
                              bytes.fromhex(el["v"]),
                          )
                          for el in data
                      )
                      count += len(data)
                      LOGGER.debug(
                          f"[federation._partition_receive] count: {count}"
                      )
                      all_data.extend(data_iter)
                      self._consume_ack(channel_info, id)
  
                      if count == partition_size:
                          channel_info.cancel()
                          return all_data
                  else:
                      ValueError(
                          f"[federation._partition_receive]properties.content_type is {properties['content_type']}, but must be application/json"
                      )
  
          except Exception as e:
              LOGGER.error(
                  f"[federation._partition_receive]catch exception {e}, while receiving {name}.{tag}"
              )
              # avoid hang on consume()
              if count == partition_size:
                  channel_info.cancel()
                  return all_data
              else:
                  raise e
  
imports: import json
  import logging
  import sys
  import typing
  from typing import List
  from fate.arch.computing.api import KVTableContext
  from fate.arch.federation.api import Federation, PartyMeta, TableMeta
  from ._datastream import Datastream
  from ._parties import Party
label_desc: Use image node_id 25 for calling the MessageQueueBasedFederation obj's underlying member method code with example usage: obj._partition_receive(index, name, tag, src_party_id, src_role, dst_party_id, dst_role, topic_infos, mq, conf) and returns: all_data, all_data