Split the check connection workflow into multiple files.

airbyte-workers/src/main/java/io/airbyte/workers/WorkerApp.java

@@ -15,12 +15,13 @@
 import io.airbyte.workers.process.KubeProcessFactory;
 import io.airbyte.workers.process.ProcessFactory;
 import io.airbyte.workers.process.WorkerHeartbeatServer;
-import io.airbyte.workers.temporal.CheckConnectionWorkflow;
 import io.airbyte.workers.temporal.DiscoverCatalogWorkflow;
 import io.airbyte.workers.temporal.SpecWorkflow;
 import io.airbyte.workers.temporal.SyncWorkflow;
 import io.airbyte.workers.temporal.TemporalJobType;
 import io.airbyte.workers.temporal.TemporalUtils;
+import io.airbyte.workers.temporal.check.connection.CheckConnectionActivityImpl;
+import io.airbyte.workers.temporal.check.connection.CheckConnectionWorkflowImpl;
 import io.fabric8.kubernetes.client.DefaultKubernetesClient;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.kubernetes.client.openapi.ApiClient;
@@ -88,9 +89,9 @@ public void start() {
 
     final Worker checkConnectionWorker =
         factory.newWorker(TemporalJobType.CHECK_CONNECTION.name(), getWorkerOptions(maxWorkers.getMaxCheckWorkers()));
-    checkConnectionWorker.registerWorkflowImplementationTypes(CheckConnectionWorkflow.WorkflowImpl.class);
+    checkConnectionWorker.registerWorkflowImplementationTypes(CheckConnectionWorkflowImpl.class);
     checkConnectionWorker
-        .registerActivitiesImplementations(new CheckConnectionWorkflow.CheckConnectionActivityImpl(processFactory, secretsHydrator, workspaceRoot));
+        .registerActivitiesImplementations(new CheckConnectionActivityImpl(processFactory, secretsHydrator, workspaceRoot));
 
     final Worker discoverWorker = factory.newWorker(TemporalJobType.DISCOVER_SCHEMA.name(), getWorkerOptions(maxWorkers.getMaxDiscoverWorkers()));
     discoverWorker.registerWorkflowImplementationTypes(DiscoverCatalogWorkflow.WorkflowImpl.class);

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

@@ -19,6 +19,7 @@
 import io.airbyte.scheduler.models.IntegrationLauncherConfig;
 import io.airbyte.scheduler.models.JobRunConfig;
 import io.airbyte.workers.WorkerUtils;
+import io.airbyte.workers.temporal.check.connection.CheckConnectionWorkflow;
 import io.temporal.client.WorkflowClient;
 import java.nio.file.Path;
 import java.util.UUID;

airbyte-workers/src/main/java/io/airbyte/workers/temporal/check/connection/CheckConnectionActivity.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.workers.temporal.check.connection;
+
+import io.airbyte.config.StandardCheckConnectionInput;
+import io.airbyte.config.StandardCheckConnectionOutput;
+import io.airbyte.scheduler.models.IntegrationLauncherConfig;
+import io.airbyte.scheduler.models.JobRunConfig;
+import io.temporal.activity.ActivityInterface;
+import io.temporal.activity.ActivityMethod;
+
+@ActivityInterface
+interface CheckConnectionActivity {
+
+  @ActivityMethod
+  StandardCheckConnectionOutput run(JobRunConfig jobRunConfig,
+                                    IntegrationLauncherConfig launcherConfig,
+                                    StandardCheckConnectionInput connectionConfiguration);
+
+}

airbyte-workers/src/main/java/io/airbyte/workers/temporal/check/connection/CheckConnectionActivityImpl.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.workers.temporal.check.connection;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.airbyte.commons.functional.CheckedSupplier;
+import io.airbyte.config.StandardCheckConnectionInput;
+import io.airbyte.config.StandardCheckConnectionOutput;
+import io.airbyte.config.persistence.split_secrets.SecretsHydrator;
+import io.airbyte.scheduler.models.IntegrationLauncherConfig;
+import io.airbyte.scheduler.models.JobRunConfig;
+import io.airbyte.workers.DefaultCheckConnectionWorker;
+import io.airbyte.workers.Worker;
+import io.airbyte.workers.WorkerUtils;
+import io.airbyte.workers.process.AirbyteIntegrationLauncher;
+import io.airbyte.workers.process.IntegrationLauncher;
+import io.airbyte.workers.process.ProcessFactory;
+import io.airbyte.workers.temporal.CancellationHandler;
+import io.airbyte.workers.temporal.TemporalAttemptExecution;
+import java.nio.file.Path;
+import java.util.function.Supplier;
+
+public class CheckConnectionActivityImpl implements CheckConnectionActivity {
+
+  private final ProcessFactory processFactory;
+  private final SecretsHydrator secretsHydrator;
+  private final Path workspaceRoot;
+
+  public CheckConnectionActivityImpl(final ProcessFactory processFactory, final SecretsHydrator secretsHydrator, final Path workspaceRoot) {
+    this.processFactory = processFactory;
+    this.secretsHydrator = secretsHydrator;
+    this.workspaceRoot = workspaceRoot;
+  }
+
+  public StandardCheckConnectionOutput run(final JobRunConfig jobRunConfig,
+                                           final IntegrationLauncherConfig launcherConfig,
+                                           final StandardCheckConnectionInput connectionConfiguration) {
+
+    final JsonNode fullConfig = secretsHydrator.hydrate(connectionConfiguration.getConnectionConfiguration());
+
+    final StandardCheckConnectionInput input = new StandardCheckConnectionInput()
+        .withConnectionConfiguration(fullConfig);
+
+    final Supplier<StandardCheckConnectionInput> inputSupplier = () -> input;
+
+    final TemporalAttemptExecution<StandardCheckConnectionInput, StandardCheckConnectionOutput> temporalAttemptExecution =
+        new TemporalAttemptExecution<>(
+            workspaceRoot,
+            jobRunConfig,
+            getWorkerFactory(launcherConfig),
+            inputSupplier,
+            new CancellationHandler.TemporalCancellationHandler());
+
+    return temporalAttemptExecution.get();
+  }
+
+  private CheckedSupplier<Worker<StandardCheckConnectionInput, StandardCheckConnectionOutput>, Exception> getWorkerFactory(
+                                                                                                                           final IntegrationLauncherConfig launcherConfig) {
+    return () -> {
+      final IntegrationLauncher integrationLauncher = new AirbyteIntegrationLauncher(
+          launcherConfig.getJobId(),
+          Math.toIntExact(launcherConfig.getAttemptId()),
+          launcherConfig.getDockerImage(),
+          processFactory,
+          WorkerUtils.DEFAULT_RESOURCE_REQUIREMENTS);
+
+      return new DefaultCheckConnectionWorker(integrationLauncher);
+    };
+  }
+
+}

airbyte-workers/src/main/java/io/airbyte/workers/temporal/check/connection/CheckConnectionWorkflow.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.workers.temporal.check.connection;
+
+import io.airbyte.config.StandardCheckConnectionInput;
+import io.airbyte.config.StandardCheckConnectionOutput;
+import io.airbyte.scheduler.models.IntegrationLauncherConfig;
+import io.airbyte.scheduler.models.JobRunConfig;
+import io.temporal.workflow.WorkflowInterface;
+import io.temporal.workflow.WorkflowMethod;
+
+@WorkflowInterface
+public interface CheckConnectionWorkflow {
+
+  @WorkflowMethod
+  StandardCheckConnectionOutput run(JobRunConfig jobRunConfig,
+                                    IntegrationLauncherConfig launcherConfig,
+                                    StandardCheckConnectionInput connectionConfiguration);
+
+}

airbyte-workers/src/main/java/io/airbyte/workers/temporal/check/connection/CheckConnectionWorkflowImpl.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.workers.temporal.check.connection;
+
+import io.airbyte.config.StandardCheckConnectionInput;
+import io.airbyte.config.StandardCheckConnectionOutput;
+import io.airbyte.scheduler.models.IntegrationLauncherConfig;
+import io.airbyte.scheduler.models.JobRunConfig;
+import io.airbyte.workers.temporal.TemporalUtils;
+import io.temporal.activity.ActivityOptions;
+import io.temporal.workflow.Workflow;
+import java.time.Duration;
+
+public class CheckConnectionWorkflowImpl implements CheckConnectionWorkflow {
+
+  final ActivityOptions options = ActivityOptions.newBuilder()
+      .setScheduleToCloseTimeout(Duration.ofHours(1))
+      .setRetryOptions(TemporalUtils.NO_RETRY)
+      .build();
+  private final CheckConnectionActivity activity = Workflow.newActivityStub(CheckConnectionActivity.class, options);
+
+  @Override
+  public StandardCheckConnectionOutput run(final JobRunConfig jobRunConfig,
+                                           final IntegrationLauncherConfig launcherConfig,
+                                           final StandardCheckConnectionInput connectionConfiguration) {
+    return activity.run(jobRunConfig, launcherConfig, connectionConfiguration);
+  }
+
+}

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

@@ -24,6 +24,7 @@
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
 import io.airbyte.scheduler.models.IntegrationLauncherConfig;
 import io.airbyte.scheduler.models.JobRunConfig;
+import io.airbyte.workers.temporal.check.connection.CheckConnectionWorkflow;
 import io.temporal.client.WorkflowClient;
 import java.io.IOException;
 import java.nio.file.Files;