public WorkflowInstance createWorkflowInstance(
      Workflow workflowDef,
      Long internalId,
      long workflowVersionId,
      RunProperties runProperties,
      RunRequest runRequest) {
    WorkflowInstance instance = new WorkflowInstance();
    instance.setWorkflowId(workflowDef.getId());
    instance.setInternalId(internalId);
    instance.setWorkflowVersionId(workflowVersionId);
    // latest workflow instance id is unknown, update it later.
    instance.setWorkflowInstanceId(Constants.LATEST_ONE);
    // set correlation id if request contains it, otherwise, update it later inside DAO
    instance.setCorrelationId(runRequest.getCorrelationId());
    instance.setRunProperties(runProperties);
    // it includes runtime params and tags. Its dag is versioned dag.
    Workflow workflow = overrideWorkflowConfig(workflowDef, runRequest);
    instance.setRuntimeWorkflow(workflow);

    // update newly created workflow instance
    updateWorkflowInstance(instance, runRequest);
    return instance;
  }
@Test
  public void testCreateWorkflowInstanceFailedInitialization() {
    RunRequest badStartRequest =
        RunRequest.builder()
            .initiator(new ManualInitiator())
            .currentPolicy(RunPolicy.START_FRESH_NEW_RUN)
            .runParams(
                Collections.singletonMap(
                    "owner", ParamDefinition.buildParamDefinition("owner", "tester")))
            .build();
    Workflow workflow = definition.getWorkflow();
    Mockito.when(paramsManager.generateMergedWorkflowParams(any(), any()))
        .thenThrow(new MaestroInternalError("test-error"));

    AssertHelper.assertThrows(
        "should throw MaestroInternalError for initialization failure",
        MaestroInternalError.class,
        "test-error",
        () ->
            workflowHelper.createWorkflowInstance(
                workflow, 12345L, 1, new RunProperties(), badStartRequest));

    RunRequest startRequest = badStartRequest.toBuilder().persistFailedRun(true).build();
    WorkflowInstance instance =
        workflowHelper.createWorkflowInstance(
            workflow, 12345L, 1, new RunProperties(), startRequest);
    assertEquals(workflow.getId(), instance.getWorkflowId());
    assertEquals(WorkflowInstance.Status.FAILED, instance.getStatus());
    assertEquals(12345L, instance.getInternalId().longValue());
    Mockito.verify(paramsManager, Mockito.times(2)).generateMergedWorkflowParams(any(), any());
  }