For a given commit, will traverse the pipeline and find the time it entered in each stage of the pipeline
@param commit
@param dashboard
@param pipeline
@return
181996
private PipelineResponseCommit applyStageTimestamps(PipelineResponseCommit commit, Dashboard dashboard, Pipeline pipeline,List<PipelineStage> pipelineStageList){
        PipelineResponseCommit returnCommit = new PipelineResponseCommit(commit);

        for(PipelineStage systemStage : pipelineStageList) {
            //get commits for a given stage
            Map<String, PipelineCommit> commitMap = findCommitsForStage(dashboard, pipeline, systemStage);

            //if this commit doesnt have a processed timestamp for this stage, add one
            PipelineCommit pipelineCommit = commitMap.get(commit.getScmRevisionNumber());
            if(pipelineCommit != null && !returnCommit.getProcessedTimestamps().containsKey(systemStage.getName())){
                Long timestamp = pipelineCommit.getTimestamp();
                returnCommit.addNewPipelineProcessedTimestamp(systemStage, timestamp);
            }
        }
        return returnCommit;
    }
547327
public Pipeline findById(Long pipelineId) {

        Assert.assertNotNull(pipelineId);
        List<Pipeline> pipeline = listByIds(pipelineId);

        if (pipeline.size() != 1) {
            String exceptionCause = "query pipeline by pipelineId:" + pipelineId + " but return " + pipeline.size()
                                    + " pipeline!";
            logger.error("ERROR ## " + exceptionCause);
            throw new ManagerException(exceptionCause);
        }
        return pipeline.get(0);
    }
1247547
public GitlabCommitComparison compareCommits(Serializable projectId, String commitHash1, String commitHash2, Pagination pagination) throws IOException {
        String tailUrl = GitlabProject.URL + "/" + sanitizeProjectId(projectId) + GitlabCommitComparison.URL;
        Query query = new Query()
                .append("from", commitHash1)
                .append("to", commitHash2);
        query.mergeWith(pagination.asQuery());
        return retrieve().to(tailUrl + query, GitlabCommitComparison.class);
    }
676039
public Optional<CommitLog> commitAndGetLogs() throws InvalidKBException {
        if (isClosed()) {
            return Optional.empty();
        }
        try {
            return commitWithLogs();
        } finally {
            String closeMessage = ErrorMessage.TX_CLOSED_ON_ACTION.getMessage("committed", keyspace());
            closeTransaction(closeMessage);
        }
    }
385787
def for_commit(commit, cmd_options={})
      [].tap do |returning|
        data = hg(["diff -c ?", commit.hash_id], cmd_options)
        chunks = data.split(/^diff/)[1..-1]
        unless chunks.nil?
          chunks.map do |piece| 
            piece = "diff" << piece
            returning << build(piece)
          end
        end
      end
    end
550302
private void commitOnePhase(Xid xid) throws XAException {
    try {
      // Check preconditions
      if (currentXid == null || !currentXid.equals(xid)) {
        // In fact, we don't know if xid is bogus, or if it just wasn't
        // associated with this
        // connection.
        // Assume it's our fault.
        throw new CloudSpannerXAException(CloudSpannerXAException.ONE_PHASE_SAME,
            Code.UNIMPLEMENTED, XAException.XAER_RMERR);
      }
      if (state != STATE_ENDED) {
        throw new CloudSpannerXAException(CloudSpannerXAException.COMMIT_BEFORE_END,
            Code.FAILED_PRECONDITION, XAException.XAER_PROTO);
      }

      // Preconditions are met. Commit
      state = STATE_IDLE;
      currentXid = null;

      conn.commit();
      conn.setAutoCommit(localAutoCommitMode);
    } catch (CloudSpannerSQLException ex) {
      throw new CloudSpannerXAException(CloudSpannerXAException.ERROR_ONE_PHASE, ex,
          XAException.XAER_RMERR);
    } catch (SQLException ex) {
      throw new CloudSpannerXAException(CloudSpannerXAException.ERROR_ONE_PHASE, ex, Code.UNKNOWN,
          XAException.XAER_RMERR);
    }
  }
288706
public void waitForPermit(Long pipelineId) throws InterruptedException {
        Assert.notNull(pipelineId);

        PermitMonitor permitMonitor = ArbitrateFactory.getInstance(pipelineId, PermitMonitor.class);
        permitMonitor.waitForPermit();// 阻塞等待授权
    }
