This method publishes output data for a single task based on the given {@link WorkUnitState}.
Output data from other tasks won't be published even if they are in the same folder.
553918
private void publishSingleTaskData(WorkUnitState state, int branchId)
      throws IOException {
    publishData(state, branchId, true, new HashSet<Path>());
    addLineageInfo(state, branchId);
  }
904889
private void publishCheckpointEvent(String stepName, long jobInstanceId, long jobExecutionId, long stepExecutionId) {
        BatchEventsPublisher publisher = getBatchEventsPublisher();

        if (publisher != null) {
            String correlationId = runtimeWorkUnitExecution.getCorrelationId();
            publisher.publishCheckpointEvent(stepName, jobInstanceId, jobExecutionId, stepExecutionId, correlationId);
        }
    }
452202
private void markOutputDirSuccessful(JobContext context)
  throws IOException {
    if (outputPath != null) {
      FileSystem fileSys = outputPath.getFileSystem(context.getConfiguration());
      if (fileSys.exists(outputPath)) {
        // create a file in the folder to mark it
        Path filePath = new Path(outputPath, SUCCEEDED_FILE_NAME);
        fileSys.create(filePath).close();
      }
    }
  }
181208
@Override
  public synchronized void write(DataOutput out) throws IOException {
    out.writeDouble(getCpuTotalGHz());
    out.writeDouble(getMemTotalGB());
    out.writeInt(getNumCpu());
    out.writeInt(getNumRunningJobs());
    out.writeInt(getNumTaskTrackers());
    out.writeDouble(getMemUsageGB());
    out.writeDouble(getCpuUsageGHz());
  }
923528
public void write(DataOutput out) throws IOException {
    Text.writeString(out, jobFile);
    taskId.write(out);
    out.writeInt(partition);
    out.writeInt(numSlotsRequired);
    taskStatus.write(out);
    skipRanges.write(out);
    out.writeBoolean(skipping);
    out.writeBoolean(jobCleanup);
    if (jobCleanup) {
      WritableUtils.writeEnum(out, jobRunStateForCleanup);
    }
    out.writeBoolean(jobSetup);
    Text.writeString(out, username);
    out.writeBoolean(writeSkipRecs);
    out.writeBoolean(taskCleanup);  
    extraData.write(out);
  }
787956
@Override
    public AnalysisResultFuture run(final AnalysisJob job) throws UnsupportedOperationException {
        logger.info("Validating distributed job: {}", job);

        failIfJobIsUnsupported(job);

        final InjectionManager injectionManager =
                _configuration.getEnvironment().getInjectionManagerFactory().getInjectionManager(_configuration, job);
        final LifeCycleHelper lifeCycleHelper = new LifeCycleHelper(injectionManager, true);
        final RowProcessingPublishers publishers = getRowProcessingPublishers(job, lifeCycleHelper);
        final RowProcessingPublisher publisher = getRowProcessingPublisher(publishers);
        publisher.initializeConsumers(new TaskListener() {
            @Override
            public void onError(final Task task, final Throwable throwable) {
                logger.error("Failed to initialize consumers at master node!", throwable);
            }

            @Override
            public void onComplete(final Task task) {
            }

            @Override
            public void onBegin(final Task task) {
            }
        });

        logger.info("Validation passed! Chunking job for distribution amongst slaves: {}", job);

        // since we always use a SingleThreadedTaskRunner, the above operation
        // will be synchronized/blocking.

        final AnalysisJobMetrics analysisJobMetrics = publishers.getAnalysisJobMetrics();
        _analysisListener.jobBegin(job, analysisJobMetrics);

        final RowProcessingMetrics rowProcessingMetrics = publisher.getRowProcessingMetrics();
        _analysisListener.rowProcessingBegin(job, rowProcessingMetrics);

        final AnalysisResultFuture resultFuture;

        try {
            final int expectedRows = rowProcessingMetrics.getExpectedRows();
            if (expectedRows == 0) {
                logger.info("Expected rows of the job was zero. Job will run on a local virtual slave.");

                // when there are no expected rows we still need to build a
                // single slave job, but run it locally, since the job lifecycle
                // still needs to be guaranteed.
                final DistributedJobContext context = new DistributedJobContextImpl(_configuration, job, 0, 1);

                // use a virtual cluster, which just runs the job locally.
                final VirtualClusterManager localCluster = new VirtualClusterManager(_configuration, 1);
                resultFuture = localCluster.dispatchJob(job, context);
            } else {
                final JobDivisionManager jobDivisionManager = _clusterManager.getJobDivisionManager();
                final int chunks = jobDivisionManager.calculateDivisionCount(job, expectedRows);
                final int rowsPerChunk = (expectedRows + 1) / chunks;

                logger.info(
                        "Expected rows was {}. A total number of {} slave jobs will be built, each of approx. {} rows.",
                        expectedRows, chunks, rowsPerChunk);

                final List<AnalysisResultFuture> results = dispatchJobs(job, chunks, rowsPerChunk, publisher);
                final DistributedAnalysisResultReducer reducer =
                        new DistributedAnalysisResultReducer(job, lifeCycleHelper, publisher, _analysisListener);
                resultFuture = new DistributedAnalysisResultFuture(results, reducer);
            }

        } catch (final RuntimeException e) {
            _analysisListener.errorUnknown(job, e);
            throw e;
        }

        if (!_analysisListener.isEmpty()) {
            awaitAndInformListener(job, analysisJobMetrics, rowProcessingMetrics, resultFuture);
        }

        return resultFuture;
    }
