text:
package org.apache.helix.task;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;

import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.apache.helix.HelixDefinedState;
import org.apache.helix.HelixManager;
import org.apache.helix.common.caches.TaskDataCache;
import org.apache.helix.controller.dataproviders.BaseControllerDataProvider;
import org.apache.helix.controller.dataproviders.WorkflowControllerDataProvider;
import org.apache.helix.controller.pipeline.AbstractBaseStage;
import org.apache.helix.controller.rebalancer.util.RebalanceScheduler;
import org.apache.helix.controller.stages.BestPossibleStateOutput;
import org.apache.helix.controller.stages.CurrentStateOutput;
import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.Message;
import org.apache.helix.model.Partition;
import org.apache.helix.model.ResourceAssignment;
import org.apache.helix.monitoring.mbeans.ClusterStatusMonitor;
import org.apache.helix.monitoring.mbeans.JobMonitor;
import org.apache.helix.task.assigner.AssignableInstance;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public abstract class AbstractTaskDispatcher {
  private static final Logger LOG = LoggerFactory.getLogger(AbstractTaskDispatcher.class);
  private static final String TASK_LATENCY_TAG = "Latency";

  // For connection management
  protected HelixManager _manager;
  protected static RebalanceScheduler _rebalanceScheduler = new RebalanceScheduler();
  protected ClusterStatusMonitor _clusterStatusMonitor;

  public void init(HelixManager manager) {
    _manager = manager;
  }

  // Job Update related methods

  public void updatePreviousAssignedTasksStatus(
      Map<String, SortedSet<Integer>> prevInstanceToTaskAssignments, Set<String> excludedInstances,
      String jobResource, CurrentStateOutput currStateOutput, JobContext jobCtx, JobConfig jobCfg,
      ResourceAssignment prevTaskToInstanceStateAssignment, TaskState jobState,
      Map<String, Set<Integer>> assignedPartitions, Set<Integer> partitionsToDropFromIs,
      Map<Integer, PartitionAssignment> paMap, TargetState jobTgtState,
      Set<Integer> skippedPartitions, WorkflowControllerDataProvider cache,
      Map<String, Set<Integer>> tasksToDrop) {

    // Get AssignableInstanceMap for releasing resources for tasks in terminal states
    AssignableInstanceManager assignableInstanceManager = cache.getAssignableInstanceManager();

    // Iterate through all instances
    for (String instance : prevInstanceToTaskAssignments.keySet()) {
      assignedPartitions.put(instance, new HashSet<>());

      // Set all dropping transitions first. These are tasks coming from Participant disconnects
      // that have some active current state (INIT or RUNNING) and the requestedState of DROPPED.
      // These need to be prioritized over any other state transitions because of the race condition
      // with the same pId (task) running on other instances. This is because in paMap, we can only
      // define one transition per pId
      if (tasksToDrop.containsKey(instance)) {
        for (int pIdToDrop : tasksToDrop.get(instance)) {
          paMap.put(pIdToDrop,
              new PartitionAssignment(instance, TaskPartitionState.DROPPED.name()));
          assignedPartitions.get(instance).add(pIdToDrop);
        }
      }

      if (excludedInstances.contains(instance)) {
        continue;
      }

      // If not an excluded instance, we must instantiate its entry in assignedPartitions
      Set<Integer> pSet = prevInstanceToTaskAssignments.get(instance);

      // We need to remove all task pId's to be dropped because we already made an assignment in
      // paMap above for them to be dropped. The following does this.
      if (tasksToDrop.containsKey(instance)) {
        pSet.removeAll(tasksToDrop.get(instance));
      }

      // Used to keep track of partitions that are in one of the final states: COMPLETED, TIMED_OUT,
      // TASK_ERROR, ERROR.
      Set<Integer> donePartitions = new TreeSet<>();
      for (int pId : pSet) {
        final String pName = pName(jobResource, pId);
        TaskPartitionState currState = updateJobContextAndGetTaskCurrentState(currStateOutput,
            jobResource, pId, pName, instance, jobCtx, jobTgtState);

        if (!instance.equals(jobCtx.getAssignedParticipant(pId))) {
          LOG.warn(
              "Instance {} does not match the assigned participant for pId {} in the job context. Skipping task scheduling.",
              instance, pId);
          continue;
        }
        // This avoids a race condition in the case that although currentState is in the following
        // error condition, the pending message (INIT->RUNNNING) might still be present.
        // This is undesirable because this prevents JobContext from getting the proper update of
        // fields including task state and task's NUM_ATTEMPTS
        if (currState == TaskPartitionState.ERROR || currState == TaskPartitionState.TASK_ERROR
            || currState == TaskPartitionState.TIMED_OUT
            || currState == TaskPartitionState.TASK_ABORTED) {
          // Do not increment the task attempt count here - it will be incremented at scheduling
          // time
          markPartitionError(jobCtx, pId, currState);
        }

        // Check for pending state transitions on this (partition, instance). If there is a pending
        // state transition, we prioritize this pending state transition and set the assignment from
        // this pending state transition, essentially "waiting" until this pending message clears
        Message pendingMessage =
            currStateOutput.getPendingMessage(jobResource, new Partition(pName), instance);
        if (pendingMessage != null && !pendingMessage.getToState().equals(currState.name())) {
          // If there is a pending message whose destination state is different from the current
          // state, just make the same assignment as the pending message. This is essentially
          // "waiting" until this state transition is complete
          processTaskWithPendingMessage(prevTaskToInstanceStateAssignment, pId, pName, instance,
              pendingMessage, jobState, currState, paMap, assignedPartitions);
          continue;
        }

        // Get AssignableInstance for this instance and TaskConfig for releasing resources
        String quotaType = jobCfg.getJobType();
        String taskId;
        if (TaskUtil.isGenericTaskJob(jobCfg)) {
          taskId = jobCtx.getTaskIdForPartition(pId);
        } else {
          taskId = pName;
        }
        TaskConfig taskConfig = jobCfg.getTaskConfig(taskId);

        // Process any requested state transitions. If there is a requested state transition, just
        // "wait" until this state transition is complete
        String requestedStateStr =
            currStateOutput.getRequestedState(jobResource, new Partition(pName), instance);
        if (requestedStateStr != null && !requestedStateStr.isEmpty()) {
          TaskPartitionState requestedState = TaskPartitionState.valueOf(requestedStateStr);
          if (requestedState.equals(currState)) {
            LOG.warn(String.format(
                "Requested state %s is the same as the current state for instance %s.",
                requestedState, instance));
          }

          // For STOPPED tasks, if the targetState is STOP, we should not honor requestedState
          // transition and make it a NOP
          if (currState == TaskPartitionState.STOPPED && jobTgtState == TargetState.STOP) {
            // This task is STOPPED and not going to be re-run, so release this task
            assignableInstanceManager.release(instance, taskConfig, quotaType);
            continue;
          }

          // This contains check is necessary because we have already traversed pIdsToDrop at the
          // beginning of this method. If we already have a dropping transition, we do not want to
          // overwrite it. Any other requestedState transitions (for example, INIT to RUNNING or
          // RUNNING to COMPLETE, can wait without affecting correctness - they will be picked up
          // in ensuing runs of the Task pipeline)
          if (!paMap.containsKey(pId)) {
            paMap.put(pId, new PartitionAssignment(instance, requestedState.name()));
          }
          assignedPartitions.get(instance).add(pId);
          if (LOG.isDebugEnabled()) {
            LOG.debug(
                String.format("Instance %s requested a state transition to %s for partition %s.",
                    instance, requestedState, pName));
          }
          continue;
        }

        switch (currState) {
        case RUNNING: {
          TaskPartitionState nextState = TaskPartitionState.RUNNING;
          if (jobState == TaskState.TIMING_OUT) {
            nextState = TaskPartitionState.TASK_ABORTED;
          } else if (jobTgtState == TargetState.STOP) {
            nextState = TaskPartitionState.STOPPED;
          } else if (jobState == TaskState.ABORTED || jobState == TaskState.FAILED
              || jobState == TaskState.FAILING || jobState == TaskState.TIMED_OUT) {
            // Drop tasks if parent job is not in progress
            paMap.put(pId, new PartitionAssignment(instance, TaskPartitionState.DROPPED.name()));
            break;
          }

          paMap.put(pId, new PartitionAssignment(instance, nextState.name()));
          assignedPartitions.get(instance).add(pId);
          if (LOG.isDebugEnabled()) {
            LOG.debug(String.format("Setting task partition %s state to %s on instance %s.", pName,
                nextState, instance));
          }
        }
          break;
        case STOPPED: {
          // TODO: This case statement might be unreachable code - Hunter
          // This code may need to be removed because once a task is STOPPED and its workflow's
          // targetState is STOP, we do not assign that stopped task. Not assigning means it will
          // not be included in previousAssignment map in the next rebalance. If it is not in
          // prevInstanceToTaskAssignments, it will never hit this part of the code
          // When the parent workflow is to be resumed (target state is START), then it will just be
          // assigned as if it were being assigned for the first time
          TaskPartitionState nextState;
          if (jobTgtState.equals(TargetState.START)) {
            nextState = TaskPartitionState.RUNNING;
          } else {
            nextState = TaskPartitionState.STOPPED;
            // This task is STOPPED and not going to be re-run, so release this task
            assignableInstanceManager.release(instance, taskConfig, quotaType);
          }
          paMap.put(pId, new JobRebalancer.PartitionAssignment(instance, nextState.name()));
          assignedPartitions.get(instance).add(pId);

          if (LOG.isDebugEnabled()) {
            LOG.debug(String.format("Setting task partition %s state to %s on instance %s.", pName,
                nextState, instance));
          }
        }
          break;
        case COMPLETED: {
          // The task has completed on this partition. Mark as such in the context object.
          donePartitions.add(pId);
          if (LOG.isDebugEnabled()) {
            LOG.debug(String.format(
                "Task partition %s has completed with state %s. Marking as such in rebalancer context.",
                pName, currState));
          }
          partitionsToDropFromIs.add(pId);
          markPartitionCompleted(jobCtx, pId);

          // This task is COMPLETED, so release this task
          assignableInstanceManager.release(instance, taskConfig, quotaType);
        }
          break;
        case TIMED_OUT:

        case TASK_ERROR:

        case TASK_ABORTED:

        case ERROR: {
          donePartitions.add(pId); // The task may be rescheduled on a different instance.
          if (LOG.isDebugEnabled()) {
            LOG.debug(String.format(
                "Task partition %s has error state %s with msg %s. Marking as such in rebalancer context.",
                pName, currState, jobCtx.getPartitionInfo(pId)));
          }
          // The error policy is to fail the task as soon a single partition fails for a specified
          // maximum number of attempts or task is in ABORTED state.
          // But notice that if job is TIMED_OUT, aborted task won't be treated as fail and won't
          // cause job fail.
          // After all tasks are aborted, they will be dropped, because of job timeout.
          if (jobState != TaskState.TIMED_OUT && jobState != TaskState.TIMING_OUT) {
            if (jobCtx.getPartitionNumAttempts(pId) >= jobCfg.getMaxAttemptsPerTask()
                || currState.equals(TaskPartitionState.TASK_ABORTED)
                || currState.equals(TaskPartitionState.ERROR)) {
              skippedPartitions.add(pId);
              partitionsToDropFromIs.add(pId);
              if (LOG.isDebugEnabled()) {
                LOG.debug("skippedPartitions:" + skippedPartitions);
              }
            } else {
              // Mark the task to be started at some later time (if enabled)
              markPartitionDelayed(jobCfg, jobCtx, pId);
            }
          }
          // Release this task
          assignableInstanceManager.release(instance, taskConfig, quotaType);
        }
          break;
        case INIT: {
          // INIT is a temporary state for tasks
          // Two possible scenarios for INIT:
          // 1. Task is getting scheduled for the first time. In this case, Task's state will go
          // from null->INIT->RUNNING, and this INIT state will be transient and very short-lived
          // 2. Task is getting scheduled for the first time, but in this case, job is timed out or
          // timing out. In this case, it will be sent back to INIT state to be removed. Here we
          // ensure that this task then goes from INIT to DROPPED so that it will be released from
          // AssignableInstance to prevent resource leak
          if (jobState == TaskState.TIMED_OUT || jobState == TaskState.TIMING_OUT
              || jobTgtState == TargetState.DELETE) {
            // Job is timed out or timing out or targetState is to be deleted, so its tasks will be
            // sent back to INIT
            // In this case, tasks' IdealState will be removed, and they will be sent to DROPPED
            partitionsToDropFromIs.add(pId);

            // Also release resources for these tasks
            assignableInstanceManager.release(instance, taskConfig, quotaType);
            break;
          } else if (jobState == TaskState.IN_PROGRESS
              && (jobTgtState != TargetState.STOP && jobTgtState != TargetState.DELETE)) {
            // Job is in progress, implying that tasks are being re-tried, so set it to RUNNING
            paMap.put(pId,
                new JobRebalancer.PartitionAssignment(instance, TaskPartitionState.RUNNING.name()));
            assignedPartitions.get(instance).add(pId);
            break;
          }
        }

        case DROPPED: {
          // currState in [INIT, DROPPED]. Do nothing, the partition is eligible to be reassigned.
          donePartitions.add(pId);
          if (LOG.isDebugEnabled()) {
            LOG.debug(String.format(
                "Task partition %s has state %s. It will be dropped from the current ideal state.",
                pName, currState));
          }
          // If it's DROPPED, release this task. If INIT, do not release
          if (currState == TaskPartitionState.DROPPED) {
            assignableInstanceManager.release(instance, taskConfig, quotaType);
          }
        }
          break;
        default:
          throw new AssertionError("Unknown enum symbol: " + currState);
        }
      }

      // Remove the set of task partitions that are completed or in one of the error states.
      pSet.removeAll(donePartitions);
    }
  }

  /**
   * Computes the partition name given the resource name and partition id.
   */
  protected String pName(String resource, int pId) {
    return String.format("%s_%s", resource, pId);
  }

  /**
   * An (instance, state) pair.
   */
  protected static class PartitionAssignment {
    public final String _instance;
    public final String _state;

    PartitionAssignment(String instance, String state) {
      _instance = instance;
      _state = state;
    }
  }

  private TaskPartitionState updateJobContextAndGetTaskCurrentState(
      CurrentStateOutput currentStateOutput, String jobResource, Integer pId, String pName,
      String instance, JobContext jobCtx, TargetState jobTgtState) {
    String currentStateString =
        currentStateOutput.getCurrentState(jobResource, new Partition(pName), instance);
    if (currentStateString == null) {
      // Task state is either DROPPED or INIT
      TaskPartitionState stateFromContext = jobCtx.getPartitionState(pId);
      // If jobTgtState is START: Since currentstate is null, this function will return INIT to
      // start the task or it will return the stateFromContext (the current context) and there is no
      // need to update the context.
      // If jobTgtState is DELETE: JobDispatcher handles this case and this part of the code will
      // not be triggered.
      // If jobTgtState is STOP:
      // If context is equal to INIT or RUNNING: Here context is set to be STOPPED.
      // Other states don't need special handling and context can remain unchanged.
      if (jobTgtState == TargetState.STOP && (stateFromContext == TaskPartitionState.RUNNING
          || stateFromContext == TaskPartitionState.INIT)) {
        jobCtx.setPartitionState(pId, TaskPartitionState.STOPPED);
        return TaskPartitionState.STOPPED;
      }
      return stateFromContext == null ? TaskPartitionState.INIT : stateFromContext;
    }
    TaskPartitionState currentState = TaskPartitionState.valueOf(currentStateString);
    jobCtx.setAssignedParticipant(pId, instance);
    jobCtx.setPartitionState(pId, currentState);
    String taskMsg = currentStateOutput.getInfo(jobResource, new Partition(pName), instance);
    if (taskMsg != null) {
      jobCtx.setPartitionInfo(pId, taskMsg);
    }
    return currentState;
  }

  /**
   * Create an assignment based on an already-existing pending message. This effectively lets the
   * Controller to "wait" until the pending state transition has been processed.
   * @param prevAssignment
   * @param pId
   * @param pName
   * @param instance
   * @param pendingMessage
   * @param jobState
   * @param currState
   * @param paMap
   * @param assignedPartitions
   */
  private void processTaskWithPendingMessage(ResourceAssignment prevAssignment, Integer pId,
      String pName, String instance, Message pendingMessage, TaskState jobState,
      TaskPartitionState currState, Map<Integer, PartitionAssignment> paMap,
      Map<String, Set<Integer>> assignedPartitions) {

    // stateMap is a mapping of Instance -> TaskPartitionState (String)
    Map<String, String> stateMap = prevAssignment.getReplicaMap(new Partition(pName));
    if (stateMap != null) {
      String prevState = stateMap.get(instance);
      if (!pendingMessage.getToState().equals(prevState)) {
        LOG.warn(String.format(
            "Task pending to-state is %s while previous assigned state is %s. This should not"
                + "happen.",
            pendingMessage.getToState(), prevState));
      }
      if (jobState == TaskState.TIMING_OUT && currState == TaskPartitionState.INIT
          && prevState.equals(TaskPartitionState.RUNNING.name())) {
        // While job is timing out, if the task is pending on INIT->RUNNING, set it back to INIT,
        // so that Helix will cancel the transition.
        paMap.put(pId, new PartitionAssignment(instance, TaskPartitionState.INIT.name()));
        assignedPartitions.get(instance).add(pId);
        if (LOG.isDebugEnabled()) {
          LOG.debug(String.format(
              "Task partition %s has a pending state transition on instance %s INIT->RUNNING. Previous state %s"
                  + "Setting it back to INIT so that Helix can cancel the transition(if enabled).",
              pName, instance, prevState));
        }
      } else {
        // Otherwise, Just copy forward
        // the state assignment from the previous ideal state.
        paMap.put(pId, new PartitionAssignment(instance, prevState));
        assignedPartitions.get(instance).add(pId);
        if (LOG.isDebugEnabled()) {
          LOG.debug(String.format(
              "Task partition %s has a pending state transition on instance %s. Using the previous ideal state which was %s.",
              pName, instance, prevState));
        }
      }
    }
  }

  protected static void markPartitionCompleted(JobContext ctx, int pId) {
    ctx.setPartitionState(pId, TaskPartitionState.COMPLETED);
    ctx.setPartitionFinishTime(pId, System.currentTimeMillis());
  }

  protected static void markPartitionError(JobContext ctx, int pId, TaskPartitionState state) {
    ctx.setPartitionState(pId, state);
    ctx.setPartitionFinishTime(pId, System.currentTimeMillis());
  }

  protected static void markAllPartitionsError(JobContext ctx) {
    for (int pId : ctx.getPartitionSet()) {
      markPartitionError(ctx, pId, TaskPartitionState.ERROR);
    }
  }

  protected static void markPartitionDelayed(JobConfig cfg, JobContext ctx, int p) {
    long delayInterval = cfg.getTaskRetryDelay();
    if (delayInterval <= 0) {
      return;
    }
    long nextStartTime = ctx.getPartitionFinishTime(p) + delayInterval;
    ctx.setNextRetryTime(p, nextStartTime);
  }

  protected void handleJobTimeout(JobContext jobCtx, WorkflowContext workflowCtx,
      String jobResource, JobConfig jobCfg) {
    jobCtx.setFinishTime(System.currentTimeMillis());
    workflowCtx.setJobState(jobResource, TaskState.TIMED_OUT);
    // Mark all INIT task to TASK_ABORTED
    for (int pId : jobCtx.getPartitionSet()) {
      if (jobCtx.getPartitionState(pId) == TaskPartitionState.INIT) {
        jobCtx.setPartitionState(pId, TaskPartitionState.TASK_ABORTED);
      }
    }
    _clusterStatusMonitor.updateJobCounters(jobCfg, TaskState.TIMED_OUT);
    _rebalanceScheduler.removeScheduledRebalance(jobResource);
    TaskUtil.cleanupJobIdealStateExtView(_manager.getHelixDataAccessor(), jobResource);
  }

  protected void failJob(String jobName, WorkflowContext workflowContext, JobContext jobContext,
      WorkflowConfig workflowConfig, Map<String, JobConfig> jobConfigMap,
      WorkflowControllerDataProvider dataProvider) {
    markJobFailed(jobName, jobContext, workflowConfig, workflowContext, jobConfigMap, dataProvider);

    // Mark all INIT task to TASK_ABORTED
    for (int pId : jobContext.getPartitionSet()) {
      if (jobContext.getPartitionState(pId) == TaskPartitionState.INIT) {
        jobContext.setPartitionState(pId, TaskPartitionState.TASK_ABORTED);
      }
    }
    _clusterStatusMonitor.updateJobCounters(jobConfigMap.get(jobName), TaskState.FAILED);
    _rebalanceScheduler.removeScheduledRebalance(jobName);
    TaskUtil.cleanupJobIdealStateExtView(_manager.getHelixDataAccessor(), jobName);
  }

  // Compute real assignment from theoretical calculation with applied throttling
  // This is the actual assigning part
  protected void handleAdditionalTaskAssignment(
      Map<String, SortedSet<Integer>> prevInstanceToTaskAssignments, Set<String> excludedInstances,
      String jobResource, CurrentStateOutput currStateOutput, JobContext jobCtx,
      final JobConfig jobCfg, final WorkflowConfig workflowConfig, WorkflowContext workflowCtx,
      final WorkflowControllerDataProvider cache,
      ResourceAssignment prevTaskToInstanceStateAssignment,
      Map<String, Set<Integer>> assignedPartitions, Map<Integer, PartitionAssignment> paMap,
      Set<Integer> skippedPartitions, TaskAssignmentCalculator taskAssignmentCal,
      Set<Integer> allPartitions, final long currentTime, Collection<String> liveInstances) {

    // See if there was LiveInstance change and cache LiveInstances from this iteration of pipeline
    boolean existsLiveInstanceOrCurrentStateChange =
        cache.getExistsLiveInstanceOrCurrentStateChange();

    // The excludeSet contains the set of task partitions that must be excluded from consideration
    // when making any new assignments.
    // This includes all completed, failed, delayed, and already assigned partitions.
    Set<Integer> excludeSet = Sets.newTreeSet();
    // Add all assigned partitions to excludeSet
    for (Set<Integer> assignedSet : assignedPartitions.values()) {
      excludeSet.addAll(assignedSet);
    }
    addCompletedTasks(excludeSet, jobCtx, allPartitions);
    addGiveupPartitions(excludeSet, jobCtx, allPartitions, jobCfg);
    excludeSet.addAll(skippedPartitions);
    Set<Integer> partitionsWithDelay = TaskUtil.getNonReadyPartitions(jobCtx, currentTime);
    excludeSet.addAll(partitionsWithDelay);

    // The following is filtering of tasks before passing them to the assigner
    // Only feed in tasks that need to be assigned (null and STOPPED)
    Set<Integer> filteredTaskPartitionNumbers = filterTasks(allPartitions, jobCtx, liveInstances);
    // Remove all excludeSet tasks to be safer because some STOPPED tasks have been already
    // re-started (excludeSet includes already-assigned partitions). Also tasks with their retry
    // limit exceed (addGiveupPartitions) will be removed as well
    filteredTaskPartitionNumbers.removeAll(excludeSet);

    Set<Integer> partitionsToRetryOnLiveInstanceChangeForTargetedJob = new HashSet<>();
    // If the job is a targeted job, in case of live instance change, we need to assign
    // non-terminal tasks so that they could be re-scheduled
    if (!TaskUtil.isGenericTaskJob(jobCfg) && existsLiveInstanceOrCurrentStateChange) {
      // This job is a targeted job, so FixedAssignmentCalculator will be used
      // There has been a live instance change. Must re-add incomplete task partitions to be
      // re-assigned and re-scheduled
      for (int partitionNum : allPartitions) {
        TaskPartitionState taskPartitionState = jobCtx.getPartitionState(partitionNum);
        if (isTaskNotInTerminalState(taskPartitionState)
            && !partitionsWithDelay.contains(partitionNum)) {
          // Some targeted tasks may have timed-out due to Participants (instances) not being
          // live, so we give tasks like these another try
          // If some of these tasks are already scheduled and running, they will be dropped as
          // well
          // Also, do not include partitions with delay that are not ready to be assigned and
          // scheduled
          partitionsToRetryOnLiveInstanceChangeForTargetedJob.add(partitionNum);
        }
      }
    }
    filteredTaskPartitionNumbers.addAll(partitionsToRetryOnLiveInstanceChangeForTargetedJob);

    // The actual assignment is computed here
    // Get instance->[partition, ...] mappings for the target resource.
    Map<String, SortedSet<Integer>> tgtPartitionAssignments = taskAssignmentCal.getTaskAssignment(
        currStateOutput, prevTaskToInstanceStateAssignment, liveInstances, jobCfg, jobCtx,
        workflowConfig, workflowCtx, filteredTaskPartitionNumbers, cache.getIdealStates());

    if (!TaskUtil.isGenericTaskJob(jobCfg) && jobCfg.isRebalanceRunningTask()) {
      // TODO: Revisit the logic for isRebalanceRunningTask() and valid use cases for it
      // TODO: isRebalanceRunningTask() was originally put in place to allow users to move
      // ("rebalance") long-running tasks, but there hasn't been a clear use case for this
      // Previously, there was a bug in the condition above (it was || where it should have been &&)
      dropRebalancedRunningTasks(tgtPartitionAssignments, prevInstanceToTaskAssignments, paMap,
          jobCtx);
    }

    // If this is a targeted job and if there was a live instance change
    if (!TaskUtil.isGenericTaskJob(jobCfg) && existsLiveInstanceOrCurrentStateChange) {
      // Drop current jobs only if they are assigned to a different instance, regardless of
      // the jobCfg.isRebalanceRunningTask() setting
      dropRebalancedRunningTasks(tgtPartitionAssignments, prevInstanceToTaskAssignments, paMap,
          jobCtx);
    }
    // Go through ALL instances and assign/throttle tasks accordingly
    for (Map.Entry<String, SortedSet<Integer>> entry : prevInstanceToTaskAssignments.entrySet()) {
      String instance = entry.getKey();
      if (!tgtPartitionAssignments.containsKey(instance)) {
        // There is no assignment made for this instance, so it is safe to skip
        continue;
      }
      if (excludedInstances.contains(instance)) {
        // There is a task assignment made for this instance, but for some reason, we cannot
        // assign to this instance. So we must skip the actual scheduling, but we must also
        // release the prematurely assigned tasks from AssignableInstance
        if (!cache.getAssignableInstanceManager().getAssignableInstanceMap()
            .containsKey(instance)) {
          continue; // This should not happen; skip!
        }
        AssignableInstanceManager assignableInstanceManager = cache.getAssignableInstanceManager();
        String quotaType = jobCfg.getJobType();
        for (int partitionNum : tgtPartitionAssignments.get(instance)) {
          // Get the TaskConfig for this partitionNumber
          String taskId = getTaskId(jobCfg, jobCtx, partitionNum);
          TaskConfig taskConfig = jobCfg.getTaskConfig(taskId);
          assignableInstanceManager.release(instance, taskConfig, quotaType);
        }
        continue;
      }
      // 1. throttled by job configuration
      // Contains the set of task partitions currently assigned to the instance.
      int jobCfgLimitation =
          jobCfg.getNumConcurrentTasksPerInstance() - assignedPartitions.get(instance).size();
      // 2. throttled by participant capacity
      int participantCapacity = cache.getInstanceConfigMap().get(instance).getMaxConcurrentTask();
      if (participantCapacity == InstanceConfig.MAX_CONCURRENT_TASK_NOT_SET) {
        participantCapacity = cache.getClusterConfig().getMaxConcurrentTaskPerInstance();
      }
      int participantLimitation =
          participantCapacity - cache.getParticipantActiveTaskCount(instance);
      // New tasks to be assigned
      int numToAssign = Math.min(jobCfgLimitation, participantLimitation);
      if (LOG.isDebugEnabled()) {
        LOG.debug(String.format(
            "Throttle tasks to be assigned to instance %s using limitation: Job Concurrent Task(%d), "
                + "Participant Max Task(%d). Remaining capacity %d.",
            instance, jobCfgLimitation, participantCapacity, numToAssign));
      }
      Set<Integer> throttledSet = new HashSet<>();
      if (numToAssign > 0) {
        List<Integer> nextPartitions = getNextPartitions(tgtPartitionAssignments.get(instance),
            excludeSet, throttledSet, numToAssign);
        for (Integer pId : nextPartitions) {
          // The following is the actual scheduling of the tasks
          String pName = pName(jobResource, pId);
          paMap.put(pId, new PartitionAssignment(instance, TaskPartitionState.RUNNING.name()));
          excludeSet.add(pId);
          jobCtx.setAssignedParticipant(pId, instance);
          jobCtx.setPartitionState(pId, TaskPartitionState.INIT);
          final long currentTimestamp = System.currentTimeMillis();
          jobCtx.setPartitionStartTime(pId, currentTimestamp);
          if (jobCtx.getExecutionStartTime() == WorkflowContext.NOT_STARTED) {
            // This means this is the very first task scheduled for this job
            jobCtx.setExecutionStartTime(currentTimestamp);
            reportSubmissionToScheduleDelay(cache, _clusterStatusMonitor, workflowConfig, jobCfg,
                currentTimestamp);
          }
          // Increment the task attempt count at schedule time
          jobCtx.incrementNumAttempts(pId);
          if (LOG.isDebugEnabled()) {
            LOG.debug(String.format("Setting task partition %s state to %s on instance %s.", pName,
                TaskPartitionState.RUNNING, instance));
          }
        }
        cache.setParticipantActiveTaskCount(instance,
            cache.getParticipantActiveTaskCount(instance) + nextPartitions.size());
      } else {
        // No assignment was actually scheduled, so this assignment needs to be released
        // Put all assignments in throttledSet. Be sure to subtract excludeSet because excludeSet is
        // already applied at filteringPartitions (excludeSet may contain partitions that are
        // currently running)
        Set<Integer> throttledSetWithExcludeSet =
            new HashSet<>(tgtPartitionAssignments.get(instance));
        throttledSetWithExcludeSet.removeAll(excludeSet); // Remove excludeSet
        throttledSet.addAll(throttledSetWithExcludeSet);
      }
      if (!throttledSet.isEmpty()) {
        // Release the tasks in throttledSet because they weren't actually assigned
        if (!cache.getAssignableInstanceManager().getAssignableInstanceMap()
            .containsKey(instance)) {
          continue;
        }
        AssignableInstanceManager assignableInstanceManager = cache.getAssignableInstanceManager();
        String quotaType = jobCfg.getJobType();
        for (int partitionNum : throttledSet) {
          // Get the TaskConfig for this partitionNumber
          String taskId = getTaskId(jobCfg, jobCtx, partitionNum);
          TaskConfig taskConfig = jobCfg.getTaskConfig(taskId);
          assignableInstanceManager.release(instance, taskConfig, quotaType);
        }
        LOG.debug(
            throttledSet.size() + "tasks are ready but throttled when assigned to participant.");
      }
    }
  }

  protected void scheduleForNextTask(String job, JobContext jobCtx, long now) {
    // Figure out the earliest schedulable time in the future of a non-complete job
    boolean shouldSchedule = false;
    long earliestTime = Long.MAX_VALUE;
    for (int p : jobCtx.getPartitionSet()) {
      long retryTime = jobCtx.getNextRetryTime(p);
      TaskPartitionState state = jobCtx.getPartitionState(p);
      state = (state != null) ? state : TaskPartitionState.INIT;
      Set<TaskPartitionState> errorStates = Sets.newHashSet(TaskPartitionState.ERROR,
          TaskPartitionState.TASK_ERROR, TaskPartitionState.TIMED_OUT);
      if (errorStates.contains(state) && retryTime > now && retryTime < earliestTime) {
        earliestTime = retryTime;
        shouldSchedule = true;
      }
    }

    // If any was found, then schedule it
    if (shouldSchedule) {
      long scheduledTime = _rebalanceScheduler.getRebalanceTime(job);
      if (scheduledTime == -1 || earliestTime < scheduledTime) {
        _rebalanceScheduler.scheduleRebalance(_manager, job, earliestTime);
      }
    }
  }

  // add all partitions that have been tried maxNumberAttempts
  protected static void addGiveupPartitions(Set<Integer> set, JobContext ctx,
      Iterable<Integer> pIds, JobConfig cfg) {
    for (Integer pId : pIds) {
      if (isTaskGivenup(ctx, cfg, pId)) {
        set.add(pId);
      }
    }
  }

  private static List<Integer> getNextPartitions(SortedSet<Integer> candidatePartitions,
      Set<Integer> excluded, Set<Integer> throttled, int n) {
    List<Integer> result = new ArrayList<>();
    for (Integer pId : candidatePartitions) {
      if (!excluded.contains(pId)) {
        if (result.size() < n) {
          result.add(pId);
        } else {
          throttled.add(pId);
        }
      }
    }
    return result;
  }

  private static void addCompletedTasks(Set<Integer> set, JobContext ctx, Iterable<Integer> pIds) {
    for (Integer pId : pIds) {
      TaskPartitionState state = ctx.getPartitionState(pId);
      if (state == TaskPartitionState.COMPLETED) {
        set.add(pId);
      }
    }
  }

  /**
   * Returns a filtered Iterable of tasks. To filter tasks in this context means to only allow tasks
   * whose contexts are either null or in STOPPED, TIMED_OUT, TASK_ERROR, or DROPPED state because
   * only the
   * tasks whose contexts are in these states are eligible to be assigned or re-tried.
   * Also, for those tasks in non-terminal states whose previously assigned instances are no longer
   * LiveInstances are re-added so that they could be re-assigned.
   * @param allPartitions
   * @param jobContext
   * @return a filter Iterable of task partition numbers
   */
  private Set<Integer> filterTasks(Iterable<Integer> allPartitions, JobContext jobContext,
      Collection<String> liveInstances) {
    Set<Integer> filteredTasks = new HashSet<>();
    for (int partitionNumber : allPartitions) {
      TaskPartitionState state = jobContext.getPartitionState(partitionNumber);
      // Allow tasks eligible for scheduling
      if (state == null || state == TaskPartitionState.STOPPED
          || state == TaskPartitionState.TIMED_OUT || state == TaskPartitionState.TASK_ERROR
          || state == TaskPartitionState.DROPPED) {
        filteredTasks.add(partitionNumber);
      }
      // Allow tasks whose assigned instances are no longer live for rescheduling
      if (isTaskNotInTerminalState(state)) {
        String assignedParticipant = jobContext.getAssignedParticipant(partitionNumber);
        if (assignedParticipant != null && !liveInstances.contains(assignedParticipant)) {
          // The assigned instance is no longer live, so mark it as DROPPED in the context
          jobContext.setPartitionState(partitionNumber, TaskPartitionState.DROPPED);
          filteredTasks.add(partitionNumber);
        }
      }
    }
    return filteredTasks;
  }

  /**
   * Returns whether if the task is not in a terminal state and could be re-scheduled.
   * @param state
   * @return
   */
  private boolean isTaskNotInTerminalState(TaskPartitionState state) {
    return state != TaskPartitionState.COMPLETED && state != TaskPartitionState.TASK_ABORTED
        && state != TaskPartitionState.DROPPED && state != TaskPartitionState.ERROR;
  }

  protected static boolean isTaskGivenup(JobContext ctx, JobConfig cfg, int pId) {
    TaskPartitionState state = ctx.getPartitionState(pId);
    if (state == TaskPartitionState.TASK_ABORTED || state == TaskPartitionState.ERROR) {
      return true;
    }
    if (state == TaskPartitionState.TIMED_OUT || state == TaskPartitionState.TASK_ERROR) {
      return ctx.getPartitionNumAttempts(pId) >= cfg.getMaxAttemptsPerTask();
    }
    return false;
  }

  /**
   * If assignment is different from previous assignment, drop the old running task if it's no
   * longer assigned to the same instance, but not removing it from excludeSet because the same task
   * should not be assigned to the new instance right away.
   * Also only drop if the old and the new assignments both have the partition (task) and they
   * differ (because that means the task has been assigned to a different instance).
   */
  private void dropRebalancedRunningTasks(Map<String, SortedSet<Integer>> newAssignment,
      Map<String, SortedSet<Integer>> oldAssignment, Map<Integer, PartitionAssignment> paMap,
      JobContext jobContext) {

    for (String instance : oldAssignment.keySet()) {
      for (int pId : oldAssignment.get(instance)) {
        if (jobContext.getPartitionState(pId) == TaskPartitionState.RUNNING) {
          // Check if the new assignment has this task on a different instance
          boolean existsInNewAssignment = false;
          for (Map.Entry<String, SortedSet<Integer>> entry : newAssignment.entrySet()) {
            if (!entry.getKey().equals(instance) && entry.getValue().contains(pId)) {
              // Found the partition number; new assignment has been made
              existsInNewAssignment = true;
              LOG.info(
                  "Currently running task partition number: {} is being dropped from instance: {} and will be newly assigned to instance: {}. This is due to a LiveInstance/CurrentState change, and because this is a targeted task.",
                  pId, instance, entry.getKey());
              break;
            }
          }
          if (existsInNewAssignment
              && instance.equals(jobContext.getAssignedParticipant(pId))
          ) {
            // We need to drop this task in the old assignment
            paMap.put(pId, new PartitionAssignment(instance, TaskPartitionState.DROPPED.name()));
            jobContext.setPartitionState(pId, TaskPartitionState.DROPPED);
            // Now it will be dropped and be rescheduled
          }
        }
      }
    }
  }

  protected void markJobComplete(final String jobName, final JobContext jobContext,
      final WorkflowConfig workflowConfig, WorkflowContext workflowContext,
      final Map<String, JobConfig> jobConfigMap,
      final WorkflowControllerDataProvider dataProvider) {
    finishJobInRuntimeJobDag(dataProvider.getTaskDataCache(), workflowConfig.getWorkflowId(),
        jobName);
    final long currentTime = System.currentTimeMillis();
    workflowContext.setJobState(jobName, TaskState.COMPLETED);
    jobContext.setFinishTime(currentTime);
    if (isWorkflowFinished(workflowContext, workflowConfig, jobConfigMap, dataProvider)) {
      workflowContext.setFinishTime(currentTime);
      updateWorkflowMonitor(workflowContext, workflowConfig);
    }
    scheduleJobCleanUp(jobConfigMap.get(jobName), workflowConfig, currentTime);

    // Job has completed successfully so report ControllerInducedDelay
    JobConfig jobConfig = jobConfigMap.get(jobName);
    if (jobConfig != null) {
      reportControllerInducedDelay(dataProvider, _clusterStatusMonitor, workflowConfig, jobConfig,
          currentTime);
    }
  }

  protected void markJobFailed(String jobName, JobContext jobContext, WorkflowConfig workflowConfig,
      WorkflowContext workflowContext, Map<String, JobConfig> jobConfigMap,
      WorkflowControllerDataProvider clusterDataCache) {
    finishJobInRuntimeJobDag(clusterDataCache.getTaskDataCache(), workflowConfig.getWorkflowId(),
        jobName);
    long currentTime = System.currentTimeMillis();
    workflowContext.setJobState(jobName, TaskState.FAILED);
    if (jobContext != null) {
      jobContext.setFinishTime(currentTime);
    }
    if (isWorkflowFinished(workflowContext, workflowConfig, jobConfigMap, clusterDataCache)) {
      workflowContext.setFinishTime(currentTime);
      updateWorkflowMonitor(workflowContext, workflowConfig);
    }
    scheduleJobCleanUp(jobConfigMap.get(jobName), workflowConfig, currentTime);
  }

  protected void scheduleJobCleanUp(JobConfig jobConfig, WorkflowConfig workflowConfig,
      long currentTime) {
    long currentScheduledTime =
        _rebalanceScheduler.getRebalanceTime(workflowConfig.getWorkflowId()) == -1 ? Long.MAX_VALUE
            : _rebalanceScheduler.getRebalanceTime(workflowConfig.getWorkflowId());
    if (currentTime + jobConfig.getExpiry() < currentScheduledTime) {
      _rebalanceScheduler.scheduleRebalance(_manager, workflowConfig.getWorkflowId(),
          currentTime + jobConfig.getExpiry());
    }
  }

  // Workflow related methods

  /**
   * Checks if the workflow has finished (either completed or failed).
   * Set the state in workflow context properly.
   * @param ctx Workflow context containing job states
   * @param cfg Workflow config containing set of jobs
   * @return returns true if the workflow
   *         1. completed (all tasks are {@link TaskState#COMPLETED})
   *         2. failed (any task is {@link TaskState#FAILED}
   *         3. workflow is {@link TaskState#TIMED_OUT}
   *         returns false otherwise.
   */
  protected boolean isWorkflowFinished(WorkflowContext ctx, WorkflowConfig cfg,
      Map<String, JobConfig> jobConfigMap, WorkflowControllerDataProvider clusterDataCache) {
    boolean incomplete = false;

    TaskState workflowState = ctx.getWorkflowState();
    if (TaskState.TIMED_OUT.equals(workflowState)) {
      // We don't update job state here as JobRebalancer will do it
      return true;
    }

    // Check if failed job count is beyond threshold and if so, fail the workflow
    // and abort in-progress jobs
    int failedJobs = 0;
    for (String job : cfg.getJobDag().getAllNodes()) {
      TaskState jobState = ctx.getJobState(job);
      if (jobState == TaskState.FAILED || jobState == TaskState.TIMED_OUT) {
        failedJobs++;
        if (!cfg.isJobQueue() && failedJobs > cfg.getFailureThreshold()) {
          ctx.setWorkflowState(TaskState.FAILED);
          LOG.info("Workflow {} reached the failure threshold, so setting its state to FAILED.",
              cfg.getWorkflowId());
          for (String jobToFail : cfg.getJobDag().getAllNodes()) {
            if (ctx.getJobState(jobToFail) == TaskState.IN_PROGRESS) {
              ctx.setJobState(jobToFail, TaskState.ABORTED);

              // Skip aborted jobs latency since they are not accurate latency for job running time
              if (_clusterStatusMonitor != null) {
                _clusterStatusMonitor.updateJobCounters(jobConfigMap.get(jobToFail),
                    TaskState.ABORTED);
              }

              // Since the job is aborted, release resources occupied by it
              // Otherwise, we run the risk of resource leak
              if (clusterDataCache != null) {
                AssignableInstanceManager assignableInstanceManager =
                    clusterDataCache.getAssignableInstanceManager();
                JobConfig jobConfig = jobConfigMap.get(jobToFail);
                String quotaType = jobConfig.getJobType();
                Map<String, TaskConfig> taskConfigMap = jobConfig.getTaskConfigMap();
                // Iterate over all tasks and release them
                for (Map.Entry<String, TaskConfig> taskEntry : taskConfigMap.entrySet()) {
                  TaskConfig taskConfig = taskEntry.getValue();
                  for (String assignableInstanceName : assignableInstanceManager
                      .getAssignableInstanceNames()) {
                    assignableInstanceManager.release(assignableInstanceName, taskConfig,
                        quotaType);
                  }
                }
              }
            }
          }
          return true;
        }
      }
      if (jobState != TaskState.COMPLETED && jobState != TaskState.FAILED
          && jobState != TaskState.TIMED_OUT) {
        incomplete = true;
      }
    }
    if (!incomplete && cfg.isTerminable()) {
      ctx.setWorkflowState(TaskState.COMPLETED);
      return true;
    }
    return false;
  }

  protected void updateWorkflowMonitor(WorkflowContext context, WorkflowConfig config) {
    if (_clusterStatusMonitor != null) {
      _clusterStatusMonitor.updateWorkflowCounters(config, context.getWorkflowState(),
          context.getFinishTime() - context.getStartTime());
    }
  }

  // Common methods

  protected Set<String> getExcludedInstances(String currentJobName, WorkflowConfig workflowCfg,
      WorkflowContext workflowContext, WorkflowControllerDataProvider cache) {
    Set<String> ret = new HashSet<>();

    if (!workflowCfg.isAllowOverlapJobAssignment()) {
      // exclude all instances that has been assigned other jobs' tasks
      for (String jobName : workflowCfg.getJobDag().getAllNodes()) {
        if (jobName.equals(currentJobName)) {
          continue;
        }
        JobContext jobContext = cache.getJobContext(jobName);
        if (jobContext == null) {
          continue;
        }
        // Also skip if the job is not currently running
        // For example, if the job here is in a terminal state (such as ABORTED), then its tasks are
        // practically not running, so we do not need to exclude instances who have tasks from dead
        // jobs
        TaskState jobState = workflowContext.getJobState(jobName);
        if (jobState != TaskState.IN_PROGRESS) {
          continue;
        }
        for (int pId : jobContext.getPartitionSet()) {
          TaskPartitionState partitionState = jobContext.getPartitionState(pId);
          if (partitionState == TaskPartitionState.INIT
              || partitionState == TaskPartitionState.RUNNING) {
            ret.add(jobContext.getAssignedParticipant(pId));
          }
        }
      }
    }
    return ret;
  }

  /**
   * Schedule the rebalancer timer for task framework elements
   * @param resourceId The resource id
   * @param startTime The resource start time
   * @param timeoutPeriod The resource timeout period. Will be -1 if it is not set.
   */
  protected void scheduleRebalanceForTimeout(String resourceId, long startTime,
      long timeoutPeriod) {
    long nextTimeout = getTimeoutTime(startTime, timeoutPeriod);
    long nextRebalanceTime = _rebalanceScheduler.getRebalanceTime(resourceId);
    if (nextTimeout >= System.currentTimeMillis()
        && (nextRebalanceTime == TaskConstants.DEFAULT_NEVER_TIMEOUT
            || nextTimeout < nextRebalanceTime)) {
      _rebalanceScheduler.scheduleRebalance(_manager, resourceId, nextTimeout);
    }
  }

  /**
   * Basic function to check task framework resources, workflow and job, are timeout
   * @param startTime Resources start time
   * @param timeoutPeriod Resources timeout period. Will be -1 if it is not set.
   * @return
   */
  protected boolean isTimeout(long startTime, long timeoutPeriod) {
    long nextTimeout = getTimeoutTime(startTime, timeoutPeriod);
    return nextTimeout != TaskConstants.DEFAULT_NEVER_TIMEOUT
        && nextTimeout <= System.currentTimeMillis();
  }

  private long getTimeoutTime(long startTime, long timeoutPeriod) {
    return (timeoutPeriod == TaskConstants.DEFAULT_NEVER_TIMEOUT
        || timeoutPeriod > Long.MAX_VALUE - startTime)
            // check long overflow
            ? TaskConstants.DEFAULT_NEVER_TIMEOUT
            : startTime + timeoutPeriod;
  }

  /**
   * Set the ClusterStatusMonitor for metrics update
   */
  public void setClusterStatusMonitor(ClusterStatusMonitor clusterStatusMonitor) {
    _clusterStatusMonitor = clusterStatusMonitor;
  }

  /**
   * Returns an appropriate TaskId depending on whether the job is targeted or not.
   * @param jobCfg
   * @param jobCtx
   * @param partitionNum
   * @return
   */
  private String getTaskId(JobConfig jobCfg, JobContext jobCtx, int partitionNum) {
    if (TaskUtil.isGenericTaskJob(jobCfg)) {
      return jobCtx.getTaskIdForPartition(partitionNum);
    }
    // This is a targeted task
    return pName(jobCfg.getJobId(), partitionNum);
  }

  /**
   * Checks if the workflow has been stopped.
   * In the case of a recurrent workflow template, we look at its TargetState.
   * @param ctx Workflow context containing task states
   * @param cfg Workflow config containing set of tasks
   * @return returns true if all tasks are {@link TaskState#STOPPED}, false otherwise.
   */
  protected boolean isWorkflowStopped(WorkflowContext ctx, WorkflowConfig cfg) {
    if (cfg.isRecurring()) {
      return cfg.getTargetState() == TargetState.STOP;
    }

    for (String job : cfg.getJobDag().getAllNodes()) {
      TaskState jobState = ctx.getJobState(job);
      if (jobState != null
          && (jobState.equals(TaskState.IN_PROGRESS) || jobState.equals(TaskState.STOPPING))) {
        return false;
      }
    }
    return true;
  }

  protected ResourceAssignment buildEmptyAssignment(String name,
      CurrentStateOutput currStateOutput) {
    ResourceAssignment assignment = new ResourceAssignment(name);
    Set<Partition> partitions = currStateOutput.getCurrentStateMappedPartitions(name);
    for (Partition partition : partitions) {
      Map<String, String> currentStateMap = currStateOutput.getCurrentStateMap(name, partition);
      Map<String, String> replicaMap = Maps.newHashMap();
      for (String instanceName : currentStateMap.keySet()) {
        replicaMap.put(instanceName, HelixDefinedState.DROPPED.toString());
      }
      assignment.addReplicaMap(partition, replicaMap);
    }
    return assignment;
  }

  /**
   * Check all the dependencies of a job to determine whether the job is ready to be scheduled.
   * @param job
   * @param workflowCfg
   * @param workflowCtx
   * @return
   */
  protected boolean isJobReadyToSchedule(String job, WorkflowConfig workflowCfg,
      WorkflowContext workflowCtx, int incompleteAllCount, Map<String, JobConfig> jobConfigMap,
      WorkflowControllerDataProvider clusterDataCache,
      AssignableInstanceManager assignableInstanceManager) {
    int notStartedCount = 0;
    int failedOrTimeoutCount = 0;
    int incompleteParentCount = 0;
    JobConfig jobConfig = jobConfigMap.get(job);

    if (jobConfig == null) {
      LOG.error(String.format("The job config is missing for job %s", job));
      return false;
    }

    String quotaType = TaskAssignmentCalculator.getQuotaType(workflowCfg, jobConfig);
    if (quotaType == null || !assignableInstanceManager.hasQuotaType(quotaType)) {
      quotaType = AssignableInstance.DEFAULT_QUOTA_TYPE;
    }

    if (!assignableInstanceManager.hasGlobalCapacity(quotaType)) {
      LOG.info(String.format(
          "Job %s not ready to schedule due to not having enough quota for quota type %s", job,
          quotaType));
      return false;
    }

    for (String parent : workflowCfg.getJobDag().getDirectParents(job)) {
      TaskState jobState = workflowCtx.getJobState(parent);
      if (jobState == null || jobState == TaskState.NOT_STARTED) {
        ++notStartedCount;
      } else if (jobState == TaskState.FAILED || jobState == TaskState.TIMED_OUT) {
        ++failedOrTimeoutCount;
      } else if (jobState != TaskState.COMPLETED) {
        incompleteParentCount++;
      }
    }

    // If there is any parent job not started, this job should not be scheduled
    if (notStartedCount > 0) {
      if (LOG.isDebugEnabled()) {
        LOG.debug(String.format("Job %s is not ready to start, notStartedParent(s)=%d.", job,
            notStartedCount));
      }
      return false;
    }

    // If there is parent job failed, schedule the job only when ignore dependent
    // job failure enabled
    if (failedOrTimeoutCount > 0 && !jobConfig.isIgnoreDependentJobFailure()) {
      markJobFailed(job, null, workflowCfg, workflowCtx, jobConfigMap, clusterDataCache);
      if (LOG.isDebugEnabled()) {
        LOG.debug(String.format("Job %s is not ready to start, failedCount(s)=%d.", job,
            failedOrTimeoutCount));
      }
      return false;
    }

    if (workflowCfg.isJobQueue()) {
      // If job comes from a JobQueue, it should apply the parallel job logics
      if (incompleteAllCount >= workflowCfg.getParallelJobs()) {
        if (LOG.isDebugEnabled()) {
          LOG.debug(String.format("Job %s is not ready to schedule, inCompleteJobs(s)=%d.", job,
              incompleteAllCount));
        }
        return false;
      }
    } else {
      // If this job comes from a generic workflow, job will not be scheduled until
      // all the direct parent jobs finished
      if (incompleteParentCount > 0) {
        if (LOG.isDebugEnabled()) {
          LOG.debug(String.format("Job %s is not ready to start, notFinishedParent(s)=%d.", job,
              incompleteParentCount));
        }
        return false;
      }
    }

    return true;
  }

  /**
   * Check if a workflow is ready to schedule.
   * @param workflowCfg the workflow to check
   * @return true if the workflow is ready for schedule, false if not ready
   */
  protected boolean isWorkflowReadyForSchedule(WorkflowConfig workflowCfg) {
    Date startTime = workflowCfg.getStartTime();
    // Workflow with non-scheduled config or passed start time is ready to schedule.
    return (startTime == null || startTime.getTime() <= System.currentTimeMillis());
  }

  public void updateBestPossibleStateOutput(String resource,
      ResourceAssignment partitionStateAssignment, BestPossibleStateOutput output) {
    // Use the internal MappingCalculator interface to compute the final assignment
    // The next release will support rebalancers that compute the mapping from start to finish
    for (Partition partition : partitionStateAssignment.getMappedPartitions()) {
      Map<String, String> newStateMap = partitionStateAssignment.getReplicaMap(partition);
      output.setState(resource, partition, newStateMap);
    }
  }

  protected void finishJobInRuntimeJobDag(TaskDataCache clusterDataCache, String workflowName,
      String jobName) {
    RuntimeJobDag runtimeJobDag = clusterDataCache.getRuntimeJobDag(workflowName);
    if (runtimeJobDag != null) {
      runtimeJobDag.finishJob(jobName);
      LOG.debug(
          String.format("Finish job %s of workflow %s for runtime job DAG", jobName, workflowName));
    } else {
      LOG.warn(String.format("Failed to find runtime job DAG for workflow %s and job %s",
          workflowName, jobName));
    }
  }

  /**
   * TODO: Move this logic to Task Framework metrics class for refactoring.
   * Computes and passes on submissionToProcessDelay to the dynamic metric.
   * @param dataProvider
   * @param clusterStatusMonitor
   * @param workflowConfig
   * @param jobConfig
   * @param currentTimestamp
   */
  protected static void reportSubmissionToProcessDelay(BaseControllerDataProvider dataProvider,
      final ClusterStatusMonitor clusterStatusMonitor, final WorkflowConfig workflowConfig,
      final JobConfig jobConfig, final long currentTimestamp) {
    AbstractBaseStage.asyncExecute(dataProvider.getAsyncTasksThreadPool(), () -> {
      // Asynchronously update the appropriate JobMonitor
      JobMonitor jobMonitor = clusterStatusMonitor
          .getJobMonitor(TaskAssignmentCalculator.getQuotaType(workflowConfig, jobConfig));
      if (jobMonitor == null) {
        return null;
      }

      // Compute SubmissionToProcessDelay
      long submissionToProcessDelay = currentTimestamp - jobConfig.getStat().getCreationTime();
      jobMonitor.updateSubmissionToProcessDelayGauge(submissionToProcessDelay);
      return null;
    });
  }

  /**
   * TODO: Move this logic to Task Framework metrics class for refactoring.
   * Computes and passes on submissionToScheduleDelay to the dynamic metric.
   * @param dataProvider
   * @param clusterStatusMonitor
   * @param workflowConfig
   * @param jobConfig
   * @param currentTimestamp
   */
  private static void reportSubmissionToScheduleDelay(BaseControllerDataProvider dataProvider,
      final ClusterStatusMonitor clusterStatusMonitor, final WorkflowConfig workflowConfig,
      final JobConfig jobConfig, final long currentTimestamp) {
    AbstractBaseStage.asyncExecute(dataProvider.getAsyncTasksThreadPool(), () -> {
      // Asynchronously update the appropriate JobMonitor
      JobMonitor jobMonitor = clusterStatusMonitor
          .getJobMonitor(TaskAssignmentCalculator.getQuotaType(workflowConfig, jobConfig));
      if (jobMonitor == null) {
        return null;
      }

      // Compute SubmissionToScheduleDelay
      long submissionToStartDelay = currentTimestamp - jobConfig.getStat().getCreationTime();
      jobMonitor.updateSubmissionToScheduleDelayGauge(submissionToStartDelay);
      return null;
    });
  }

  /**
   * TODO: Move this logic to Task Framework metrics class for refactoring.
   * Computes and passes on controllerInducedDelay to the dynamic metric.
   * @param dataProvider
   * @param clusterStatusMonitor
   * @param workflowConfig
   * @param jobConfig
   * @param currentTimestamp
   */
  private static void reportControllerInducedDelay(BaseControllerDataProvider dataProvider,
      final ClusterStatusMonitor clusterStatusMonitor, final WorkflowConfig workflowConfig,
      final JobConfig jobConfig, final long currentTimestamp) {
    AbstractBaseStage.asyncExecute(dataProvider.getAsyncTasksThreadPool(), () -> {
      // Asynchronously update the appropriate JobMonitor
      JobMonitor jobMonitor = clusterStatusMonitor
          .getJobMonitor(TaskAssignmentCalculator.getQuotaType(workflowConfig, jobConfig));
      if (jobMonitor == null) {
        return null;
      }

      // Compute ControllerInducedDelay only if the workload is a test load
      // NOTE: this metric cannot be computed for general user-submitted workloads because
      // the actual runtime of the tasks vary, and there could exist multiple tasks per
      // job
      // NOTE: a test workload will have the "latency" field in the mapField of the
      // JobConfig (taskConfig)
      String firstTask = jobConfig.getTaskConfigMap().keySet().iterator().next();
      if (jobConfig.getTaskConfig(firstTask).getConfigMap().containsKey(TASK_LATENCY_TAG)) {
        long taskDuration =
            Long.valueOf(jobConfig.getTaskConfig(firstTask).getConfigMap().get(TASK_LATENCY_TAG));
        long controllerInducedDelay =
            currentTimestamp - jobConfig.getStat().getCreationTime() - taskDuration;
        jobMonitor.updateControllerInducedDelayGauge(controllerInducedDelay);
      }
      return null;
    });
  }
}


