docs: def fill_slot(self, filler_pipeline_key, slot, value):
      """Fills a slot, enqueueing a task to trigger pending barriers.
  
      Args:
        filler_pipeline_key: db.Key or stringified key of the _PipelineRecord
          that filled this slot.
        slot: The Slot instance to fill.
        value: The serializable value to assign.
  
      Raises:
        UnexpectedPipelineError if the _SlotRecord for the 'slot' could not
        be found in the Datastore.
      """
      if not isinstance(filler_pipeline_key, db.Key):
        filler_pipeline_key = db.Key(filler_pipeline_key)
  
      if _TEST_MODE:
        slot._set_value_test(filler_pipeline_key, value)
      else:
        encoded_value = json.dumps(value,
                                         sort_keys=True,
                                         cls=mr_util.JsonEncoder)
        value_text = None
        value_blob = None
        if len(encoded_value) <= _MAX_JSON_SIZE:
          value_text = db.Text(encoded_value)
        else:
          # The encoded value is too big. Save it as a blob.
          value_blob = _write_json_blob(encoded_value, filler_pipeline_key.name())
  
        def txn():
          slot_record = db.get(slot.key)
          if slot_record is None:
            raise UnexpectedPipelineError(
                'Tried to fill missing slot "%s" '
                'by pipeline ID "%s" with value: %r'
                % (slot.key, filler_pipeline_key.name(), value))
          # NOTE: Always take the override value here. If down-stream pipelines
          # need a consitent view of all up-stream outputs (meaning, all of the
          # outputs came from the same retry attempt of the upstream pipeline),
          # the down-stream pipeline must also wait for the 'default' output
          # of these up-stream pipelines.
          slot_record.filler = filler_pipeline_key
          slot_record.value_text = value_text
          slot_record.value_blob = value_blob
          slot_record.status = _SlotRecord.FILLED
          slot_record.fill_time = self._gettime()
          slot_record.put()
          task = taskqueue.Task(
              url=self.barrier_handler_path,
              params=dict(
                  slot_key=slot.key,
                  use_barrier_indexes=True),
              headers={'X-Ae-Slot-Key': slot.key,
                       'X-Ae-Filler-Pipeline-Key': filler_pipeline_key})
          task.add(queue_name=self.queue_name, transactional=True)
        db.run_in_transaction_options(
            db.create_transaction_options(propagation=db.ALLOWED), txn)
  
      self.session_filled_output_names.add(slot.name)