275706
@Override
	public CompletableFuture<Acknowledge> submitTask(
			TaskDeploymentDescriptor tdd,
			JobMasterId jobMasterId,
			Time timeout) {

		try {
			final JobID jobId = tdd.getJobId();
			final JobManagerConnection jobManagerConnection = jobManagerTable.get(jobId);

			if (jobManagerConnection == null) {
				final String message = "Could not submit task because there is no JobManager " +
					"associated for the job " + jobId + '.';

				log.debug(message);
				throw new TaskSubmissionException(message);
			}

			if (!Objects.equals(jobManagerConnection.getJobMasterId(), jobMasterId)) {
				final String message = "Rejecting the task submission because the job manager leader id " +
					jobMasterId + " does not match the expected job manager leader id " +
					jobManagerConnection.getJobMasterId() + '.';

				log.debug(message);
				throw new TaskSubmissionException(message);
			}

			if (!taskSlotTable.tryMarkSlotActive(jobId, tdd.getAllocationId())) {
				final String message = "No task slot allocated for job ID " + jobId +
					" and allocation ID " + tdd.getAllocationId() + '.';
				log.debug(message);
				throw new TaskSubmissionException(message);
			}

			// re-integrate offloaded data:
			try {
				tdd.loadBigData(blobCacheService.getPermanentBlobService());
			} catch (IOException | ClassNotFoundException e) {
				throw new TaskSubmissionException("Could not re-integrate offloaded TaskDeploymentDescriptor data.", e);
			}

			// deserialize the pre-serialized information
			final JobInformation jobInformation;
			final TaskInformation taskInformation;
			try {
				jobInformation = tdd.getSerializedJobInformation().deserializeValue(getClass().getClassLoader());
				taskInformation = tdd.getSerializedTaskInformation().deserializeValue(getClass().getClassLoader());
			} catch (IOException | ClassNotFoundException e) {
				throw new TaskSubmissionException("Could not deserialize the job or task information.", e);
			}

			if (!jobId.equals(jobInformation.getJobId())) {
				throw new TaskSubmissionException(
					"Inconsistent job ID information inside TaskDeploymentDescriptor (" +
						tdd.getJobId() + " vs. " + jobInformation.getJobId() + ")");
			}

			TaskMetricGroup taskMetricGroup = taskManagerMetricGroup.addTaskForJob(
				jobInformation.getJobId(),
				jobInformation.getJobName(),
				taskInformation.getJobVertexId(),
				tdd.getExecutionAttemptId(),
				taskInformation.getTaskName(),
				tdd.getSubtaskIndex(),
				tdd.getAttemptNumber());

			InputSplitProvider inputSplitProvider = new RpcInputSplitProvider(
				jobManagerConnection.getJobManagerGateway(),
				taskInformation.getJobVertexId(),
				tdd.getExecutionAttemptId(),
				taskManagerConfiguration.getTimeout());

			TaskManagerActions taskManagerActions = jobManagerConnection.getTaskManagerActions();
			CheckpointResponder checkpointResponder = jobManagerConnection.getCheckpointResponder();
			GlobalAggregateManager aggregateManager = jobManagerConnection.getGlobalAggregateManager();

			LibraryCacheManager libraryCache = jobManagerConnection.getLibraryCacheManager();
			ResultPartitionConsumableNotifier resultPartitionConsumableNotifier = jobManagerConnection.getResultPartitionConsumableNotifier();
			PartitionProducerStateChecker partitionStateChecker = jobManagerConnection.getPartitionStateChecker();

			final TaskLocalStateStore localStateStore = localStateStoresManager.localStateStoreForSubtask(
				jobId,
				tdd.getAllocationId(),
				taskInformation.getJobVertexId(),
				tdd.getSubtaskIndex());

			final JobManagerTaskRestore taskRestore = tdd.getTaskRestore();

			final TaskStateManager taskStateManager = new TaskStateManagerImpl(
				jobId,
				tdd.getExecutionAttemptId(),
				localStateStore,
				taskRestore,
				checkpointResponder);

			Task task = new Task(
				jobInformation,
				taskInformation,
				tdd.getExecutionAttemptId(),
				tdd.getAllocationId(),
				tdd.getSubtaskIndex(),
				tdd.getAttemptNumber(),
				tdd.getProducedPartitions(),
				tdd.getInputGates(),
				tdd.getTargetSlotNumber(),
				taskExecutorServices.getMemoryManager(),
				taskExecutorServices.getIOManager(),
				taskExecutorServices.getNetworkEnvironment(),
				taskExecutorServices.getKvStateService(),
				taskExecutorServices.getBroadcastVariableManager(),
				taskExecutorServices.getTaskEventDispatcher(),
				taskStateManager,
				taskManagerActions,
				inputSplitProvider,
				checkpointResponder,
				aggregateManager,
				blobCacheService,
				libraryCache,
				fileCache,
				taskManagerConfiguration,
				taskMetricGroup,
				resultPartitionConsumableNotifier,
				partitionStateChecker,
				getRpcService().getExecutor());

			log.info("Received task {}.", task.getTaskInfo().getTaskNameWithSubtasks());

			boolean taskAdded;

			try {
				taskAdded = taskSlotTable.addTask(task);
			} catch (SlotNotFoundException | SlotNotActiveException e) {
				throw new TaskSubmissionException("Could not submit task.", e);
			}

			if (taskAdded) {
				task.startTaskThread();

				return CompletableFuture.completedFuture(Acknowledge.get());
			} else {
				final String message = "TaskManager already contains a task for id " +
					task.getExecutionId() + '.';

				log.debug(message);
				throw new TaskSubmissionException(message);
			}
		} catch (TaskSubmissionException e) {
			return FutureUtils.completedExceptionally(e);
		}
	}