text:
package it.uniroma2.sag.kelp.tree;

import org.junit.Assert;
import org.junit.Test;

import edu.stanford.nlp.ie.machinereading.RelationFeatureFactory.DEPENDENCY_TYPE;
import it.uniroma2.sag.kelp.data.representation.tree.TreeRepresentation;
import it.uniroma2.sag.kelp.input.parser.DependencyParser;
import it.uniroma2.sag.kelp.input.parser.impl.StanfordParserWrapper;
import it.uniroma2.sag.kelp.input.parser.model.DependencyGraph;
import it.uniroma2.sag.kelp.input.tree.TreeRepresentationGenerator;
import it.uniroma2.sag.kelp.input.tree.generators.OriginalPOSLabelGenerator;
import it.uniroma2.sag.kelp.input.tree.generators.PosElementLabelGenerator;
import it.uniroma2.sag.kelp.input.tree.generators.LemmaCompactPOSLabelGeneratorLowerCase;
import it.uniroma2.sag.kelp.input.tree.generators.LexicalElementLabelGenerator;
import it.uniroma2.sag.kelp.input.tree.generators.SyntElementLabelGenerator;
import it.uniroma2.sag.kelp.input.tree.generators.RelationNameLabelGenerator;

public class CompositionalGrctTreeRepresentationGeneratorTest {
	private String testSentence1 = "The cat runs over the grass.";
	private String testSentence2 = "Yesterday, I was at the sea looking for a shell.";