19682
private boolean initProcessStage(Long processId, List<String> currentStages) {
        Collections.sort(currentStages, new StageComparator());
        if (logger.isDebugEnabled()) {
            logger.debug("pipeline[{}] processId[{}] with stage{}", new Object[] { getPipelineId(), processId,
                    currentStages });
        }

        this.currentStages.put(processId, currentStages);// 更新下stage数据
        stageChanged(processId, currentStages);// 通知stage变化
        // 2.1 判断是否存在了loaded节点，此节点为最后一个节点
        if (currentStages.contains(ArbitrateConstants.NODE_TRANSFORMED)) {
            return true;// 不需要监听了
        } else {
            return false;// 其他的状态需要监听
        }
    }
157899
def log(duration, message=None, use_last_commit_message=False):
    
    branch = git.branch
    issue = jira.get_issue(branch)
    # Create the comment
    comment = "Working on issue %s" % branch
    if message:
        comment = message
    elif use_last_commit_message:
        comment = git.get_last_commit_message()

    if issue:
        # If the duration is provided use it, otherwise use the elapsed time since the last mark
        duration = jira.get_elapsed_time(issue) if duration == '.' else duration

        if duration:
            # Add the worklog
            jira.add_worklog(issue, timeSpent=duration, adjustEstimate=None, newEstimate=None, reduceBy=None,
                             comment=comment)

            print "Logged %s against issue %s (%s)" % (duration, branch, comment)
        else:
            print "No time logged, less than 0m elapsed."
408784
protected Long performCommitLogic(EDBCommit commit) throws EDBException {
        if (!(commit instanceof JPACommit)) {
            throw new EDBException("The given commit type is not supported.");
        }
        if (commit.isCommitted()) {
            throw new EDBException("EDBCommit is already commitet.");
        }
        if (revisionCheckEnabled && commit.getParentRevisionNumber() != null
                && !commit.getParentRevisionNumber().equals(getCurrentRevisionNumber())) {
            throw new EDBException("EDBCommit do not have the correct head revision number.");
        }
        runBeginCommitHooks(commit);
        EDBException exception = runPreCommitHooks(commit);
        if (exception != null) {
            return runErrorHooks(commit, exception);
        }
        Long timestamp = performCommit((JPACommit) commit);
        runEDBPostHooks(commit);

        return timestamp;
    }
538864
private static void addTimePerComponent(HashMap<String, Long> mapComponentTimes,	Component component) {
    Long currentTimeOfComponent = 0L;
    String key = component.getComponentType();
    if (mapComponentTimes.containsKey(key)) {
      currentTimeOfComponent = mapComponentTimes.get(key);
    }
    //when transactions are run in parallel, we should log the longest transaction only to avoid that 
    //for ex 'Total time' would be 100ms and transactions in parallel to hornetQ will be 2000ms 
    Long maxTime =  Math.max(component.getTime(), currentTimeOfComponent);
    mapComponentTimes.put(key, maxTime);
  }
409459
public boolean check(Long pipelineId) {
        MainstemMonitor mainstemMonitor = ArbitrateFactory.getInstance(pipelineId, MainstemMonitor.class);
        return mainstemMonitor.check();
    }
1054569
@Override
    public CompletableFuture<Void> completeCommittingTransactions(VersionedMetadata<CommittingTransactionsRecord> record) {
        // create all transaction entries in committing txn list. 
        // remove all entries from active txn in epoch. 
        // reset CommittingTxnRecord
        long time = System.currentTimeMillis();

        Map<String, byte[]> completedRecords = record.getObject().getTransactionsToCommit().stream()
                                                               .collect(Collectors.toMap(UUID::toString, 
                                                                       x -> new CompletedTxnRecord(time, TxnStatus.COMMITTED).toBytes()));
        CompletableFuture<Void> future;
        if (record.getObject().getTransactionsToCommit().size() == 0) {
            future = CompletableFuture.completedFuture(null);
        } else {
            future = createCompletedTxEntries(completedRecords)
                    .thenCompose(x -> getTransactionsInEpochTable(record.getObject().getEpoch())
                            .thenCompose(table -> storeHelper.removeEntries(table, completedRecords.keySet())))
                    .thenCompose(x -> tryRemoveOlderTransactionsInEpochTables(epoch -> epoch < record.getObject().getEpoch()));
        }
        return future
                .thenCompose(x -> Futures.toVoid(updateCommittingTxnRecord(new VersionedMetadata<>(CommittingTransactionsRecord.EMPTY,
                        record.getVersion()))));
    }
530529
public QueryBuilder withCommitId(BigDecimal commitId) {
        Validate.argumentIsNotNull(commitId);
        queryParamsBuilder.commitId(CommitId.valueOf(commitId));
        return this;
    }
