def notify_barriers(self,
                      slot_key,
                      cursor,
                      use_barrier_indexes,
                      max_to_notify=_MAX_BARRIERS_TO_NOTIFY):
    """Searches for barriers affected by a slot and triggers completed ones.

    Args:
      slot_key: db.Key or stringified key of the _SlotRecord that was filled.
      cursor: Stringified Datastore cursor where the notification query
        should pick up.
      use_barrier_indexes: When True, use _BarrierIndex records to determine
        which _Barriers to trigger by having this _SlotRecord filled. When
        False, use the old method that queries for _BarrierRecords by
        the blocking_slots parameter.
      max_to_notify: Used for testing.

    Raises:
      PipelineStatusError: If any of the barriers are in a bad state.
    """
    if not isinstance(slot_key, db.Key):
      slot_key = db.Key(slot_key)
    logging.debug('Notifying slot %r', slot_key)

    if use_barrier_indexes:
      # Please see models.py:_BarrierIndex to understand how _BarrierIndex
      # entities relate to _BarrierRecord entities.
      query = (
          _BarrierIndex.all(cursor=cursor, keys_only=True)
          .ancestor(slot_key))
      barrier_index_list = query.fetch(max_to_notify)
      barrier_key_list = [
          _BarrierIndex.to_barrier_key(key) for key in barrier_index_list]

      # If there are task and pipeline kickoff retries it's possible for a
      # _BarrierIndex to exist for a _BarrierRecord that was not successfully
      # written. It's safe to ignore this because the original task that wrote
      # the _BarrierIndex and _BarrierRecord would not have made progress to
      # kick off a real pipeline or child pipeline unless all of the writes for
      # these dependent entities went through. We assume that the instigator
      # retried from scratch and somehwere there exists a good _BarrierIndex and
      # corresponding _BarrierRecord that tries to accomplish the same thing.
      barriers = db.get(barrier_key_list)
      results = []
      for barrier_key, barrier in zip(barrier_key_list, barriers):
        if barrier is None:
          logging.debug('Ignoring that Barrier "%r" is missing, '
                        'relies on Slot "%r"', barrier_key, slot_key)
        else:
          results.append(barrier)
    else:
      # TODO(user): Delete this backwards compatible codepath and
      # make use_barrier_indexes the assumed default in all cases.
      query = (
          _BarrierRecord.all(cursor=cursor)
          .filter('blocking_slots =', slot_key))
      results = query.fetch(max_to_notify)

    # Fetch all blocking _SlotRecords for any potentially triggered barriers.
    blocking_slot_keys = []
    for barrier in results:
      blocking_slot_keys.extend(barrier.blocking_slots)

    blocking_slot_dict = {}
    for slot_record in db.get(blocking_slot_keys):
      if slot_record is None:
        continue
      blocking_slot_dict[slot_record.key()] = slot_record

    task_list = []
    updated_barriers = []
    for barrier in results:
      ready_slots = []
      for blocking_slot_key in barrier.blocking_slots:
        slot_record = blocking_slot_dict.get(blocking_slot_key)
        if slot_record is None:
          raise UnexpectedPipelineError(
              'Barrier "%r" relies on Slot "%r" which is missing.' %
              (barrier.key(), blocking_slot_key))
        if slot_record.status == _SlotRecord.FILLED:
          ready_slots.append(blocking_slot_key)

      # When all of the blocking_slots have been filled, consider the barrier
      # ready to trigger. We'll trigger it regardless of the current
      # _BarrierRecord status, since there could be task queue failures at any
      # point in this flow; this rolls forward the state and de-dupes using
      # the task name tombstones.
      pending_slots = set(barrier.blocking_slots) - set(ready_slots)
      if not pending_slots:
        if barrier.status != _BarrierRecord.FIRED:
          barrier.status = _BarrierRecord.FIRED
          barrier.trigger_time = self._gettime()
          updated_barriers.append(barrier)

        purpose = barrier.key().name()
        if purpose == _BarrierRecord.START:
          path = self.pipeline_handler_path
          countdown = None
        else:
          path = self.finalized_handler_path
          # NOTE: Wait one second before finalization to prevent
          # contention on the _PipelineRecord entity.
          countdown = 1
        pipeline_key = _BarrierRecord.target.get_value_for_datastore(barrier)
        pipeline_record = db.get(pipeline_key)
        logging.debug('Firing barrier %r', barrier.key())
        task_list.append(taskqueue.Task(
            url=path,
            countdown=countdown,
            name='ae-barrier-fire-%s-%s' % (pipeline_key.name(), purpose),
            params=dict(pipeline_key=pipeline_key, purpose=purpose),
            headers={'X-Ae-Pipeline-Key': pipeline_key},
            target=pipeline_record.params['target']))
      else:
        logging.debug('Not firing barrier %r, Waiting for slots: %r',
                      barrier.key(), pending_slots)

    # Blindly overwrite _BarrierRecords that have an updated status. This is
    # acceptable because by this point all finalization barriers for
    # generator children should have already had their final outputs assigned.
    if updated_barriers:
      db.put(updated_barriers)

    # Task continuation with sequence number to prevent fork-bombs.
    if len(results) == max_to_notify:
      the_match = re.match('(.*)-ae-barrier-notify-([0-9]+)', self.task_name)
      if the_match:
        prefix = the_match.group(1)
        end = int(the_match.group(2)) + 1
      else:
        prefix = self.task_name
        end = 0
      task_list.append(taskqueue.Task(
          name='%s-ae-barrier-notify-%d' % (prefix, end),
          url=self.barrier_handler_path,
          params=dict(
              slot_key=slot_key,
              cursor=query.cursor(),
              use_barrier_indexes=use_barrier_indexes)))

    if task_list:
      try:
        taskqueue.Queue(self.queue_name).add(task_list)
      except (taskqueue.TombstonedTaskError, taskqueue.TaskAlreadyExistsError):
        pass