Pass job id from resetConnection to synchronousResetConnection to avoid race condition (#14434)

* pass job id from resetConnection to synchronousResetConnection to avoid race condition * return reset job ID in synchronous method

Author: lmossman

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

@@ -386,6 +386,8 @@ public ManualOperationResult resetConnection(final UUID connectionId, final List
  Optional.empty());
  }
 
+ Optional<Long> newJobId;
+
  do {
  try {
  Thread.sleep(DELAY_BETWEEN_QUERY_MS);
@@ -394,23 +396,22 @@ public ManualOperationResult resetConnection(final UUID connectionId, final List
  Optional.of("Didn't manage to reset a sync for: " + connectionId),
  Optional.empty());
  }
- } while (!newJobStarted(connectionId, oldJobId));
+ newJobId = getNewJobId(connectionId, oldJobId);
+ } while (newJobId.isEmpty());
 
  log.info("end of reset submission");
 
- final long jobId = ConnectionManagerUtils.getCurrentJobId(client, connectionId);
-
  return new ManualOperationResult(
  Optional.empty(),
- Optional.of(jobId));
+ newJobId);
  }
 
- private boolean newJobStarted(final UUID connectionId, final long oldJobId) {
+ private Optional<Long> getNewJobId(final UUID connectionId, final long oldJobId) {
  final long currentJobId = ConnectionManagerUtils.getCurrentJobId(client, connectionId);
  if (currentJobId == NON_RUNNING_JOB_ID || currentJobId == oldJobId) {
- return false;
+ return Optional.empty();
  } else {
- return true;
+ return Optional.of(currentJobId);
  }
  }
 
@@ -426,16 +427,7 @@ public ManualOperationResult synchronousResetConnection(final UUID connectionId,
  return resetResult;
  }
 
- try {
- ConnectionManagerUtils.getConnectionManagerWorkflow(client, connectionId);
- } catch (final Exception e) {
- log.error("Encountered exception retrieving workflow after reset.", e);
- return new ManualOperationResult(
- Optional.of(e.getMessage()),
- Optional.empty());
- }
-
- final long oldJobId = ConnectionManagerUtils.getCurrentJobId(client, connectionId);
+ final long resetJobId = resetResult.getJobId().get();
 
  do {
  try {
@@ -445,15 +437,13 @@ public ManualOperationResult synchronousResetConnection(final UUID connectionId,
  Optional.of("Didn't manage to reset a sync for: " + connectionId),
  Optional.empty());
  }
- } while (ConnectionManagerUtils.getCurrentJobId(client, connectionId) == oldJobId);
+ } while (ConnectionManagerUtils.getCurrentJobId(client, connectionId) == resetJobId);
 
  log.info("End of reset");
 
- final long jobId = ConnectionManagerUtils.getCurrentJobId(client, connectionId);
-
  return new ManualOperationResult(
  Optional.empty(),
- Optional.of(jobId));
+ Optional.of(resetJobId));
  }
 
  private <T> T getWorkflowStub(final Class<T> workflowClass, final TemporalJobType jobType) {

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

@@ -231,17 +231,15 @@ public void testSynchronousResetConnection() throws IOException {
  final WorkflowState mWorkflowState = mock(WorkflowState.class);
  when(mConnectionManagerWorkflow.getState()).thenReturn(mWorkflowState);
  when(mWorkflowState.isDeleted()).thenReturn(false);
- final long jobId1 = 1L;
- final long jobId2 = 2L;
- final long jobId3 = 3L;
+ final long resetJobId = 1L;
 
  when(mConnectionManagerWorkflow.getJobInformation()).thenReturn(
- new JobInformation(jobId1, 0),
- new JobInformation(jobId2, 0),
- new JobInformation(jobId2, 0),
- new JobInformation(jobId2, 0),
- new JobInformation(jobId3, 0),
- new JobInformation(jobId3, 0));
+ new JobInformation(NON_RUNNING_JOB_ID, 0),
+ new JobInformation(NON_RUNNING_JOB_ID, 0),
+ new JobInformation(resetJobId, 0),
+ new JobInformation(resetJobId, 0),
+ new JobInformation(NON_RUNNING_JOB_ID, 0),
+ new JobInformation(NON_RUNNING_JOB_ID, 0));
 
  doReturn(true).when(temporalClient).isWorkflowReachable(any(UUID.class));
 
@@ -253,7 +251,7 @@ public void testSynchronousResetConnection() throws IOException {
  verify(streamResetPersistence).createStreamResets(CONNECTION_ID, streamsToReset);
  verify(mConnectionManagerWorkflow).resetConnection();
 
- assertEquals(manualOperationResult.getJobId().get(), jobId3);
+ assertEquals(manualOperationResult.getJobId().get(), resetJobId);
  }
 
  }