code: def _send_kv(
      self,
      name,
      tag,
      data,
      channel_infos,
      partition_size,
      partitions,
      message_key,
  ):
      headers = json.dumps(
          {
              "partition_size": partition_size,
              "partitions": partitions,
              "message_key": message_key,
          }
      )
      for info in channel_infos:
          properties = {
              "content_type": "application/json",
              "app_id": info._dst_party_id,
              "message_id": name,
              "correlation_id": tag,
              "headers": headers,
          }
          LOGGER.debug(
              f"[federation._send_kv]info: {info}, properties: {properties}."
          )
          info.produce(body=data, properties=properties)
  
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 20 for calling the MessageQueueBasedFederation obj's underlying member method code with example usage: obj._send_kv(name, tag, data, channel_infos, partition_size, partitions, message_key) without return types