feat: deterministic workflow Id (#13948)

Author: benmoriceau

airbyte-commons-temporal/src/main/java/io/airbyte/commons/temporal/TemporalClient.java

@@ -413,7 +413,7 @@ public TemporalResponse<ConnectorJobOutput> submitGetSpec(final UUID jobId,
         .withDockerImage(config.getDockerImage())
         .withIsCustomConnector(config.getIsCustomConnector());
     return execute(jobRunConfig,
-        () -> getWorkflowStub(SpecWorkflow.class, TemporalJobType.GET_SPEC).run(jobRunConfig, launcherConfig));
+        () -> getWorkflowStub(SpecWorkflow.class, TemporalJobType.GET_SPEC, jobId).run(jobRunConfig, launcherConfig));
 
   }
 
@@ -450,7 +450,7 @@ public TemporalResponse<ConnectorJobOutput> submitCheckConnection(final UUID job
         .withActorContext(context);
 
     return execute(jobRunConfig,
-        () -> getWorkflowStubWithTaskQueue(CheckConnectionWorkflow.class, taskQueue).run(jobRunConfig, launcherConfig, input));
+        () -> getWorkflowStubWithTaskQueue(CheckConnectionWorkflow.class, taskQueue, jobId).run(jobRunConfig, launcherConfig, input));
   }
 
   /**
@@ -484,7 +484,7 @@ public TemporalResponse<ConnectorJobOutput> submitDiscoverSchema(final UUID jobI
         .withResourceRequirements(config.getResourceRequirements()).withActorContext(context).withManual(true);
 
     return execute(jobRunConfig,
-        () -> getWorkflowStubWithTaskQueue(DiscoverCatalogWorkflow.class, taskQueue).run(jobRunConfig, launcherConfig, input));
+        () -> getWorkflowStubWithTaskQueue(DiscoverCatalogWorkflow.class, taskQueue, jobId).run(jobRunConfig, launcherConfig, input));
   }
 
   /**
@@ -542,12 +542,12 @@ <T> TemporalResponse<T> execute(final JobRunConfig jobRunConfig, final Supplier<
     return new TemporalResponse<>(operationOutput, metadata);
   }
 
-  private <T> T getWorkflowStub(final Class<T> workflowClass, final TemporalJobType jobType) {
-    return workflowClientWrapped.newWorkflowStub(workflowClass, TemporalWorkflowUtils.buildWorkflowOptions(jobType));
+  private <T> T getWorkflowStub(final Class<T> workflowClass, final TemporalJobType jobType, final UUID jobId) {
+    return workflowClientWrapped.newWorkflowStub(workflowClass, TemporalWorkflowUtils.buildWorkflowOptions(jobType, jobId));
   }
 
-  private <T> T getWorkflowStubWithTaskQueue(final Class<T> workflowClass, final String taskQueue) {
-    return workflowClientWrapped.newWorkflowStub(workflowClass, TemporalWorkflowUtils.buildWorkflowOptionsWithTaskQueue(taskQueue));
+  private <T> T getWorkflowStubWithTaskQueue(final Class<T> workflowClass, final String taskQueue, final UUID jobId) {
+    return workflowClientWrapped.newWorkflowStub(workflowClass, TemporalWorkflowUtils.buildWorkflowOptionsWithTaskQueue(taskQueue, jobId));
   }
 
   /**

airbyte-commons-temporal/src/main/java/io/airbyte/commons/temporal/TemporalWorkflowUtils.java

@@ -66,8 +66,8 @@ public static WorkflowOptions buildWorkflowOptions(final TemporalJobType jobType
    * @param jobType job type
    * @return workflow options
    */
