private static void handleSlotFilled(HandleSlotFilledTask hsfTask) {
    Key slotKey = hsfTask.getSlotKey();
    Slot slot = querySlotOrAbandonTask(slotKey, true);
    List<Barrier> waitingList = slot.getWaitingOnMeInflated();
    if (null == waitingList) {
      throw new RuntimeException("Internal logic error: " + slot + " is not inflated");
    }
    // For each barrier that is waiting on the slot ...
    for (Barrier barrier : waitingList) {
      logger.finest("Checking " + barrier);
      // unless the barrier has already been released,
      if (!barrier.isReleased()) {
        // we check whether the barrier should be released.
        boolean shouldBeReleased = true;
        if (null == barrier.getWaitingOnInflated()) {
          throw new RuntimeException("Internal logic error: " + barrier + " is not inflated.");
        }
        // For each slot that the barrier is waiting on...
        for (SlotDescriptor sd : barrier.getWaitingOnInflated()) {
          // see if it is full.
          if (!sd.slot.isFilled()) {
            logger.finest("Not filled: " + sd.slot);
            shouldBeReleased = false;
            break;
          }
        }
        if (shouldBeReleased) {
          Key jobKey = barrier.getJobKey();
          JobRecord jobRecord = queryJobOrAbandonTask(jobKey, JobRecord.InflationType.NONE);
          jobRecord.getQueueSettings().merge(hsfTask.getQueueSettings());
          Task task;
          switch (barrier.getType()) {
            case RUN:
              task = new RunJobTask(jobKey, jobRecord.getQueueSettings());
              break;
            case FINALIZE:
              task = new FinalizeJobTask(jobKey, jobRecord.getQueueSettings());
              break;
            default:
              throw new RuntimeException("Unknown barrier type " + barrier.getType());
          }
          try {
            backEnd.enqueue(task);
          } catch (TaskAlreadyExistsException e) {
            // OK. Some other thread has already enqueued this task.
          }
        }
      }
    }
  }
private static void handleSlotFilled ( HandleSlotFilledTask hsfTask ) { Key slotKey = hsfTask . getSlotKey ( ) ; Slot slot = querySlotOrAbandonTask ( slotKey , true ) ; List < Barrier > waitingList = slot . getWaitingOnMeInflated ( ) ; if ( null == waitingList ) { throw new RuntimeException ( "Internal logic error: " + slot + " is not inflated" ) ; } for ( Barrier barrier : waitingList ) { logger . finest ( "Checking " + barrier ) ; if ( ! barrier . isReleased ( ) ) { boolean shouldBeReleased = true ; if ( null == barrier . getWaitingOnInflated ( ) ) { throw new RuntimeException ( "Internal logic error: " + barrier + " is not inflated." ) ; } for ( SlotDescriptor sd : barrier . getWaitingOnInflated ( ) ) { if ( ! sd . slot . isFilled ( ) ) { logger . finest ( "Not filled: " + sd . slot ) ; shouldBeReleased = false ; break ; } } if ( shouldBeReleased ) { Key jobKey = barrier . getJobKey ( ) ; JobRecord jobRecord = queryJobOrAbandonTask ( jobKey , JobRecord . InflationType . NONE ) ; jobRecord . getQueueSettings ( ) . merge ( hsfTask . getQueueSettings ( ) ) ; Task task ; switch ( barrier . getType ( ) ) { case RUN : task = new RunJobTask ( jobKey , jobRecord . getQueueSettings ( ) ) ; break ; case FINALIZE : task = new FinalizeJobTask ( jobKey , jobRecord . getQueueSettings ( ) ) ; break ; default : throw new RuntimeException ( "Unknown barrier type " + barrier . getType ( ) ) ; } try { backEnd . enqueue ( task ) ; } catch ( TaskAlreadyExistsException e ) { } } } } }
Handle the fact that the slot with the given key has been filled.
<p>
For each barrier that is waiting on the slot, if all of the slots that the
barrier is waiting on are now filled then the barrier should be released.
Release the barrier by enqueueing an appropriate task (either
{@link RunJobTask} or {@link FinalizeJobTask}.