	@Test
	public void testCGRCTExtractionBasic() {
		DependencyParser parser = new StanfordParserWrapper();
		parser.initialize();
		SyntElementLabelGenerator rg = new RelationNameLabelGenerator();
		LexicalElementLabelGenerator ng = new LemmaCompactPOSLabelGeneratorLowerCase();
		PosElementLabelGenerator ig = new OriginalPOSLabelGenerator();

		DependencyGraph parse = parser.parse(testSentence1);
		TreeRepresentation grctGenerator1 = TreeRepresentationGenerator.cgrctGenerator(parse, rg, ng, ig);
		String grct1 = grctGenerator1.getTextFromData();
		System.out.println(grct1.replace("(", "[").replace(")", "]"));
		String expected1 = "(COMP##root<*::*,run::v>(COMP##nsubj<run::v,cat::n>(COMP##det<cat::n,the::d>(POS##DT(LEX##the::d)))(POS##NN(LEX##cat::n)))(POS##VBZ(LEX##run::v))(COMP##nmod<run::v,grass::n>(COMP##case<grass::n,over::i>(POS##IN(LEX##over::i)))(COMP##det<grass::n,the::d>(POS##DT(LEX##the::d)))(POS##NN(LEX##grass::n)))(COMP##punct<run::v,.::.>(POS##.(LEX##.::.))))";
		Assert.assertEquals(expected1, grct1);

		parse = parser.parse(testSentence2);
		TreeRepresentation grctGenerator2 = TreeRepresentationGenerator.cgrctGenerator(parse, rg, ng, ig);
		String grct2 = grctGenerator2.getTextFromData();
		System.out.println(grct2.replace("(", "[").replace(")", "]"));
		String expected2 = "(COMP##root<*::*,look::v>(COMP##nmod:tmod<look::v,yesterday::n>(POS##NN(LEX##yesterday::n)))(COMP##punct<look::v,_comma_::_comma_>(POS##,(LEX##,::,)))(COMP##nsubj<look::v,i::p>(POS##PRP(LEX##i::p)))(COMP##aux<look::v,be::v>(POS##VBD(LEX##be::v)))(COMP##advmod<look::v,sea::n>(COMP##case<sea::n,at::i>(POS##IN(LEX##at::i)))(COMP##det<sea::n,the::d>(POS##DT(LEX##the::d)))(POS##NN(LEX##sea::n)))(POS##VBG(LEX##look::v))(COMP##nmod<look::v,shell::n>(COMP##case<shell::n,for::i>(POS##IN(LEX##for::i)))(COMP##det<shell::n,a::d>(POS##DT(LEX##a::d)))(POS##NN(LEX##shell::n)))(COMP##punct<look::v,.::.>(POS##.(LEX##.::.))))";
		Assert.assertEquals(expected2, grct2);
	}