-  public static WorkflowOptions buildWorkflowOptions(final TemporalJobType jobType) {
-    return buildWorkflowOptionsWithTaskQueue(jobType.name());
+  public static WorkflowOptions buildWorkflowOptions(final TemporalJobType jobType, final UUID jobd) {
+    return buildWorkflowOptionsWithTaskQueue(jobType.name(), jobd);
   }
 
   /**
@@ -76,13 +76,14 @@ public static WorkflowOptions buildWorkflowOptions(final TemporalJobType jobType
    * @param taskQueue task queue
    * @return workflow options
    */
-  public static WorkflowOptions buildWorkflowOptionsWithTaskQueue(final String taskQueue) {
+  public static WorkflowOptions buildWorkflowOptionsWithTaskQueue(final String taskQueue, final UUID jobID) {
     return WorkflowOptions.newBuilder()
         .setTaskQueue(taskQueue)
         .setWorkflowTaskTimeout(Duration.ofSeconds(27)) // TODO parker - temporarily increasing this to a recognizable number to see if it changes
         // error I'm seeing
         // todo (cgardens) we do not leverage Temporal retries.
         .setRetryOptions(RetryOptions.newBuilder().setMaximumAttempts(1).build())
+        .setWorkflowId(jobID.toString())
         .build();
   }
 

airbyte-commons-temporal/src/test/java/io/airbyte/commons/temporal/TemporalClientTest.java

@@ -230,21 +230,22 @@ class TestJobSubmission {
     @Test
     void testSubmitGetSpec() {
       final SpecWorkflow specWorkflow = mock(SpecWorkflow.class);
-      when(workflowClient.newWorkflowStub(SpecWorkflow.class, TemporalWorkflowUtils.buildWorkflowOptions(TemporalJobType.GET_SPEC)))
+      when(workflowClient.newWorkflowStub(SpecWorkflow.class, TemporalWorkflowUtils.buildWorkflowOptions(TemporalJobType.GET_SPEC, JOB_UUID)))
           .thenReturn(specWorkflow);
       final JobGetSpecConfig getSpecConfig = new JobGetSpecConfig().withDockerImage(IMAGE_NAME1);
 
       temporalClient.submitGetSpec(JOB_UUID, ATTEMPT_ID, WORKSPACE_ID, getSpecConfig);
       specWorkflow.run(JOB_RUN_CONFIG, UUID_LAUNCHER_CONFIG);
-      verify(workflowClient).newWorkflowStub(SpecWorkflow.class, TemporalWorkflowUtils.buildWorkflowOptions(TemporalJobType.GET_SPEC));
+      verify(workflowClient).newWorkflowStub(SpecWorkflow.class, TemporalWorkflowUtils.buildWorkflowOptions(TemporalJobType.GET_SPEC, JOB_UUID));
     }
 
     @Test
     void testSubmitCheckConnection() {
       final CheckConnectionWorkflow checkConnectionWorkflow = mock(CheckConnectionWorkflow.class);
       when(
-          workflowClient.newWorkflowStub(CheckConnectionWorkflow.class, TemporalWorkflowUtils.buildWorkflowOptions(TemporalJobType.CHECK_CONNECTION)))
-              .thenReturn(checkConnectionWorkflow);
+          workflowClient.newWorkflowStub(CheckConnectionWorkflow.class,
+              TemporalWorkflowUtils.buildWorkflowOptions(TemporalJobType.CHECK_CONNECTION, JOB_UUID)))
+                  .thenReturn(checkConnectionWorkflow);
       final JobCheckConnectionConfig checkConnectionConfig = new JobCheckConnectionConfig()
           .withDockerImage(IMAGE_NAME1)
           .withConnectionConfiguration(Jsons.emptyObject());
@@ -254,14 +255,15 @@ void testSubmitCheckConnection() {
       temporalClient.submitCheckConnection(JOB_UUID, ATTEMPT_ID, WORKSPACE_ID, CHECK_TASK_QUEUE, checkConnectionConfig, new ActorContext());
       checkConnectionWorkflow.run(JOB_RUN_CONFIG, UUID_LAUNCHER_CONFIG, input);
       verify(workflowClient).newWorkflowStub(CheckConnectionWorkflow.class,
-          TemporalWorkflowUtils.buildWorkflowOptions(TemporalJobType.CHECK_CONNECTION));
+          TemporalWorkflowUtils.buildWorkflowOptions(TemporalJobType.CHECK_CONNECTION, JOB_UUID));
     }
 
     @Test
     void testSubmitDiscoverSchema() {
       final DiscoverCatalogWorkflow discoverCatalogWorkflow = mock(DiscoverCatalogWorkflow.class);
-      when(workflowClient.newWorkflowStub(DiscoverCatalogWorkflow.class, TemporalWorkflowUtils.buildWorkflowOptions(TemporalJobType.DISCOVER_SCHEMA)))
-          .thenReturn(discoverCatalogWorkflow);
+      when(workflowClient.newWorkflowStub(DiscoverCatalogWorkflow.class,
+          TemporalWorkflowUtils.buildWorkflowOptions(TemporalJobType.DISCOVER_SCHEMA, JOB_UUID)))
+              .thenReturn(discoverCatalogWorkflow);
       final JobDiscoverCatalogConfig checkConnectionConfig = new JobDiscoverCatalogConfig()
           .withDockerImage(IMAGE_NAME1)
           .withConnectionConfiguration(Jsons.emptyObject());
@@ -272,7 +274,7 @@ void testSubmitDiscoverSchema() {
           WorkloadPriority.DEFAULT);
       discoverCatalogWorkflow.run(JOB_RUN_CONFIG, UUID_LAUNCHER_CONFIG, input);
       verify(workflowClient).newWorkflowStub(DiscoverCatalogWorkflow.class,
-          TemporalWorkflowUtils.buildWorkflowOptions(TemporalJobType.DISCOVER_SCHEMA));
+          TemporalWorkflowUtils.buildWorkflowOptions(TemporalJobType.DISCOVER_SCHEMA, JOB_UUID));
     }
 
   }