@VisibleForTesting
  boolean isStepSkipped(WorkflowSummary workflowSummary, StepRuntimeSummary runtimeSummary) {
    RestartConfig stepRestartConfig =
        ObjectHelper.valueOrDefault(
            runtimeSummary.getRestartConfig(), workflowSummary.getRestartConfig());
    if (stepRestartConfig != null
        && stepRestartConfig.getRestartPath() != null
        && stepRestartConfig.getRestartPath().size() == 1
        && stepRestartConfig.getSkipSteps() != null) {
      boolean skipped = stepRestartConfig.getSkipSteps().contains(runtimeSummary.getStepId());
      if (skipped) {
        LOG.info(
            "workflow {}'s step {} is skipped.",
            workflowSummary.getIdentity(),
            runtimeSummary.getIdentity());
        runtimeSummary.markTerminated(StepInstance.Status.SKIPPED, tracingManager);
        runtimeSummary.addTimeline(
            TimelineLogEvent.info("Step is skipped because of a user skip action."));
      }
      return skipped;
    }
    return false;
  }
@Test
  public void testIsStepSkipped() {
    WorkflowSummary summary = new WorkflowSummary();
    summary.setWorkflowId("test-workflow");
    summary.setWorkflowInstanceId(1L);
    summary.setWorkflowRunId(1L);

    StepRuntimeState runtimeState = new StepRuntimeState();
    runtimeState.setStatus(StepInstance.Status.USER_FAILED);
    Timeline timeline = new Timeline(new ArrayList<>());
    StepInstance.StepRetry stepRetry = new StepInstance.StepRetry();
    stepRetry.setRetryable(true);
    StepRuntimeSummary runtimeSummary =
        StepRuntimeSummary.builder()
            .stepId("step1")
            .timeline(timeline)
            .runtimeState(runtimeState)
            .stepRetry(stepRetry)
            .build();
    Assert.assertFalse(maestroTask.isStepSkipped(summary, runtimeSummary));

    summary.setRestartConfig(RestartConfig.builder().build());
    Assert.assertFalse(maestroTask.isStepSkipped(summary, runtimeSummary));

    summary.setRestartConfig(
        RestartConfig.builder()
            .addRestartNode("test-workflow", 1, "step1")
            .addRestartNode("test-workflow", 2, "step1")
            .build());
    Assert.assertFalse(maestroTask.isStepSkipped(summary, runtimeSummary));

    summary.setRestartConfig(
        RestartConfig.builder()
            .addRestartNode("test-workflow", 1, "step1")
            .restartPolicy(RunPolicy.RESTART_FROM_SPECIFIC)
            .build());
    Assert.assertFalse(maestroTask.isStepSkipped(summary, runtimeSummary));

    summary.setRestartConfig(
        RestartConfig.builder()
            .addRestartNode("test-workflow", 1, "step1")
            .restartPolicy(RunPolicy.RESTART_FROM_SPECIFIC)
            .skipSteps(Collections.singleton("step2"))
            .stepRestartParams(Collections.singletonMap("step2", Collections.emptyMap()))
            .build());
    Assert.assertFalse(maestroTask.isStepSkipped(summary, runtimeSummary));

    summary.setRestartConfig(
        RestartConfig.builder()
            .addRestartNode("test-workflow", 1, "step1")
            .restartPolicy(RunPolicy.RESTART_FROM_SPECIFIC)
            .skipSteps(Collections.singleton("step1"))
            .stepRestartParams(Collections.singletonMap("step1", Collections.emptyMap()))
            .build());
    Assert.assertTrue(maestroTask.isStepSkipped(summary, runtimeSummary));
    assertThat(timeline.getTimelineEvents())
        .hasSize(1)
        .usingRecursiveFieldByFieldElementComparatorIgnoringFields("timestamp")
        .contains(TimelineLogEvent.info("Step is skipped because of a user skip action."));
    Assert.assertTrue(stepRetry.isRetryable());
    Assert.assertEquals(DbOperation.UPSERT, runtimeSummary.getDbOperation());
    Assert.assertEquals(StepInstance.Status.SKIPPED, runtimeState.getStatus());
    assertThat(runtimeSummary.getPendingRecords())
        .hasSize(1)
        .usingRecursiveFieldByFieldElementComparatorIgnoringFields("eventTime")
        .contains(
            StepInstanceUpdateJobEvent.createRecord(
                StepInstance.Status.USER_FAILED, StepInstance.Status.SKIPPED, 0L));

    summary.setRestartConfig(null);
    runtimeSummary.getPendingRecords().clear();
    runtimeState.setStatus(StepInstance.Status.USER_FAILED);
    runtimeSummary.setRestartConfig(
        RestartConfig.builder()
            .addRestartNode("test-workflow", 1, "step1")
            .restartPolicy(RunPolicy.RESTART_FROM_SPECIFIC)
            .skipSteps(Collections.singleton("step1"))
            .stepRestartParams(Collections.singletonMap("step1", Collections.emptyMap()))
            .build());
    Assert.assertTrue(maestroTask.isStepSkipped(summary, runtimeSummary));
    assertThat(timeline.getTimelineEvents())
        .hasSize(1)
        .usingRecursiveFieldByFieldElementComparatorIgnoringFields("timestamp")
        .contains(TimelineLogEvent.info("Step is skipped because of a user skip action."));
    Assert.assertTrue(stepRetry.isRetryable());
    Assert.assertEquals(DbOperation.UPSERT, runtimeSummary.getDbOperation());
    Assert.assertEquals(StepInstance.Status.SKIPPED, runtimeState.getStatus());
    assertThat(runtimeSummary.getPendingRecords())
        .hasSize(1)
        .usingRecursiveFieldByFieldElementComparatorIgnoringFields("eventTime")
        .contains(
            StepInstanceUpdateJobEvent.createRecord(
                StepInstance.Status.USER_FAILED, StepInstance.Status.SKIPPED, 0L));
  }