1185066
public void importGit(ProgressMonitor... monitors) throws IOException {
        if (inProgress)
            throw new IOException("Asset already in progress...");

        try {
            inProgress = true;
            Props props = new Props(this);
            VcInfo vcInfo = new VcInfo(getGitRoot(), props);

            getOut().println("Importing from Git into: " + getProjectDir() + "...");

            // CLI dependencies
            Git git = new Git(getReleasesUrl(), vcInfo, "checkVersionConsistency", vcInfo.getBranch(), getAssetLoc());
            git.run(monitors);

            // Check Asset inconsistencies
            Vercheck vercheck = new Vercheck();
            vercheck.setConfigLoc(getConfigLoc());
            vercheck.setAssetLoc(getAssetLoc());
            vercheck.setGitRoot(getGitRoot());
            vercheck.setDebug(true);
            vercheck.run();
            if (vercheck.getErrorCount() > 0) {
                throw new IOException("Asset version conflict(s).  See log for details");
            }

            // perform import (Git pull)
            git = new Git(getReleasesUrl(), vcInfo, "hardCheckout", vcInfo.getBranch(), isHardReset());
            git.run(monitors);

            // capture new Refs in ASSET_REF after import (Git pull)
            DbInfo dbInfo = new DbInfo(props);
            Checkpoint checkpoint = new Checkpoint(getReleasesUrl(), vcInfo, getAssetRoot(), dbInfo);
            try {
                checkpoint.run(monitors).updateRefs();
            }
            catch (SQLException ex) {
                throw new IOException(ex.getMessage(), ex);
            }
        }
        catch (Throwable ex) {
            if (ex instanceof IOException)
                throw ex;
            else
                throw new IOException(ex.getMessage(), ex);
        }
        finally {
            inProgress = false;
        }
    }
125790
protected void logHeuristic(boolean commit) throws SystemException {
        if (tc.isEntryEnabled())
            Tr.entry(tc, "logHeuristic", commit);

        if (_subordinate && _status.getState() != TransactionState.STATE_PREPARING) {
            getResources().logHeuristic(commit);
        }

        if (tc.isEntryEnabled())
            Tr.exit(tc, "logHeuristic");
    }
1178289
@Override
	public StepExecutionImpl createStepExecution(long rootJobExecId, StepContextImpl stepContext) {
		StepExecutionImpl stepExecution = null;
		String batchStatus = stepContext.getBatchStatus() == null ? BatchStatus.STARTING.name() : stepContext.getBatchStatus().name();
		String exitStatus = stepContext.getExitStatus();
		String stepName = stepContext.getStepName();
		if (logger.isLoggable(Level.FINE)) {
			logger.fine("batchStatus: " + batchStatus + " | stepName: " + stepName);
		}
		long readCount = 0;
		long writeCount = 0;
		long commitCount = 0;
		long rollbackCount = 0;
		long readSkipCount = 0;
		long processSkipCount = 0;
		long filterCount = 0;
		long writeSkipCount = 0;
		Timestamp startTime = stepContext.getStartTimeTS();
		Timestamp endTime = stepContext.getEndTimeTS();

		Metric[] metrics = stepContext.getMetrics();
		for (int i = 0; i < metrics.length; i++) {
			if (metrics[i].getType().equals(MetricImpl.MetricType.READ_COUNT)) {
				readCount = metrics[i].getValue();
			} else if (metrics[i].getType().equals(MetricImpl.MetricType.WRITE_COUNT)) {
				writeCount = metrics[i].getValue();	
			} else if (metrics[i].getType().equals(MetricImpl.MetricType.PROCESS_SKIP_COUNT)) {
				processSkipCount = metrics[i].getValue();	
			} else if (metrics[i].getType().equals(MetricImpl.MetricType.COMMIT_COUNT)) {
				commitCount = metrics[i].getValue();
			} else if (metrics[i].getType().equals(MetricImpl.MetricType.ROLLBACK_COUNT)) {
				rollbackCount = metrics[i].getValue();	
			} else if (metrics[i].getType().equals(MetricImpl.MetricType.READ_SKIP_COUNT)) {
				readSkipCount = metrics[i].getValue();
			} else if (metrics[i].getType().equals(MetricImpl.MetricType.FILTER_COUNT)) {
				filterCount = metrics[i].getValue();	
			} else if (metrics[i].getType().equals(MetricImpl.MetricType.WRITE_SKIP_COUNT)) {
				writeSkipCount = metrics[i].getValue();	
			}		
		}
		Serializable persistentData = stepContext.getPersistentUserData();

		stepExecution = createStepExecution(rootJobExecId, batchStatus, exitStatus, stepName, readCount, 
				writeCount, commitCount, rollbackCount, readSkipCount, processSkipCount, filterCount, writeSkipCount, startTime,
				endTime, persistentData);

		return stepExecution;
	}