private Optional<StepAction> getAction(WorkflowSummary summary, String stepId) {
    Initiator initiator = summary.getInitiator();
    List<UpstreamInitiator.Info> path = new ArrayList<>();
    StringBuilder sqlBuilder = new StringBuilder(GET_ACTION_QUERY);
    if (initiator instanceof UpstreamInitiator) {
      path.addAll(((UpstreamInitiator) initiator).getAncestors());
      sqlBuilder
          .append(" OR (")
          .append(String.join(") OR (", Collections.nCopies(path.size(), CONDITION_POSTFIX)))
          .append(')');
    }
    String sql = sqlBuilder.toString();

    UpstreamInitiator.Info self = new UpstreamInitiator.Info();
    self.setWorkflowId(summary.getWorkflowId());
    self.setInstanceId(summary.getWorkflowInstanceId());
    self.setRunId(summary.getWorkflowRunId());
    self.setStepId(stepId);
    path.add(self);

    List<StepAction> allActions =
        withMetricLogError(
            () ->
                withRetryableQuery(
                    sql,
                    stmt -> {
                      int idx = 0;
                      for (UpstreamInitiator.Info node : path) {
                        stmt.setString(++idx, node.getWorkflowId());
                        stmt.setLong(++idx, node.getInstanceId());
                        stmt.setLong(++idx, node.getRunId());
                        stmt.setString(++idx, node.getStepId());
                      }
                    },
                    result -> {
                      List<StepAction> actions = new ArrayList<>();
                      while (result.next()) {
                        StepAction res =
                            fromJson(result.getString(PAYLOAD_COLUMN), StepAction.class);
                        res.setCreateTime(result.getTimestamp("create_ts").getTime());
                        actions.add(res);
                      }
                      return actions;
                    }),
            "getActions",
            "Failed to get the actions for step {}[{}] with parents",
            summary.getIdentity(),
            stepId);

    AtomicInteger ordinal = new AtomicInteger(0);
    Map<UpstreamInitiator.Info, Integer> orders =
        path.stream()
            .collect(Collectors.toMap(Function.identity(), p -> ordinal.incrementAndGet()));

    allActions.sort(
        (a1, a2) ->
            Integer.compare(
                orders.getOrDefault(a2.toInfo(), 0), orders.getOrDefault(a1.toInfo(), 0)));

    // pick the final action from all of them.
    StepAction stepAction = null;
    for (StepAction action : allActions) {
      if (action.getAction() != null && action.getAction().isUsingUpstream()) {
        stepAction = action;
        break;
      } else if (System.currentTimeMillis() - action.getCreateTime() < ACTION_TIMEOUT
          && action.getWorkflowId().equals(self.getWorkflowId())
          && action.getWorkflowInstanceId() == self.getInstanceId()
          && action.getWorkflowRunId() == self.getRunId()
          && action.getStepId().equals(self.getStepId())) {
        stepAction = action;
      }
    }
    if (stepAction != null) {
      LOG.info(
          "Pick a pending action [{}] for step {}[{}] among total [{}] pending actions: [{}]",
          stepAction,
          summary.getIdentity(),
          stepId,
          allActions.size(),
          allActions);
    }
    return Optional.ofNullable(stepAction);
  }
@Test
  public void testGetAction() {
    WorkflowSummary summary = new WorkflowSummary();
    summary.setWorkflowId("sample-dag-test-3");
    summary.setWorkflowInstanceId(1);
    summary.setWorkflowRunId(1);
    Optional<StepAction> stepAction = actionDao.tryGetAction(summary, "job1");
    Assert.assertFalse(stepAction.isPresent());
    testRestartDirectly();
    stepAction = actionDao.tryGetAction(summary, "job1");
    Assert.assertTrue(stepAction.isPresent());
    stepAction.ifPresent(
        action -> {
          Assert.assertEquals("sample-dag-test-3", action.getWorkflowId());
          Assert.assertEquals(1, action.getWorkflowInstanceId());
          Assert.assertEquals(1, action.getWorkflowRunId());
          Assert.assertEquals("job1", action.getStepId());
          Assert.assertEquals(RESTART, action.getAction());
          Assert.assertEquals(user, action.getUser());
          Assert.assertEquals(
              Collections.singletonMap("foo", ParamDefinition.buildParamDefinition("foo", "bar")),
              action.getRunParams());
        });
    Mockito.verify(publisher, Mockito.times(1)).publish(any(StepInstanceWakeUpEvent.class));
  }