Fail the connection state instead of throwing an exception (#13728)

* Fail the connection state instead of throwin exception

* Fix test

* rm import

* Update airbyte-workers/src/main/java/io/airbyte/workers/general/DefaultCheckConnectionWorker.java

Co-authored-by: Evan Tahler <evan@airbyte.io>

Co-authored-by: Evan Tahler <evan@airbyte.io>

airbyte-workers/src/main/java/io/airbyte/workers/general/DefaultCheckConnectionWorker.java

@@ -79,11 +79,19 @@ public StandardCheckConnectionOutput run(final StandardCheckConnectionInput inpu
         LOGGER.debug("Check connection job received output: {}", output);
         return output;
       } else {
-        throw new WorkerException(String.format("Error checking connection, status: %s, exit code: %d", status, exitCode));
+        String message = String.format("Error checking connection, status: %s, exit code: %d", status, exitCode);
+
+        LOGGER.error(message);
+        return new StandardCheckConnectionOutput()
+            .withStatus(Status.FAILED)
+            .withMessage(message);
       }
 
     } catch (final Exception e) {
-      throw new WorkerException("Error while getting checking connection.", e);
+      LOGGER.error("Error while checking connection: ", e);
+      return new StandardCheckConnectionOutput()
+          .withStatus(Status.FAILED)
+          .withMessage("Error while getting checking connection, because of: " + e.getMessage());
     }
   }
 

airbyte-workers/src/test/java/io/airbyte/workers/general/DefaultCheckConnectionWorkerTest.java

@@ -6,7 +6,6 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
@@ -101,19 +100,23 @@ public void testFailedConnection() throws WorkerException {
   }
 
   @Test
-  public void testProcessFail() {
+  public void testProcessFail() throws WorkerException {
     when(process.exitValue()).thenReturn(1);
 
     final DefaultCheckConnectionWorker worker = new DefaultCheckConnectionWorker(workerConfigs, integrationLauncher, failureStreamFactory);
-    assertThrows(WorkerException.class, () -> worker.run(input, jobRoot));
+    final StandardCheckConnectionOutput output = worker.run(input, jobRoot);
+
+    assertEquals(Status.FAILED, output.getStatus());
   }
 
   @Test
   public void testExceptionThrownInRun() throws WorkerException {
     doThrow(new RuntimeException()).when(integrationLauncher).check(jobRoot, WorkerConstants.SOURCE_CONFIG_JSON_FILENAME, Jsons.serialize(CREDS));
 
     final DefaultCheckConnectionWorker worker = new DefaultCheckConnectionWorker(workerConfigs, integrationLauncher, failureStreamFactory);
-    assertThrows(WorkerException.class, () -> worker.run(input, jobRoot));
+    final StandardCheckConnectionOutput output = worker.run(input, jobRoot);
+
+    assertEquals(Status.FAILED, output.getStatus());
   }
 
   @Test