@JsonIgnore
  public List<Tag> deriveRuntimeTagPermits(StepRuntimeSummary runtimeSummary) {
    List<Tag> runtimeTagPermits = new ArrayList<>();
    Long stepConcurrency = runProperties.getStepConcurrency();
    if (instanceStepConcurrency != null) { // instance_step_concurrency is enabled.
      // if step_concurrency is set, then set a tag permit for it
      if (stepConcurrency != null) {
        runtimeTagPermits.add(buildATag(workflowId, stepConcurrency));
      }
      // only if instance_step_concurrency is set for a leaf step, then add a tag permit for it.
      if (runtimeSummary.getType().isLeaf()) {
        runtimeTagPermits.add(buildATag(correlationId, instanceStepConcurrency));
      }
    } else { // instance_step_concurrency is disabled.
      // if step_concurrency is unset, use the default, then set a tag permit for it
      if (stepConcurrency == null) {
        stepConcurrency = Defaults.DEFAULT_STEP_CONCURRENCY;
      }
      runtimeTagPermits.add(buildATag(workflowId, stepConcurrency));
      // always add a default step level step concurrency limit with the default value.
      runtimeTagPermits.add(
          buildATag(
              workflowId + ":" + runtimeSummary.getStepId(), Defaults.DEFAULT_STEP_CONCURRENCY));
    }

    return runtimeTagPermits;
  }
@Test
  public void testDeriveRuntimeTagPermits() throws Exception {
    WorkflowSummary summary =
        loadObject("fixtures/parameters/sample-wf-summary-params.json", WorkflowSummary.class);
    summary.setCorrelationId("correlation_id");

    Tag t1 = new Tag();
    t1.setName(Constants.MAESTRO_PREFIX + summary.getWorkflowId());
    t1.setPermit((int) Defaults.DEFAULT_STEP_CONCURRENCY);
    Tag t2 = new Tag();
    t2.setName(Constants.MAESTRO_PREFIX + summary.getWorkflowId() + ":stepid");
    t2.setPermit((int) Defaults.DEFAULT_STEP_CONCURRENCY);
    Tag t3 = new Tag();
    t3.setName(Constants.MAESTRO_PREFIX + summary.getCorrelationId());
    t3.setPermit((int) Defaults.DEFAULT_INSTANCE_STEP_CONCURRENCY);

    StepRuntimeSummary runtimeSummary =
        StepRuntimeSummary.builder().stepId("stepid").type(StepType.NOOP).build();

    // set default (workflow level and step level) permits if no concurrences are set
    assertEquals(Arrays.asList(t1, t2), summary.deriveRuntimeTagPermits(runtimeSummary));

    // set permit for workflow level and step level step_concurrency
    summary.getRunProperties().setStepConcurrency(10L);
    t1.setPermit(10);
    assertEquals(Arrays.asList(t1, t2), summary.deriveRuntimeTagPermits(runtimeSummary));

    // set permit for instance_step_concurrency only when workflow level and step level
    // step_concurrency are not set
    summary.getRunProperties().setStepConcurrency(null);
    summary.setInstanceStepConcurrency(15L);
    t3.setPermit(15);
    assertEquals(Collections.singletonList(t3), summary.deriveRuntimeTagPermits(runtimeSummary));

    // set tag permits for both step_concurrency and instance_step_concurrency
    summary.getRunProperties().setStepConcurrency(10L);
    summary.setInstanceStepConcurrency(15L);
    t3.setPermit(15);
    assertEquals(Arrays.asList(t1, t3), summary.deriveRuntimeTagPermits(runtimeSummary));

    // don't set tag permits for instance_step_concurrency if non-leaf steps
    runtimeSummary = StepRuntimeSummary.builder().stepId("stepid").type(StepType.FOREACH).build();
    summary.getRunProperties().setStepConcurrency(10L);
    summary.setInstanceStepConcurrency(15L);
    assertEquals(Collections.singletonList(t1), summary.deriveRuntimeTagPermits(runtimeSummary));

    // don't set any tag permit if no step_concurrency set for non-leaf steps
    summary.getRunProperties().setStepConcurrency(null);
    summary.setInstanceStepConcurrency(15L);
    assertEquals(Collections.emptyList(), summary.deriveRuntimeTagPermits(runtimeSummary));
  }