/*
(non-Javadoc)

@see org.activiti.engine.impl.history.HistoryManagerInterface#recordTaskEnd (java.lang.String, java.lang.String)
182507
@Override
  public void recordTaskEnd(String taskId, String deleteReason) {
    if (isHistoryLevelAtLeast(HistoryLevel.AUDIT)) {
      HistoricTaskInstanceEntity historicTaskInstance = getHistoricTaskInstanceEntityManager().findById(taskId);
      if (historicTaskInstance != null) {
        historicTaskInstance.markEnded(deleteReason);
      }
    }
  }
789559
@Override public RecordWriter<K,VariantContextWritable> getRecordWriter(
			TaskAttemptContext ctx)
		throws IOException
	{
		Configuration conf = ctx.getConfiguration();
		boolean isCompressed = getCompressOutput(ctx);
		CompressionCodec codec = null;
		String extension = "";
		if (isCompressed) {
			Class<? extends CompressionCodec> codecClass =
					getOutputCompressorClass(ctx, BGZFCodec.class);
			codec = ReflectionUtils.newInstance(codecClass, conf);
			extension = codec.getDefaultExtension();
		}
		Path file = getDefaultWorkFile(ctx, extension);
		if (!isCompressed) {
			return getRecordWriter(ctx, file);
		} else {
			FileSystem fs = file.getFileSystem(conf);
			return getRecordWriter(ctx, codec.createOutputStream(fs.create(file)));
		}
	}
723040
@Override
	public void collect(T record)
	{
		this.delegate.setInstance(record);
		try {
			for (int i = 0; i < writers.length; i++) {
				this.writers[i].emit(this.delegate);
			}
		}
		catch (IOException e) {
			throw new RuntimeException("Emitting the record caused an I/O exception: " + e.getMessage(), e);
		}
		catch (InterruptedException e) {
			throw new RuntimeException("Emitting the record was interrupted: " + e.getMessage(), e);
		}
	}
1344055
public void recordSlowTask(String trackerName) {
    synchronized (this) {
      NodeUsageReport usageReport = getReportUnprotected(trackerName);
      usageReport.setNumSlow(usageReport.getNumSlow() + 1);
    }
  }
1102548
@Override
  public void recordTaskProcessDefinitionChange(String taskId, String processDefinitionId) {
    if (isHistoryLevelAtLeast(HistoryLevel.ACTIVITY)) {
      HistoricTaskInstanceEntity historicTaskInstance = getHistoricTaskInstanceEntityManager().findById(taskId);
      if (historicTaskInstance != null) {
        historicTaskInstance.setProcessDefinitionId(processDefinitionId);
      }
    }
  }
487721
@Override
  public RecordWriter<K, V> getRecordWriter(TaskAttemptContext context)
      throws IOException, InterruptedException {
    Configuration conf = context.getConfiguration();
    return getDelegate(conf).getRecordWriter(context);
  }
1213572
private void writeRecurringTask(RecurringTask record) throws IOException
   {
      m_buffer.setLength(0);

      m_buffer.append(MPXConstants.RECURRING_TASK_RECORD_NUMBER);
      m_buffer.append(m_delimiter);
      m_buffer.append("1");

      if (record.getRecurrenceType() != null)
      {
         boolean monthlyRelative = record.getRecurrenceType() == RecurrenceType.MONTHLY && record.getRelative();
         boolean monthlyAbsolute = record.getRecurrenceType() == RecurrenceType.MONTHLY && !record.getRelative();
         boolean yearlyRelative = record.getRecurrenceType() == RecurrenceType.YEARLY && record.getRelative();
         boolean yearlyAbsolute = record.getRecurrenceType() == RecurrenceType.YEARLY && !record.getRelative();

         m_buffer.append(m_delimiter);
         m_buffer.append(format(formatDateTime(record.getStartDate())));
         m_buffer.append(m_delimiter);
         m_buffer.append(format(formatDateTime(record.getFinishDate())));
         m_buffer.append(m_delimiter);
         m_buffer.append(format(RecurrenceUtility.getDurationValue(m_projectFile.getProjectProperties(), record.getDuration())));
         m_buffer.append(m_delimiter);
         m_buffer.append(format(RecurrenceUtility.getDurationUnits(record)));
         m_buffer.append(m_delimiter);
         m_buffer.append(format(record.getOccurrences()));
         m_buffer.append(m_delimiter);
         m_buffer.append(format(RecurrenceUtility.getRecurrenceValue(record.getRecurrenceType())));
         m_buffer.append(m_delimiter);
         m_buffer.append("0");
         m_buffer.append(m_delimiter);
         m_buffer.append(record.getUseEndDate() ? "1" : "0");
         m_buffer.append(m_delimiter);
         m_buffer.append(record.isWorkingDaysOnly() ? "1" : "0");
         m_buffer.append(m_delimiter);
         m_buffer.append(format(RecurrenceUtility.getDays(record)));
         m_buffer.append(m_delimiter);
         m_buffer.append(monthlyRelative ? "1" : "0");
         m_buffer.append(m_delimiter);
         m_buffer.append(yearlyAbsolute ? "1" : "0");
         m_buffer.append(m_delimiter);
         m_buffer.append(format(record.getRecurrenceType() == RecurrenceType.DAILY ? record.getFrequency() : "1"));
         m_buffer.append(m_delimiter);
         m_buffer.append(format(record.getRecurrenceType() == RecurrenceType.WEEKLY ? record.getFrequency() : "1"));
         m_buffer.append(m_delimiter);
         m_buffer.append(format(monthlyRelative ? record.getDayNumber() : "1"));
         m_buffer.append(m_delimiter);
         m_buffer.append(format(RecurrenceUtility.getDay(monthlyRelative ? record.getDayOfWeek() : Day.MONDAY)));
         m_buffer.append(m_delimiter);
         m_buffer.append(format(monthlyRelative ? record.getFrequency() : "1"));
         m_buffer.append(m_delimiter);
         m_buffer.append(format(monthlyAbsolute ? record.getDayNumber() : "1"));
         m_buffer.append(m_delimiter);
         m_buffer.append(format(monthlyAbsolute ? record.getFrequency() : "1"));
         m_buffer.append(m_delimiter);
         m_buffer.append(format(yearlyRelative ? record.getDayNumber() : "1"));
         m_buffer.append(m_delimiter);
         m_buffer.append(format(RecurrenceUtility.getDay(yearlyRelative ? record.getDayOfWeek() : Day.MONDAY)));
         m_buffer.append(m_delimiter);
         m_buffer.append(format(record.getMonthNumber()));
         m_buffer.append(m_delimiter);
         m_buffer.append(format(formatDateTime(RecurrenceUtility.getYearlyAbsoluteAsDate(record))));

         stripTrailingDelimiters(m_buffer);
      }
      m_buffer.append(MPXConstants.EOL);

      m_writer.write(m_buffer.toString());
   }
624050
public void logTaskUpdates(TaskID taskId, long finishTime) {
    if (disableHistory) {
      return;
    }

    JobID id = taskId.getJobID();
    if (!this.jobId.equals(id)) {
      throw new RuntimeException("JobId from task: " + id +
                                 " does not match expected: " + jobId);
    }

    if (null != writers) {
      log(writers, RecordTypes.Task,
          new Keys[]{Keys.TASKID, Keys.FINISH_TIME},
          new String[]{ taskId.toString(),
                        String.valueOf(finishTime)});
    }
  }
641890
public void completeTask(int taskId) {
		getResourceFactory().getApiResource("/task/" + taskId + "/complete")
				.entity(new Empty(), MediaType.APPLICATION_JSON_TYPE).post();
	}
711826
@Override
    public RecordWriter<NullWritable, WARCWritable> getRecordWriter(TaskAttemptContext context)
            throws IOException, InterruptedException {
        return new WARCWriter(context);
    }
1113056
private void populateRecurringTask(Record record, RecurringTask task) throws MPXJException
   {
      //System.out.println(record);
      task.setStartDate(record.getDateTime(1));
      task.setFinishDate(record.getDateTime(2));
      task.setDuration(RecurrenceUtility.getDuration(m_projectFile.getProjectProperties(), record.getInteger(3), record.getInteger(4)));
      task.setOccurrences(record.getInteger(5));
      task.setRecurrenceType(RecurrenceUtility.getRecurrenceType(record.getInteger(6)));
      task.setUseEndDate(NumberHelper.getInt(record.getInteger(8)) == 1);
      task.setWorkingDaysOnly(NumberHelper.getInt(record.getInteger(9)) == 1);
      task.setWeeklyDaysFromBitmap(RecurrenceUtility.getDays(record.getString(10)), RecurrenceUtility.RECURRING_TASK_DAY_MASKS);

      RecurrenceType type = task.getRecurrenceType();
      if (type != null)
      {
         switch (task.getRecurrenceType())
         {
            case DAILY:
            {
               task.setFrequency(record.getInteger(13));
               break;
            }

            case WEEKLY:
            {
               task.setFrequency(record.getInteger(14));
               break;
            }

            case MONTHLY:
            {
               task.setRelative(NumberHelper.getInt(record.getInteger(11)) == 1);
               if (task.getRelative())
               {
                  task.setFrequency(record.getInteger(17));
                  task.setDayNumber(record.getInteger(15));
                  task.setDayOfWeek(RecurrenceUtility.getDay(record.getInteger(16)));
               }
               else
               {
                  task.setFrequency(record.getInteger(19));
                  task.setDayNumber(record.getInteger(18));
               }
               break;
            }

            case YEARLY:
            {
               task.setRelative(NumberHelper.getInt(record.getInteger(12)) != 1);
               if (task.getRelative())
               {
                  task.setDayNumber(record.getInteger(20));
                  task.setDayOfWeek(RecurrenceUtility.getDay(record.getInteger(21)));
                  task.setMonthNumber(record.getInteger(22));
               }
               else
               {
                  task.setYearlyAbsoluteFromDate(record.getDateTime(23));
               }
               break;
            }
         }
      }

      //System.out.println(task);
   }
28990
public void record(
    TaskInProgress tip, String host, long inputBytes) {
    synchronized (localityRecords) {
      localityRecords.add(new Record(tip, host, inputBytes));
      localityRecords.notify();
    }
  }
151789
@SuppressWarnings("UnusedDeclaration")
  public void recordChange( StateEditable change )
  {
    beginUndoAtom();
    try
    {
      StateEdit transaction = new StateEdit( change );
      transaction.end();
      addEdit( transaction );
    }
    finally
    {
      endUndoAtom();
    }
  }
180773
@SuppressWarnings("deprecation")
  private boolean completedTaskUnprotected(TaskInProgress tip, TaskStatus status,
      TaskTrackerInfo ttStatus) {
    int oldNumAttempts = tip.getActiveTasks().size();
    // Metering
    meterTaskAttemptUnprotected(tip, status);

    // It _is_ safe to not decrement running{Map|Reduce}Tasks and
    // finished{Map|Reduce}Tasks variables here because one and only
    // one task-attempt of a TIP gets to completedTask. This is because
    // the TaskCommitThread in the JobTracker marks other, completed,
    // speculative tasks as _complete_.
    TaskAttemptID taskId = status.getTaskID();
    if (tip.isComplete()) {
      // Mark this task as KILLED
      tip.alreadyCompletedTask(taskId);
      return false;
    }

    LOG.info("Task '" + taskId + "' has completed " + tip.getTIPId() +
           " successfully.");
    // Mark the TIP as complete
    tip.completed(taskId);

    // Update jobhistory
    String taskType = getTaskType(tip);
    if (status.getIsMap()){
      jobHistory.logMapTaskStarted(status.getTaskID(), status.getStartTime(),
                                       status.getTaskTracker(),
                                       ttStatus.getHttpPort(),
                                       taskType);
      jobHistory.logMapTaskFinished(status.getTaskID(), status.getFinishTime(),
                                        ttStatus.getHost(), taskType,
                                        status.getStateString(),
                                        status.getCounters());
    }else{
      jobHistory.logReduceTaskStarted( status.getTaskID(), status.getStartTime(),
                                          status.getTaskTracker(),
                                          ttStatus.getHttpPort(),
                                          taskType);
      jobHistory.logReduceTaskFinished(status.getTaskID(), status.getShuffleFinishTime(),
                                           status.getSortFinishTime(), status.getFinishTime(),
                                           ttStatus.getHost(),
                                           taskType,
                                           status.getStateString(),
                                           status.getCounters());
    }
    jobHistory.logTaskFinished(tip.getTIPId(),
                                taskType,
                                tip.getExecFinishTime(),
                                status.getCounters());

    int newNumAttempts = tip.getActiveTasks().size();
    if (tip.isJobSetupTask()) {
      // setup task has finished. kill the extra setup tip
      killSetupTipUnprotected(!tip.isMapTask());
      setupCompleteUnprotected();
    } else if (tip.isJobCleanupTask()) {
      // cleanup task has finished. Kill the extra cleanup tip
      if (tip.isMapTask()) {
        // kill the reduce tip
        cleanup[1].kill();
      } else {
        cleanup[0].kill();
      }

      if (jobFailed) {
        terminateJob(JobStatus.FAILED);
      } else if (jobKilled) {
        terminateJob(JobStatus.KILLED);
      } else {
        jobCompleteUnprotected();
      }
    } else if (tip.isMapTask()) {
      runningMapTasks--;
      // Update locality counters.
      long inputBytes = tip.getCounters()
                .getGroup("org.apache.hadoop.mapred.Task$Counter")
                .getCounter("Map input bytes");
      jobStats.incTotalMapInputBytes(inputBytes);
      localityStats.record(tip, ttStatus.getHost(), inputBytes);
      // check if this was a speculative task.
      if (oldNumAttempts > 1) {
        speculativeMapTasks -= (oldNumAttempts - newNumAttempts);
        jobStats.incNumSpeculativeSucceededMaps();
      }
      finishedMapTasks += 1;
      jobStats.incNumMapTasksCompleted();
      if (!tip.isJobSetupTask() && hasSpeculativeMaps) {
        updateTaskTrackerStats(tip,ttStatus,trackerMapStats,mapTaskStats);
      }
      // remove the completed map from the resp running caches
      retireMapUnprotected(tip);
      if ((finishedMapTasks + failedMapTIPs) == (numMapTasks)) {
        this.status.setMapProgress(1.0f);
      }
    } else {
      runningReduceTasks -= 1;
      if (oldNumAttempts > 1) {
        speculativeReduceTasks -= (oldNumAttempts - newNumAttempts);
        jobStats.incNumSpeculativeSucceededReduces();
      }
      finishedReduceTasks += 1;
      jobStats.incNumReduceTasksCompleted();
      if (!tip.isJobSetupTask() && hasSpeculativeReduces) {
        updateTaskTrackerStats(tip,ttStatus,trackerReduceStats,reduceTaskStats);
      }
      // remove the completed reduces from the running reducers set
      retireReduceUnprotected(tip);
      if ((finishedReduceTasks + failedReduceTIPs) == (numReduceTasks)) {
        this.status.setReduceProgress(1.0f);
      }
    }

    // is job complete?
    if (!jobSetupCleanupNeeded && canLaunchJobCleanupTaskUnprotected()) {
      jobCompleteUnprotected();
    }

    return true;
  }
1251628
@Override
  public void recordProcessInstanceEnd(String processInstanceId, String deleteReason, String activityId) {

    if (isHistoryLevelAtLeast(HistoryLevel.ACTIVITY)) {
      HistoricProcessInstanceEntity historicProcessInstance = getHistoricProcessInstanceEntityManager().findById(processInstanceId);

      if (historicProcessInstance != null) {
        historicProcessInstance.markEnded(deleteReason);
        historicProcessInstance.setEndActivityId(activityId);
        
        // Fire event
        ActivitiEventDispatcher activitiEventDispatcher = getEventDispatcher();
        if (activitiEventDispatcher != null && activitiEventDispatcher.isEnabled()) {
          activitiEventDispatcher.dispatchEvent(
              ActivitiEventBuilder.createEntityEvent(ActivitiEventType.HISTORIC_PROCESS_INSTANCE_ENDED, historicProcessInstance));
        }
        
      }
    }
  }
286086
public void markTaskCompletion() {
    if (this.countDownLatch.isPresent()) {
      this.countDownLatch.get().countDown();
    }

    this.taskState.setProp(ConfigurationKeys.TASK_RETRIES_KEY, this.retryCount.get());
  }
208551
@Override
  public void recordActivityEnd(ExecutionEntity executionEntity, String deleteReason) {
    if (isHistoryLevelAtLeast(HistoryLevel.ACTIVITY)) {
      HistoricActivityInstanceEntity historicActivityInstance = findActivityInstance(executionEntity, false, true);
      if (historicActivityInstance != null) {
        historicActivityInstance.markEnded(deleteReason);
        
        // Fire event
        ActivitiEventDispatcher activitiEventDispatcher = getEventDispatcher();
        if (activitiEventDispatcher != null && activitiEventDispatcher.isEnabled()) {
          activitiEventDispatcher.dispatchEvent(
              ActivitiEventBuilder.createEntityEvent(ActivitiEventType.HISTORIC_ACTIVITY_INSTANCE_ENDED, historicActivityInstance));
        }
      }
    }
  }