170980
public void publish(final List<StepImplementationsDescriptor> stepimplementationDescriptors) {
        final Map<String, List<StepDescriptor>> sectionSorted = new TreeMap<String, List<StepDescriptor>>();

        for (final StepImplementationsDescriptor descriptor : stepimplementationDescriptors) {

            for (final StepDescriptor stepTag : descriptor.getExpressions()) {

                String section = stepTag.getSection();
                if (section == null || section.isEmpty()) {
                    section = "Miscellaneous";
                }

                List<StepDescriptor> subList = sectionSorted.get(section);

                if (subList == null) {
                    subList = new ArrayList<StepDescriptor>();
                    sectionSorted.put(section, subList);
                }
                subList.add(stepTag);
            }
        }

        final String html = buildHtml(sectionSorted);

        outputFile.delete();

        // write out
        try {
            if (outputFile.createNewFile()) {
                Files.write(html, outputFile, Charset.defaultCharset());
            } else {
                // TODO
            }
        } catch (final IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
471626
public void sendOutState(State<Serializable, Serializable> state,
                           String checkpointId,
                           boolean spillState,
                           String location) {
    outputter.sendOutState(state, checkpointId, spillState, location);
  }
1095399
private void submitTaskStateEvents(JobState jobState) {
    // Build Job Metadata applicable for TaskStates
    ImmutableMap.Builder<String, String> jobMetadataBuilder = new ImmutableMap.Builder<>();
    jobMetadataBuilder.put(METADATA_JOB_ID, jobState.getJobId());
    jobMetadataBuilder.put(METADATA_JOB_NAME, jobState.getJobName());
    jobMetadataBuilder.put(METADATA_JOB_TRACKING_URL, jobState.getTrackingURL().or(UNKNOWN_VALUE));
    Map<String, String> jobMetadata = jobMetadataBuilder.build();

    // Submit event for each TaskState
    for (TaskState taskState : jobState.getTaskStates()) {
      submitTaskStateEvent(taskState, jobMetadata);
    }
  }
230643
def publish_state(self, state):
        
        message = json.dumps({'state': {'reported': state}})
        self.client.publish(self.topic, message)
        self._state = state
611911
long tryToGetOutputSize(TaskAttemptID taskId, JobConf conf) {

    try{
      TaskInProgress tip;
      synchronized(this) {
        tip = tasks.get(taskId);
      }
      if(tip == null)
         return -1;

      if (!tip.getTask().isMapTask() ||
          tip.getRunState() != TaskStatus.State.SUCCEEDED) {
        return -1;
      }

      MapOutputFile mapOutputFile = new MapOutputFile();
      mapOutputFile.setJobId(taskId.getJobID());
      mapOutputFile.setConf(conf);

      // In simulation mode, maps/reduces complete instantly and don't produce
      // any output
      if (this.simulatedTaskMode) {
        return 0;
      }

      Path tmp_output = null;
      try {
        tmp_output =  mapOutputFile.getOutputFile(taskId);
      } catch (DiskErrorException dex) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Error getting map output of a task " + taskId, dex);
        }
      }
      if(tmp_output == null)
        return 0;
      FileSystem localFS = FileSystem.getLocal(conf);
      FileStatus stat = localFS.getFileStatus(tmp_output);
      if(stat == null)
        return 0;
      else
        return stat.getLen();
    } catch(IOException e) {
      LOG.info(e);
      return -1;
    }
  }