	@Test
	public void testCGRCTExtractionCollapsed() {
		DependencyParser parser = new StanfordParserWrapper(DEPENDENCY_TYPE.COLLAPSED);
		parser.initialize();
		SyntElementLabelGenerator rg = new RelationNameLabelGenerator();
		LexicalElementLabelGenerator ng = new LemmaCompactPOSLabelGeneratorLowerCase();
		PosElementLabelGenerator ig = new OriginalPOSLabelGenerator();

		DependencyGraph parse = parser.parse(testSentence1);
		TreeRepresentation grctGenerator1 = TreeRepresentationGenerator.cgrctGenerator(parse, rg, ng, ig);
		String grct1 = grctGenerator1.getTextFromData();
		System.out.println(grct1.replace("(", "[").replace(")", "]"));
		String expected1 = "(COMP##root<*::*,run::v>(COMP##nsubj<run::v,cat::n>(COMP##det<cat::n,the::d>(POS##DT(LEX##the::d)))(POS##NN(LEX##cat::n)))(POS##VBZ(LEX##run::v))(COMP##nmod:over<run::v,grass::n>(COMP##case<grass::n,over::i>(POS##IN(LEX##over::i)))(COMP##det<grass::n,the::d>(POS##DT(LEX##the::d)))(POS##NN(LEX##grass::n)))(COMP##punct<run::v,.::.>(POS##.(LEX##.::.))))";
		Assert.assertEquals(expected1, grct1);

		parse = parser.parse(testSentence2);
		TreeRepresentation grctGenerator2 = TreeRepresentationGenerator.cgrctGenerator(parse, rg, ng, ig);
		String grct2 = grctGenerator2.getTextFromData();
		System.out.println(grct2.replace("(", "[").replace(")", "]"));
		String expected2 = "(COMP##root<*::*,look::v>(COMP##nmod:tmod<look::v,yesterday::n>(POS##NN(LEX##yesterday::n)))(COMP##punct<look::v,_comma_::_comma_>(POS##,(LEX##,::,)))(COMP##nsubj<look::v,i::p>(POS##PRP(LEX##i::p)))(COMP##aux<look::v,be::v>(POS##VBD(LEX##be::v)))(COMP##advmod<look::v,sea::n>(COMP##case<sea::n,at::i>(POS##IN(LEX##at::i)))(COMP##det<sea::n,the::d>(POS##DT(LEX##the::d)))(POS##NN(LEX##sea::n)))(POS##VBG(LEX##look::v))(COMP##nmod:for<look::v,shell::n>(COMP##case<shell::n,for::i>(POS##IN(LEX##for::i)))(COMP##det<shell::n,a::d>(POS##DT(LEX##a::d)))(POS##NN(LEX##shell::n)))(COMP##punct<look::v,.::.>(POS##.(LEX##.::.))))";
		Assert.assertEquals(expected2, grct2);
	}

