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 testCreateWorkflowInstance() {
    RunRequest request =
        RunRequest.builder()
            .initiator(new ManualInitiator())
            .currentPolicy(RunPolicy.START_FRESH_NEW_RUN)
            .build();
    Workflow workflow = definition.getWorkflow();
    WorkflowInstance instance =
        workflowHelper.createWorkflowInstance(workflow, 12345L, 1, new RunProperties(), request);
    assertEquals(workflow.getId(), instance.getWorkflowId());
    assertEquals(WorkflowInstance.Status.CREATED, instance.getStatus());
    assertEquals(12345L, instance.getInternalId().longValue());
    assertNotNull(instance.getParams());
    assertNotNull(instance.getWorkflowUuid());
    Mockito.verify(paramsManager, Mockito.times(1)).generateMergedWorkflowParams(any(), any());
  }