609113
@Override
	public void triggerPartitionProducerStateCheck(
		JobID jobId,
		final IntermediateDataSetID intermediateDataSetId,
		final ResultPartitionID resultPartitionId) {

		CompletableFuture<ExecutionState> futurePartitionState =
			partitionProducerStateChecker.requestPartitionProducerState(
				jobId,
				intermediateDataSetId,
				resultPartitionId);

		futurePartitionState.whenCompleteAsync(
			(ExecutionState executionState, Throwable throwable) -> {
				try {
					if (executionState != null) {
						onPartitionStateUpdate(
							intermediateDataSetId,
							resultPartitionId,
							executionState);
					} else if (throwable instanceof TimeoutException) {
						// our request timed out, assume we're still running and try again
						onPartitionStateUpdate(
							intermediateDataSetId,
							resultPartitionId,
							ExecutionState.RUNNING);
					} else if (throwable instanceof PartitionProducerDisposedException) {
						String msg = String.format("Producer %s of partition %s disposed. Cancelling execution.",
							resultPartitionId.getProducerId(), resultPartitionId.getPartitionId());
						LOG.info(msg, throwable);
						cancelExecution();
					} else {
						failExternally(throwable);
					}
				} catch (IOException | InterruptedException e) {
					failExternally(e);
				}
			},
			executor);
	}
122949
public synchronized void write(DataOutput out) throws IOException {
    jobid.write(out);
    out.writeFloat(setupProgress);
    out.writeFloat(mapProgress);
    out.writeFloat(reduceProgress);
    out.writeFloat(cleanupProgress);
    out.writeInt(runState);
    out.writeLong(startTime);
    Text.writeString(out, user);
    WritableUtils.writeEnum(out, priority);
    Text.writeString(out, schedulingInfo);
  }