	@Test
	public void testCGRCTExtractionCollapsedCC() {
		DependencyParser parser = new StanfordParserWrapper(DEPENDENCY_TYPE.COLLAPSED_CCPROCESSED);
		parser.initialize();
		SyntElementLabelGenerator rg = new RelationNameLabelGenerator();
		LexicalElementLabelGenerator ng = new LemmaCompactPOSLabelGeneratorLowerCase();
		PosElementLabelGenerator ig = new OriginalPOSLabelGenerator();

		DependencyGraph parse = parser.parse(testSentence1);
		TreeRepresentation grctGenerator1 = TreeRepresentationGenerator.cgrctGenerator(parse, rg, ng, ig);
		String grct1 = grctGenerator1.getTextFromData();
		System.out.println(grct1.replace("(", "[").replace(")", "]"));
		String expected1 = "(COMP##root<*::*,run::v>(COMP##nsubj<run::v,cat::n>(COMP##det<cat::n,the::d>(POS##DT(LEX##the::d)))(POS##NN(LEX##cat::n)))(POS##VBZ(LEX##run::v))(COMP##nmod:over<run::v,grass::n>(COMP##case<grass::n,over::i>(POS##IN(LEX##over::i)))(COMP##det<grass::n,the::d>(POS##DT(LEX##the::d)))(POS##NN(LEX##grass::n)))(COMP##punct<run::v,.::.>(POS##.(LEX##.::.))))";
		Assert.assertEquals(expected1, grct1);

		parse = parser.parse(testSentence2);
		TreeRepresentation grctGenerator2 = TreeRepresentationGenerator.cgrctGenerator(parse, rg, ng, ig);
		String grct2 = grctGenerator2.getTextFromData();
		System.out.println(grct2.replace("(", "[").replace(")", "]"));
		String expected2 = "(COMP##root<*::*,look::v>(COMP##nmod:tmod<look::v,yesterday::n>(POS##NN(LEX##yesterday::n)))(COMP##punct<look::v,_comma_::_comma_>(POS##,(LEX##,::,)))(COMP##nsubj<look::v,i::p>(POS##PRP(LEX##i::p)))(COMP##aux<look::v,be::v>(POS##VBD(LEX##be::v)))(COMP##advmod<look::v,sea::n>(COMP##case<sea::n,at::i>(POS##IN(LEX##at::i)))(COMP##det<sea::n,the::d>(POS##DT(LEX##the::d)))(POS##NN(LEX##sea::n)))(POS##VBG(LEX##look::v))(COMP##nmod:for<look::v,shell::n>(COMP##case<shell::n,for::i>(POS##IN(LEX##for::i)))(COMP##det<shell::n,a::d>(POS##DT(LEX##a::d)))(POS##NN(LEX##shell::n)))(COMP##punct<look::v,.::.>(POS##.(LEX##.::.))))";
		Assert.assertEquals(expected2, grct2);
	}
}

