Fix workflow retrieval (#9096)

use the shared method to get workflow names
airbytehq · Dec 24, 2021 · 0bce8c8 · 0bce8c8
1 parent a7b906b
commit 0bce8c8
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 9 deletions.
diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/temporal/TemporalClient.java b/airbyte-workers/src/main/java/io/airbyte/workers/temporal/TemporalClient.java
@@ -185,7 +185,7 @@ public class ManualSyncSubmissionResult {
 
   public ManualSyncSubmissionResult startNewManualSync(final UUID connectionId) {
     log.info("Manual sync request");
-    final boolean isWorflowRunning = isWorkflowRunning("connection_updater_" + connectionId);
+    final boolean isWorflowRunning = isWorkflowRunning(getConnectionManagerName(connectionId));
 
     if (!isWorflowRunning) {
       return new ManualSyncSubmissionResult(
@@ -194,7 +194,7 @@ public ManualSyncSubmissionResult startNewManualSync(final UUID connectionId) {
     }
 
     final ConnectionManagerWorkflow connectionManagerWorkflow =
-        getExistingWorkflow(ConnectionManagerWorkflow.class, "connection_updater_" + connectionId);
+        getExistingWorkflow(ConnectionManagerWorkflow.class, getConnectionManagerName(connectionId));
     final WorkflowState workflowState = connectionManagerWorkflow.getState();
 
     if (workflowState.isRunning()) {
@@ -236,7 +236,7 @@ public class NewCancellationSubmissionResult {
   public ManualSyncSubmissionResult startNewCancelation(final UUID connectionId) {
     log.info("Manual sync request");
 
-    final boolean isWorflowRunning = isWorkflowRunning("connection_updater_" + connectionId);
+    final boolean isWorflowRunning = isWorkflowRunning(getConnectionManagerName(connectionId));
 
     if (!isWorflowRunning) {
       return new ManualSyncSubmissionResult(
@@ -245,7 +245,7 @@ public ManualSyncSubmissionResult startNewCancelation(final UUID connectionId) {
     }
 
     final ConnectionManagerWorkflow connectionManagerWorkflow =
-        getExistingWorkflow(ConnectionManagerWorkflow.class, "connection_updater_" + connectionId);
+        getExistingWorkflow(ConnectionManagerWorkflow.class, getConnectionManagerName(connectionId));
 
     connectionManagerWorkflow.cancelJob();
 

diff --git a/...s/src/test/java/io/airbyte/workers/temporal/scheduling/ConnectionManagerWorkflowTest.java b/...s/src/test/java/io/airbyte/workers/temporal/scheduling/ConnectionManagerWorkflowTest.java
@@ -159,7 +159,7 @@ public void retryAfterFail() {
           workflowState);
 
       WorkflowClient.start(workflow::run, input);
-      testEnv.sleep(Duration.ofSeconds(4L));
+      testEnv.sleep(Duration.ofSeconds(50L));
       final Queue<ChangedStateEvent> events = testStateListener.events(testId);
 
       Assertions.assertThat(events)
@@ -190,6 +190,7 @@ public void manualRun() {
           workflowState);
 
       WorkflowClient.start(workflow::run, input);
+      testEnv.sleep(Duration.ofSeconds(2L));
       workflow.submitManualSync();
       testEnv.sleep(Duration.ofSeconds(50L));
 
@@ -230,8 +231,9 @@ public void updatedSignalRecieved() {
           workflowState);
 
       WorkflowClient.start(workflow::run, input);
+      testEnv.sleep(Duration.ofSeconds(2L));
       workflow.connectionUpdated();
-      testEnv.sleep(Duration.ofSeconds(4L));
+      testEnv.sleep(Duration.ofSeconds(50L));
 
       final Queue<ChangedStateEvent> events = testStateListener.events(testId);
 
@@ -270,9 +272,9 @@ public void cancelNonRunning() {
           workflowState);
 
       WorkflowClient.start(workflow::run, input);
-      testEnv.sleep(Duration.ofSeconds(4L));
+      testEnv.sleep(Duration.ofSeconds(50L));
       workflow.cancelJob();
-      testEnv.sleep(Duration.ofSeconds(4L));
+      testEnv.sleep(Duration.ofSeconds(2L));
 
       final Queue<ChangedStateEvent> events = testStateListener.events(testId);
 
@@ -312,7 +314,7 @@ public void deleteSync() {
 
       WorkflowClient.start(workflow::run, input);
       workflow.deleteConnection();
-      testEnv.sleep(Duration.ofSeconds(4L));
+      testEnv.sleep(Duration.ofSeconds(50L));
 
       final Queue<ChangedStateEvent> events = testStateListener.events(testId);