903803
private void publishFileSet(CopyEntity.DatasetAndPartition datasetAndPartition,
      Collection<WorkUnitState> datasetWorkUnitStates) throws IOException {
    Map<String, String> additionalMetadata = Maps.newHashMap();

    Preconditions.checkArgument(!datasetWorkUnitStates.isEmpty(),
        "publishFileSet received an empty collection work units. This is an error in code.");

    CopyableDatasetMetadata metadata = CopyableDatasetMetadata
        .deserialize(datasetWorkUnitStates.iterator().next().getProp(CopySource.SERIALIZED_COPYABLE_DATASET));
    Path datasetWriterOutputPath = new Path(this.writerOutputDir, datasetAndPartition.identifier());

    log.info("Merging all split work units.");
    DistcpFileSplitter.mergeAllSplitWorkUnits(this.fs, datasetWorkUnitStates);

    log.info(String.format("[%s] Publishing fileSet from %s for dataset %s", datasetAndPartition.identifier(),
        datasetWriterOutputPath, metadata.getDatasetURN()));

    List<CommitStep> prePublish = getCommitSequence(datasetWorkUnitStates, PrePublishStep.class);
    List<CommitStep> postPublish = getCommitSequence(datasetWorkUnitStates, PostPublishStep.class);
    log.info(String.format("[%s] Found %d prePublish steps and %d postPublish steps.", datasetAndPartition.identifier(),
        prePublish.size(), postPublish.size()));

    executeCommitSequence(prePublish);
    if (hasCopyableFiles(datasetWorkUnitStates)) {
      // Targets are always absolute, so we start moving from root (will skip any existing directories).
      HadoopUtils.renameRecursively(this.fs, datasetWriterOutputPath, new Path("/"));
    } else {
      log.info(String.format("[%s] No copyable files in dataset. Proceeding to postpublish steps.", datasetAndPartition.identifier()));
    }
    executeCommitSequence(postPublish);

    this.fs.delete(datasetWriterOutputPath, true);

    long datasetOriginTimestamp = Long.MAX_VALUE;
    long datasetUpstreamTimestamp = Long.MAX_VALUE;
    Optional<String> fileSetRoot = Optional.<String>absent();

    for (WorkUnitState wus : datasetWorkUnitStates) {
      if (wus.getWorkingState() == WorkingState.SUCCESSFUL) {
        wus.setWorkingState(WorkUnitState.WorkingState.COMMITTED);
      }
      CopyEntity copyEntity = CopySource.deserializeCopyEntity(wus);
      if (copyEntity instanceof CopyableFile) {
        CopyableFile copyableFile = (CopyableFile) copyEntity;
        DataFileVersionStrategy srcVS = this.srcDataFileVersionStrategy;
        DataFileVersionStrategy dstVS = this.dstDataFileVersionStrategy;

        // Prefer to use copyableFile's specific version strategy
        if (copyableFile.getDataFileVersionStrategy() != null) {
          Config versionStrategyConfig = ConfigFactory.parseMap(ImmutableMap.of(
              DataFileVersionStrategy.DATA_FILE_VERSION_STRATEGY_KEY, copyableFile.getDataFileVersionStrategy()));
          srcVS = DataFileVersionStrategy.instantiateDataFileVersionStrategy(this.srcFs, versionStrategyConfig);
          dstVS = DataFileVersionStrategy.instantiateDataFileVersionStrategy(this.fs, versionStrategyConfig);
        }

        if (copyableFile.getPreserve().preserve(PreserveAttributes.Option.VERSION)
            && dstVS.hasCharacteristic(DataFileVersionStrategy.Characteristic.SETTABLE)) {
          dstVS.setVersion(copyableFile.getDestination(),
              srcVS.getVersion(copyableFile.getOrigin().getPath()));
        }

        if (wus.getWorkingState() == WorkingState.COMMITTED) {
          CopyEventSubmitterHelper.submitSuccessfulFilePublish(this.eventSubmitter, copyableFile, wus);
          // Dataset Output path is injected in each copyableFile.
          // This can be optimized by having a dataset level equivalent class for copyable entities
          // and storing dataset related information, e.g. dataset output path, there.

          // Currently datasetOutputPath is only present for hive datasets.
          if (!fileSetRoot.isPresent() && copyableFile.getDatasetOutputPath() != null) {
            fileSetRoot = Optional.of(copyableFile.getDatasetOutputPath());
          }
          if (lineageInfo.isPresent()) {
            lineageInfo.get().putDestination(copyableFile.getDestinationData(), 0, wus);
          }
        }
        if (datasetOriginTimestamp > copyableFile.getOriginTimestamp()) {
          datasetOriginTimestamp = copyableFile.getOriginTimestamp();
        }
        if (datasetUpstreamTimestamp > copyableFile.getUpstreamTimestamp()) {
          datasetUpstreamTimestamp = copyableFile.getUpstreamTimestamp();
        }
      }
    }

    // if there are no valid values for datasetOriginTimestamp and datasetUpstreamTimestamp, use
    // something more readable
    if (Long.MAX_VALUE == datasetOriginTimestamp) {
      datasetOriginTimestamp = 0;
    }
    if (Long.MAX_VALUE == datasetUpstreamTimestamp) {
      datasetUpstreamTimestamp = 0;
    }

    additionalMetadata.put(SlaEventKeys.SOURCE_URI, this.state.getProp(SlaEventKeys.SOURCE_URI));
    additionalMetadata.put(SlaEventKeys.DESTINATION_URI, this.state.getProp(SlaEventKeys.DESTINATION_URI));
    additionalMetadata.put(SlaEventKeys.DATASET_OUTPUT_PATH, fileSetRoot.or("Unknown"));
    CopyEventSubmitterHelper.submitSuccessfulDatasetPublish(this.eventSubmitter, datasetAndPartition,
        Long.toString(datasetOriginTimestamp), Long.toString(datasetUpstreamTimestamp), additionalMetadata);
    }