redpajama_set_name: RedPajamaGithub


text:
<?php

namespace Prophecy\Doubler\Generator\Node;

/**
 * Argument node.
 *
 * @author Konstantin Kudryashov <ever.zet@gmail.com>
 */
class ArgumentNode {

    private $name;
    private $typeHint;
    private $default;
    private $optional = false;
    private $byReference = false;

    /**
     * @param string $name
     */
    public function __construct($name) {
        $this->name = $name;
    }

    public function getName() {
        return $this->name;
    }

    public function getTypeHint() {
        return $this->typeHint;
    }

    public function setTypeHint($typeHint = null) {
        $this->typeHint = $typeHint;
    }

    public function getDefault() {
        return $this->default;
    }

    public function setDefault($default = null) {
        $this->optional = true;
        $this->default = $default;
    }

    public function isOptional() {
        return $this->optional;
    }

    public function setAsPassedByReference($byReference = true) {
        $this->byReference = $byReference;
    }

    public function isPassedByReference() {
        return $this->byReference;
    }

}


text:
package uk.co.marketplace.repository;

import org.springframework.data.jpa.repository.Modifying;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.CrudRepository;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import uk.co.marketplace.domain.Offer;

import java.util.List;

/**
 * Offer repository
 */
public interface OfferRepository extends CrudRepository<Offer, Integer> {

    /**
     * Add an offer
     * @param offer
     * @return saved offer
     */
     Offer save(Offer offer);

    /**
     * List the offers for a given seller
     * @param sellerId
     * @return list of offers for a given user
     */
    List<Offer> findByUser(Integer sellerId);

    /**
     * Retrieve the current offer price for a given item ID (i.e. the lowest price of all offers for that item)
     * @param itemId
     * @return lowest offer price for an item
     */
    @Query("SELECT MIN(pricePerUnit) FROM Offer WHERE ITEMID=?1")
    Integer findMinimumOfferPriceOfItem(String itemId);

    /**
     * Find all the offers with a given price
     * @param price
     * @return List of all offers with a given price
     */
    List<Offer> findByPricePerUnit(Integer price);

    /**
     * Find an offer by it's id
     * @param id
     * @return the offer with the given id
     */
    Offer findById(Integer id);

    /**
     * Delete by Id
     * @param id
     */
    @Modifying
    @Transactional
    @Query("DELETE FROM Offer WHERE id=?1")
    void deleteById(Integer id);

    /**
     * Get all the offers in the database
     * @return list of offers
     */
    List<Offer> findAll();

    /**
     * Get a list of offers with the same itemId
     * @return a list of offers
     */
    List<Offer> findByItemId(String itemId);

    /**
     * Get a list of offers with the same quantity
     * @param quantity
     * @return a list of offers with the given quantity
     */
    List<Offer> findByQuantity(Integer quantity);

}


text:
<wicket:panel>
    <div wicket:id="menuPanel"></div>
</wicket:panel>


text:
<?php

namespace PCI\Providers\User;

use Illuminate\Support\ServiceProvider;
use PCI\Models\Address;
use PCI\Repositories\Interfaces\User\AddressRepositoryInterface;
use PCI\Repositories\Interfaces\User\EmployeeRepositoryInterface;
use PCI\Repositories\User\AddressRepository;

class AddressRepositoryProvider extends ServiceProvider
{

    /**
     * Register the application services.
     * @return void
     */
    public function register()
    {
        // asi como el repositorio de empleado
        // debemos pasarle parametros adicionales
        // al constructor del repositorio de direccion.
        $this->app->bind(AddressRepositoryInterface::class, function ($app) {
            return new AddressRepository(
                $app[Address::class],
                $app[EmployeeRepositoryInterface::class]
            );
        });
    }
}


text:
<?php
/**
 *
 * @author h.woltersdorf
 */

namespace Fortuneglobe\IceHawk\Responses;

use Fortuneglobe\IceHawk\Constants\Http;

/**
 * Class Forbidden
 *
 * @package Fortuneglobe\IceHawk\Responses
 */
final class Forbidden extends BaseResponse
{

	/** @var string */
	private $charset;

	/**
	 * @param string $charset
	 */
	public function __construct( $charset = 'utf-8' )
	{
		$this->charset = $charset;
	}

	public function respond()
	{
		header( 'Content-Type: text/plain; charset=' . $this->charset, true, Http::FORBIDDEN );
		echo "Forbidden!";
	}
}


text:
{
  title: 'Jenkins with GitHub integration',
  date: '2014-07-07',
  author: 'Alexander Schmidt'
}

