docs: def train(cluster_info, cluster_meta, feed_timeout=600, qname='input'):
    """Feeds Spark partitions into the shared multiprocessing.Queue.
  
    Args:
      :cluster_info: node reservation information for the cluster (e.g. host, executor_id, pid, ports, etc)
      :cluster_meta: dictionary of cluster metadata (e.g. cluster_id, reservation.Server address, etc)
      :feed_timeout: number of seconds after which data feeding times out (600 sec default)
      :qname: *INTERNAL_USE*
  
    Returns:
      A dataRDD.mapPartitions() function
    """
    def _train(iter):
      # get shared queue, reconnecting if necessary
      mgr = _get_manager(cluster_info, util.get_ip_address(), util.read_executor_id())
      try:
        queue = mgr.get_queue(qname)
        equeue = mgr.get_queue('error')
      except (AttributeError, KeyError):
        msg = "Queue '{}' not found on this node, check for exceptions on other nodes.".format(qname)
        raise Exception(msg)
  
      state = str(mgr.get('state'))
      logging.info("mgr.state={0}".format(state))
      terminating = state == "'terminating'"
      if terminating:
        logging.info("mgr is terminating, skipping partition")
        count = sum(1 for item in iter)
        logging.info("Skipped {0} items from partition".format(count))
      else:
        logging.info("Feeding partition {0} into {1} queue {2}".format(iter, qname, queue))
        count = 0
        for item in iter:
          count += 1
          queue.put(item, block=True)
  
        # wait for consumers to finish processing all items in queue before "finishing" this iterator
        joinThr = Thread(target=queue.join)
        joinThr.start()
        timeout = feed_timeout
        while (joinThr.isAlive()):
          if (not equeue.empty()):
            e_str = equeue.get()
            equeue.task_done()
            raise Exception("exception in worker:\n" + e_str)
          time.sleep(1)
          timeout -= 1
          if timeout <= 0:
            raise Exception("Timeout while feeding partition")
  
        logging.info("Processed {0} items in partition".format(count))
  
      # check if TF is terminating feed after this partition
      if not terminating:
        state = str(mgr.get('state'))
        terminating = state == "'terminating'"
        if terminating:
          try:
            logging.info("TFSparkNode: requesting stop")
            client = reservation.Client(cluster_meta['server_addr'])
            client.request_stop()
            client.close()
          except Exception as e:
            # ignore any errors while requesting stop
            logging.debug("Error while requesting stop: {0}".format(e))
  
      return [terminating]
  
    return _train