33362
public void commit() {
    boolean isTaskFailed = false;

    try {
      // Check if all forks succeeded
      List<Integer> failedForkIds = new ArrayList<>();
      for (Optional<Fork> fork : this.forks.keySet()) {
        if (fork.isPresent()) {
          if (fork.get().isSucceeded()) {
            if (!fork.get().commit()) {
              failedForkIds.add(fork.get().getIndex());
            }
          } else {
            failedForkIds.add(fork.get().getIndex());
          }
        }
      }

      if (failedForkIds.size() == 0) {
        // Set the task state to SUCCESSFUL. The state is not set to COMMITTED
        // as the data publisher will do that upon successful data publishing.
        if (this.taskState.getWorkingState() != WorkUnitState.WorkingState.FAILED) {
          this.taskState.setWorkingState(WorkUnitState.WorkingState.SUCCESSFUL);
        }
      } else {
        ForkThrowableHolder holder = Task.getForkThrowableHolder(this.taskState.getTaskBroker());
        LOG.info("Holder for this task {} is {}", this.taskId, holder);
        if (!holder.isEmpty()) {
          failTask(holder.getAggregatedException(failedForkIds, this.taskId));
        } else {
          // just in case there are some corner cases where Fork throw an exception but doesn't add into holder
          failTask(new ForkException("Fork branches " + failedForkIds + " failed for task " + this.taskId));
        }
      }
    } catch (Throwable t) {
      failTask(t);
      isTaskFailed = true;
    } finally {
      addConstructsFinalStateToTaskState(extractor, converter, rowChecker);

      this.taskState.setProp(ConfigurationKeys.WRITER_RECORDS_WRITTEN, getRecordsWritten());
      this.taskState.setProp(ConfigurationKeys.WRITER_BYTES_WRITTEN, getBytesWritten());

      this.submitTaskCommittedEvent();

      try {
        closer.close();
      } catch (Throwable t) {
        LOG.error("Failed to close all open resources", t);
        if ((!isIgnoreCloseFailures) && (!isTaskFailed)) {
          LOG.error("Setting the task state to failed.");
          failTask(t);
        }
      }

      for (Map.Entry<Optional<Fork>, Optional<Future<?>>> forkAndFuture : this.forks.entrySet()) {
        if (forkAndFuture.getKey().isPresent() && forkAndFuture.getValue().isPresent()) {
          try {
            forkAndFuture.getValue().get().cancel(true);
          } catch (Throwable t) {
            LOG.error(String.format("Failed to cancel Fork \"%s\"", forkAndFuture.getKey().get()), t);
          }
        }
      }

      try {
        if (shouldPublishDataInTask()) {
          // If data should be published by the task, publish the data and set the task state to COMMITTED.
          // Task data can only be published after all forks have been closed by closer.close().
          if (this.taskState.getWorkingState() == WorkUnitState.WorkingState.SUCCESSFUL) {
            publishTaskData();
            this.taskState.setWorkingState(WorkUnitState.WorkingState.COMMITTED);
          }
        }
      } catch (IOException ioe) {
        failTask(ioe);
      } finally {
        long endTime = System.currentTimeMillis();
        this.taskState.setEndTime(endTime);
        this.taskState.setTaskDuration(endTime - startTime);
        this.taskStateTracker.onTaskCommitCompletion(this);
      }
    }
  }
897025
protected void publishState(State from, State to) throws Exception {
    final Collection<StateListener> listeners = getStateListeners();
    synchronized (listeners) {
      for (StateListener listener : listeners) {
        listener.stateChanged(from, to);
      }
    }
  }