Even now that [Travis CI](https://travis-ci.org/) is having more and more success (one of the main point being its simplicity and ease of use) not everyone has it in his or her technology stack as it is not free for private company use.
One of the great features `Travis` offers is its integration with GitHub. At least this point can be achieved with [Jenkins](http://jenkins-ci.org/) too by installing some plugins and taking some time to configure it appropriately.

The feature you will achieve by following the steps below will be:

- Push code to a GitHub repository and a build is triggered
- Have the result of the build displayed in GitHub (e.g. in a pull request). This is very useful as you will see if everything (seems to be) ok prior to merging the changes back to a stable branch like master

As I found out (twice) the configuration part is not as easy as it could and should be. That's why I decided to write down some of my experiences.

Prepare GitHub
--------------

Register a ssh key for your `Jenkins` server:

- If you don't have a ssh key I highly recommend adding one - just follow the instructions in [the GitHub help](https://help.github.com/articles/generating-ssh-keys). Note: You have to use the public part of your generated key here
- You can use the private part of your key for `Jenkins` later

Configure the web hooks:

- Visit your repository on GitHub and click `Settings`
- Select `Webhooks & Services`
- Click `Add Service -> Jenkins (GitHub plugin)`
- The configuration depends on the fact if your `Jenkins` server is reachable over the internet
  - If yes: You can see the URL under which your `Jenkins` waits for messages from the outside when you go to the general settings and click on the question mark icon next to the `Let Jenkins auto-manage hook URLs` in the `GitHub Web Hook` area. Enter this URL in the GitHub `Jenkins hook url` field
  - If no: Unfortunately this means you have to let `Jenkins` poll for changes. Check the `Poll SCM` option and enter a schedule for it. To check every minute for changes you can use `* * * * *`. This works quite good for pull requests but doesn't for single commits but as I said at least you will have the major advantage of seeing if your branch was building succesful

Create a `personal access token` on your GitHub users `application` settings page:

- Important: Make sure that at least the following checkboxes are selected otherwise you won't have the right permissions
  - `repo`
  - `admin:org`
  - `admin:public_key`
  - `admin:repo_hook`
- Remember this generated token - you will need it later as your `OAuth token` for `Jenkins` (and won't be able to see it again once you leave the page where it was shown)

Prepare Jenkins
---------------

Install the Git and GitHub plugin for `Jenkins`:

- Startup Jenkins
- Go to `Manage Jenkins -> Manage Plugins -> Available`
- Select `Git Plugin` and `GitHub Plugin` (if they are not available you might need to go to the advanced options and manually trigger `Check now`)
- Click `Download now and install after restart` - the download of the packages and dependencies will begin
- Tick the checkbox for restarting the server after this step has finished and the server will restart
- Add a new `New Item` or select one of your already existing jobs and click `Configure`

Configure the `Git Plugin`:

- In `Source Code Management` select `Git`
  - Point the `Repository URL` to your GitHub repository (e.g. https://github.com/yourUserName/TestRepo)
  - Select the correct credentials for your GitHub repository or add them now
  - If you created a keypair earlier You can use the private key part of it here
  - On a production environment you would most likely have your own ssh keys on the machine that is running `Jenkins` - in this case you would use them
  - As always with such keypairs you have the private key here for which you entered the public key on GitHub)
  - The username you use for the credential is only for finding this credentials in `Jenkins` they are not important for something else
- Clear the content of the `Branches to Build` field - this configures `Jenkins` to build all branches if changed

Configure the `GitHub Plugin`:

- In `Build Triggers` select `Build when a change is pushed to GitHub`
- In the `Post-build Actions` section click `Add post-build action -> Set build status on GitHub commit`
- Go back to the general configuration for `Jenkins` (the `Manage Jenkins` settings we opened already earlier) and select `Configure System`
- Under `GitHub Web Hook` select `Let Jenkins auto-manage hook URLs`. This enables you to enter credentials etc.
- In `OAuth token` enter the token you created on the GitHub page for your user earlier

Word of warning
---------------

This guide is to get you started on integrating `Jenkins` with GitHub. You absolutely want to get some professional help for making things secure regarding the keys used and how your `Jenkins` server is available from the outside.


text:
MIT © [Sebastian Raff](https://github.com/hobbyquaker) and homekit2mqtt contributors

[mit-badge]: https://img.shields.io/badge/License-MIT-blue.svg?style=flat
[mit-url]: LICENSE


text:
package com.example.haeunkim.mealtime.viewmodel;

import android.content.Context;
import android.databinding.ObservableField;
import android.support.annotation.NonNull;
import android.text.Editable;
import android.text.TextWatcher;
import android.view.View;

import com.example.haeunkim.mealtime.R;
import com.example.haeunkim.mealtime.model.Auth;
import com.example.haeunkim.mealtime.model.Util;
import com.example.haeunkim.mealtime.view.SignUpActivity;

public class LoginViewModel implements ViewModel {
    private Context context;

    public final ObservableField<String> email = new ObservableField<>();
    public final ObservableField<String> pwd = new ObservableField<>();

    public LoginViewModel(@NonNull Context context) {
        this.context = context;
    }

    @Override
    public void onCreate() {}

    @Override
    public void onPause() {}

    @Override
    public void onResume() {}

    @Override
    public void onDestroy() {}

    // Login button event
    public void onClickLogin(View v) {
        String email = this.email.get();
        String pwd = this.pwd.get();

        if (email == null || pwd == null || email.length() < 1 || pwd.length() < 1) {
            Util.showMessage(context, context.getString(R.string.error_empty));
        } else {
            Auth auth = new Auth();
            auth.signInUser(context, email, pwd);
        }
    }
    // Sign up button event
    public void onClickSignUp(View v) {
        Util.goActivity(context, SignUpActivity.class);
    }

    public TextWatcher emailWatcher = new TextWatcher() {
        @Override
        public void beforeTextChanged(CharSequence s, int start, int count, int after) {}

        @Override
        public void onTextChanged(CharSequence s, int start, int before, int count) {
            email.set(s.toString());
        }

        @Override
        public void afterTextChanged(Editable s) {}
    };

    public TextWatcher pwdWatcher = new TextWatcher() {
        @Override
        public void beforeTextChanged(CharSequence s, int start, int count, int after) {}

        @Override
        public void onTextChanged(CharSequence s, int start, int before, int count) {
            pwd.set(s.toString());
        }

        @Override
        public void afterTextChanged(Editable s) {
        }
    };
}


text:
<?php
	\Magento\Framework\Component\ComponentRegistrar::register(
		\Magento\Framework\Component\ComponentRegistrar::MODULE,
        'Cnx_Developer',
        __DIR__
    );
?>


text:
extern crate removing_extern_crate as foo;
extern crate core;

mod another {
    extern crate removing_extern_crate as foo;
    extern crate core;
}

fn main() {}


text:
SET(CMAKE_DEPENDS_LANGUAGES
  "CXX"
  )
# The set of files for implicit dependencies of each language:
SET(CMAKE_DEPENDS_CHECK_CXX
  "/wg/stor5/mpomarlan/ompl/tests/datastructures/heap.cpp" "/wg/stor5/mpomarlan/ompl/build/tests/CMakeFiles/test_heap.dir/datastructures/heap.cpp.o"
  )
SET(CMAKE_CXX_COMPILER_ID "GNU")

# Preprocessor definitions for this target.
SET(CMAKE_TARGET_DEFINITIONS
  "BOOST_TEST_DYN_LINK"
  "dDOUBLE"
  )

# Targets to which this target links.
SET(CMAKE_TARGET_LINKED_INFO_FILES
  "/wg/stor5/mpomarlan/ompl/build/src/ompl/CMakeFiles/ompl.dir/DependInfo.cmake"
  )


text:
package org.elasticsearch.xpack.core.security.action.role;

import org.elasticsearch.action.support.nodes.BaseNodeRequest;
import org.elasticsearch.action.support.nodes.BaseNodesRequest;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;

import java.io.IOException;

/**
 * The request used to clear the cache for native roles stored in an index.
 */
public class ClearRolesCacheRequest extends BaseNodesRequest<ClearRolesCacheRequest> {

    String[] names;

    public ClearRolesCacheRequest() {
        super((String[]) null);
    }

    public ClearRolesCacheRequest(StreamInput in) throws IOException {
        super(in);
        names = in.readOptionalStringArray();
    }
    /**
     * Sets the roles for which caches will be evicted. When not set all the roles will be evicted from the cache.
     *
     * @param names    The role names
     */
    public ClearRolesCacheRequest names(String... names) {
        this.names = names;
        return this;
    }

    /**
     * @return an array of role names that will have the cache evicted or null if all
     */
    public String[] names() {
        return names;
    }

    @Override
    public void writeTo(StreamOutput out) throws IOException {
        super.writeTo(out);
        out.writeOptionalStringArray(names);
    }

    public static class Node extends BaseNodeRequest {
        private String[] names;

        public Node(StreamInput in) throws IOException {
            super(in);
            names = in.readOptionalStringArray();
        }

        public Node(ClearRolesCacheRequest request) {
            this.names = request.names();
        }

        public String[] getNames() { return names; }

        @Override
        public void writeTo(StreamOutput out) throws IOException {
            super.writeTo(out);
            out.writeOptionalStringArray(names);
        }
    }
}


text:
package org.codehaus.groovy.classgen.asm.sc;

import org.codehaus.groovy.ast.ClassNode;
import org.codehaus.groovy.ast.expr.BitwiseNegationExpression;
import org.codehaus.groovy.ast.expr.EmptyExpression;
import org.codehaus.groovy.ast.expr.Expression;
import org.codehaus.groovy.ast.expr.NotExpression;
import org.codehaus.groovy.ast.expr.UnaryMinusExpression;
import org.codehaus.groovy.ast.expr.UnaryPlusExpression;
import org.codehaus.groovy.classgen.asm.TypeChooser;
import org.codehaus.groovy.classgen.asm.UnaryExpressionHelper;
import org.codehaus.groovy.classgen.asm.WriterController;
import org.objectweb.asm.Label;

import static org.codehaus.groovy.ast.ClassHelper.boolean_TYPE;
import static org.codehaus.groovy.ast.tools.GeneralUtils.bytecodeX;
import static org.codehaus.groovy.ast.ClassHelper.isPrimitiveBoolean;
import static org.codehaus.groovy.ast.ClassHelper.isPrimitiveByte;
import static org.codehaus.groovy.ast.ClassHelper.isPrimitiveChar;
import static org.codehaus.groovy.ast.ClassHelper.isPrimitiveDouble;
import static org.codehaus.groovy.ast.ClassHelper.isPrimitiveFloat;
import static org.codehaus.groovy.ast.ClassHelper.isPrimitiveInt;
import static org.codehaus.groovy.ast.ClassHelper.isPrimitiveLong;
import static org.codehaus.groovy.ast.ClassHelper.isPrimitiveShort;
import static org.objectweb.asm.Opcodes.DNEG;
import static org.objectweb.asm.Opcodes.FNEG;
import static org.objectweb.asm.Opcodes.GOTO;
import static org.objectweb.asm.Opcodes.I2B;
import static org.objectweb.asm.Opcodes.I2C;
import static org.objectweb.asm.Opcodes.I2S;
import static org.objectweb.asm.Opcodes.ICONST_0;
import static org.objectweb.asm.Opcodes.ICONST_1;
import static org.objectweb.asm.Opcodes.ICONST_M1;
import static org.objectweb.asm.Opcodes.IFNE;
import static org.objectweb.asm.Opcodes.INEG;
import static org.objectweb.asm.Opcodes.IXOR;
import static org.objectweb.asm.Opcodes.LNEG;
import static org.objectweb.asm.Opcodes.LXOR;

/**
 * An expression helper which generates optimized bytecode depending on the
 * current type on top of the operand stack.
 */
public class StaticTypesUnaryExpressionHelper extends UnaryExpressionHelper {

    private static final BitwiseNegationExpression EMPTY_BITWISE_NEGATE = new BitwiseNegationExpression(EmptyExpression.INSTANCE);
    private static final UnaryMinusExpression EMPTY_UNARY_MINUS = new UnaryMinusExpression(EmptyExpression.INSTANCE);
    private static final UnaryPlusExpression EMPTY_UNARY_PLUS = new UnaryPlusExpression(EmptyExpression.INSTANCE);

    public StaticTypesUnaryExpressionHelper(final WriterController controller) {
        super(controller);
    }

    @Override
    public void writeBitwiseNegate(final BitwiseNegationExpression expression) {
        expression.getExpression().visit(controller.getAcg());
        ClassNode top = controller.getOperandStack().getTopOperand();
        if (isPrimitiveInt(top) || isPrimitiveLong(top) || isPrimitiveShort(top) || isPrimitiveByte(top) || isPrimitiveChar(top)) {
            bytecodeX(mv -> {
                if (isPrimitiveLong(top)) {
                    mv.visitLdcInsn(-1L);
                    mv.visitInsn(LXOR);
                } else {
                    mv.visitInsn(ICONST_M1);
                    mv.visitInsn(IXOR);
                    if (isPrimitiveByte(top)) {
                        mv.visitInsn(I2B);
                    } else if (isPrimitiveChar(top)) {
                        mv.visitInsn(I2C);
                    } else if (isPrimitiveShort(top)) {
                        mv.visitInsn(I2S);
                    }
                }
            }).visit(controller.getAcg());
            controller.getOperandStack().remove(1);
        } else {
            super.writeBitwiseNegate(EMPTY_BITWISE_NEGATE);
        }
    }

    @Override
    public void writeNotExpression(final NotExpression expression) {
        Expression subExpression = expression.getExpression();
        TypeChooser typeChooser = controller.getTypeChooser();
        if (isPrimitiveBoolean(typeChooser.resolveType(subExpression, controller.getClassNode()))) {
            subExpression.visit(controller.getAcg());
            controller.getOperandStack().doGroovyCast(boolean_TYPE);
            bytecodeX(mv -> {
                Label ne = new Label();
                mv.visitJumpInsn(IFNE, ne);
                mv.visitInsn(ICONST_1);
                Label out = new Label();
                mv.visitJumpInsn(GOTO, out);
                mv.visitLabel(ne);
                mv.visitInsn(ICONST_0);
                mv.visitLabel(out);
            }).visit(controller.getAcg());
            controller.getOperandStack().remove(1);
        } else {
            super.writeNotExpression(expression);
        }
    }

    @Override
    public void writeUnaryMinus(final UnaryMinusExpression expression) {
        expression.getExpression().visit(controller.getAcg());
        ClassNode top = controller.getOperandStack().getTopOperand();
        if (isPrimitiveInt(top) || isPrimitiveLong(top) || isPrimitiveShort(top)|| isPrimitiveFloat(top)
                || isPrimitiveDouble(top) || isPrimitiveByte(top) || isPrimitiveChar(top)) {
            bytecodeX(mv -> {
                if (isPrimitiveInt(top) || isPrimitiveShort(top) || isPrimitiveByte(top) || isPrimitiveChar(top)) {
                    mv.visitInsn(INEG);
                    if (isPrimitiveByte(top)) {
                        mv.visitInsn(I2B);
                    } else if (isPrimitiveChar(top)) {
                        mv.visitInsn(I2C);
                    } else if (isPrimitiveShort(top)) {
                        mv.visitInsn(I2S);
                    }
                } else if (isPrimitiveLong(top)) {
                    mv.visitInsn(LNEG);
                } else if (isPrimitiveFloat(top)) {
                    mv.visitInsn(FNEG);
                } else if (isPrimitiveDouble(top)) {
                    mv.visitInsn(DNEG);
                }
            }).visit(controller.getAcg());
            controller.getOperandStack().remove(1);
        } else {
            super.writeUnaryMinus(EMPTY_UNARY_MINUS);
        }
    }

    @Override
    public void writeUnaryPlus(final UnaryPlusExpression expression) {
        expression.getExpression().visit(controller.getAcg());
        ClassNode top = controller.getOperandStack().getTopOperand();
        if (isPrimitiveInt(top) || isPrimitiveLong(top) || isPrimitiveShort(top)|| isPrimitiveFloat(top)
                || isPrimitiveDouble(top) || isPrimitiveByte(top) || isPrimitiveChar(top)) {
            // only visit the sub-expression
        } else {
            super.writeUnaryPlus(EMPTY_UNARY_PLUS);
        }
    }
}


text:
/**
 * @file   Arc2Test.cpp
 * @brief  Arc2 class tester.
 * @author zer0
 * @date   2020-01-19
 */

#include <gtest/gtest.h>
#include <libtbag/geometry/Arc2.hpp>

using namespace libtbag;
using namespace libtbag::geometry;

TEST(Arc2Test, Default)
{
    Arc2 arc;
}

TEST(Arc2Test, String)
{
    auto const text = Arc2().toString();
    ASSERT_FALSE(text.empty());
    std::cout << "Arc2: " << text << std::endl;
}


text:
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.ComponentModel.DataAnnotations;
using System.Linq;
using System.Threading.Tasks;

namespace Air.Common.Dto
{

    public partial class MessageDTO
    {
        /// <summary>
        /// Initializes a new instance of the Message class.
        /// </summary>
        public MessageDTO() { }

        /// <summary>
        /// Initializes a new instance of the Message class.
        /// </summary>
        public MessageDTO(string id = default(string), string application = default(string), string detail = default(string), string hostname = default(string), string title = default(string), string source = default(string), int? statusCode = default(int?), DateTime? dateTime = default(DateTime?), string type = default(string), string user = default(string), string severity = default(string), string url = default(string), string method = default(string), string version = default(string), IList<DataItem> cookies = default(IList<DataItem>), IList<DataItem> form = default(IList<DataItem>), IList<DataItem> queryString = default(IList<DataItem>), IList<DataItem> serverVariables = default(IList<DataItem>), IList<DataItem> data = default(IList<DataItem>))
        {
            Id = id;
            Application = application;
            Detail = detail;
            Hostname = hostname;
            Title = title;
            Source = source;
            StatusCode = statusCode;
            DateTime = dateTime;
            Type = type;
            User = user;
            Severity = severity;
            Url = url;
            Method = method;
            Version = version;
            Cookies = cookies;
            Form = form;
            QueryString = queryString;
            ServerVariables = serverVariables;
            Data = data;
        }

        /// <summary>
        /// The ID of this message.
        /// </summary>
        [JsonProperty(PropertyName = "id")]
        public string Id { get; set; }

        /// <summary>
        /// Used to identify which application logged this message. You can
        /// use this if you have multiple applications and services logging
        /// to the same log
        /// </summary>
        [JsonProperty(PropertyName = "application")]
        public string Application { get; set; }

        /// <summary>
        /// A longer description of the message. For errors this could be a
        /// stacktrace, but it's really up to you what to log in there.
        /// </summary>
        [JsonProperty(PropertyName = "detail")]
        public string Detail { get; set; }

        /// <summary>
        /// The hostname of the server logging the message.
        /// </summary>
        [JsonProperty(PropertyName = "hostname")]
        public string Hostname { get; set; }

        /// <summary>
        /// The textual title or headline of the message to log.
        /// </summary>
        [JsonProperty(PropertyName = "title")]
        [Required]
        public string Title { get; set; }

        /// <summary>
        /// The source of the code logging the message. This could be the
        /// assembly name.
        /// </summary>
        [JsonProperty(PropertyName = "source")]
        public string Source { get; set; }

        /// <summary>
        /// If the message logged relates to a HTTP status code, you can put
        /// the code in this property. This would probably only be relevant
        /// for errors,
        /// but could be used for logging successful status codes
        /// as well.
        /// </summary>
        [JsonProperty(PropertyName = "statusCode")]
        public int? StatusCode { get; set; }

        /// <summary>
        /// The date and time in UTC of the message. If you don't provide us
        /// with a value in dateTime, we will set the current date and time
        /// in UTC.
        /// </summary>
        [JsonProperty(PropertyName = "dateTime")]
        [Required]
        public DateTime? DateTime { get; set; }

        /// <summary>
        /// The type of message. If logging an error, the type of the
        /// exception would go into type but you can put anything in there,
        /// that makes sense for your domain.
        /// </summary>
        [JsonProperty(PropertyName = "type")]
        public string Type { get; set; }

        /// <summary>
        /// An identification of the user triggering this message. You can put
        /// the users email address or your user key into this property.
        /// </summary>
        [JsonProperty(PropertyName = "user")]
        public string User { get; set; }

        /// <summary>
        /// An enum value representing the severity of this message. The
        /// following values are allowed: Verbose, Debug, Information,
        /// Warning, Error, Fatal
        /// </summary>
        [JsonProperty(PropertyName = "severity")]
        public string Severity { get; set; }

        /// <summary>
        /// If message relates to a HTTP request, you may send the URL of that
        /// request. If you don't provide us with an URL, we will try to find
        /// a key named URL in serverVariables.
        /// </summary>
        [JsonProperty(PropertyName = "url")]
        public string Url { get; set; }

        /// <summary>
        /// If message relates to a HTTP request, you may send the HTTP method
        /// of that request. If you don't provide us with a method, we will
        /// try to find a key named REQUEST_METHOD in serverVariables.
        /// </summary>
        [JsonProperty(PropertyName = "method")]
        public string Method { get; set; }

        /// <summary>
        /// Versions can be used to distinguish messages from different
        /// versions of your software. The value of version can be a SemVer
        /// compliant string or any other
        /// syntax that you are using as your version numbering
        /// scheme.
        /// </summary>
        [JsonProperty(PropertyName = "version")]
        public string Version { get; set; }

        /// <summary>
        /// A key/value pair of cookies. This property only makes sense for
        /// logging messages related to web requests.
        /// </summary>
        [JsonProperty(PropertyName = "cookies")]
        public IList<DataItem> Cookies { get; set; }

        /// <summary>
        /// A key/value pair of form fields and their values. This property
        /// makes sense if logging message related to users inputting data in
        /// a form.
        /// </summary>
        [JsonProperty(PropertyName = "form")]
        public IList<DataItem> Form { get; set; }

        /// <summary>
        /// A key/value pair of query string parameters. This property makes
        /// sense if logging message related to a HTTP request.
        /// </summary>
        [JsonProperty(PropertyName = "queryString")]
        public IList<DataItem> QueryString { get; set; }

        /// <summary>
        /// A key/value pair of server values. Server variables are typically
        /// related to handling requests in a webserver but could be used for
        /// other types of information as well.
        /// </summary>
        [JsonProperty(PropertyName = "serverVariables")]
        public IList<DataItem> ServerVariables { get; set; }

        /// <summary>
        /// A key/value pair of user-defined fields and their values. When
        /// logging an exception, the Data dictionary of
        /// the exception is copied to this property. You can add
        /// additional key/value pairs, by modifying the Data
        /// dictionary on the exception or by supplying additional
        /// key/values to this API.
        /// </summary>
        [JsonProperty(PropertyName = "data")]
        public IList<DataItem> Data { get; set; }

        public override string ToString()
        {
            return $"[{DateTime}][{Severity}] - {Title}";
        }

    }

    public partial class DataItem
    {
        /// <summary>
        /// Initializes a new instance of the Item class.
        /// </summary>
        public DataItem() { }

        /// <summary>
        /// Initializes a new instance of the Item class.
        /// </summary>
        public DataItem(string key = default(string), string value = default(string))
        {
            Key = key;
            Value = value;
        }

        /// <summary>
        /// </summary>
        [JsonProperty(PropertyName = "key")]
        public string Key { get; set; }

        /// <summary>
        /// </summary>
        [JsonProperty(PropertyName = "value")]
        public string Value { get; set; }

    }

}


text:
-- Update TermsAndConditionsAcceptedOn for all users created afer 28th July 2021
	 Update U
     Set U.TermAndConditionsAcceptedOn = T.CreatedDate
	 From [employer_account].[User] U
     Inner join (
			Select UserId, CreatedDate from (
				Select *, ROW_NUMBER() over (partition by M.UserId order by M.CreatedDate) As Rank2
				from [employer_account].[Membership] M
			) Result
			Where Result.Rank2 = 1
			and CreatedDate >= CONVERT(DATETIME,'28/07/2021',103) --dd/MM/yyyy
		) T on U.Id = T.UserId