Provide Serializable Interface for Creating WorkerRuns #2232

cgardens · 2021-02-27T01:52:01Z

closes #2177

What

Adjust the interface for submitting jobs so that we have an interface that takes environment specific information for worker runs (in the future this information will be provided via temporal instead of the scheduler app) and a serializable configuration for setting up a WorkerRun of a specific type. In other words Temporal will run WorkerRuns (instead of just Workers) to use our existing naming.

How

Create a WorkerRunFactory for each job type that takes in the 2 types of configurations mentioned above. It looks silly right now, but should start to make more sense once the creation of the worker run happens in temporal.
What used to be called WorkerRunFactory is now called SchedulerWorkerRunAssembly. It has the environment specific information required for the Scheduler implementation of the executor to work. It also takes in the job type specific WorkerRunFactories. WorkerRunFactory is an interface for how to create WorkerRuns.

Pre-merge Checklist

lots of tests to fix

Recommended reading order

SchedulerWorkerRunAssembly
WorkerRunFactory (and classes that implement it)
the rest

michel-tricot

There are a lot of class indirections and classes with similar names or that are using Assembly because Factory was taken already.

I think the PR in its current state is going to make the code harder to follow.

airbyte-scheduler/src/main/java/io/airbyte/scheduler/SchedulerWorkerRunAssembly.java

michel-tricot · 2021-03-02T05:51:26Z

airbyte-scheduler/src/main/java/io/airbyte/scheduler/SchedulerWorkerRunAssembly.java

+ * This class is a runnable that give a job id and db connection figures out how to run the
+ * appropriate worker for a given job.
+ */
+public class SchedulerWorkerRunAssembly {


What do you call it an assembly? It seems to be a factory.

...src/main/java/io/airbyte/scheduler/worker_run_factories/CheckConnectionWorkerRunFactory.java

...eduler/src/main/java/io/airbyte/scheduler/worker_run_factories/DiscoverWorkerRunFactory.java

...heduler/src/main/java/io/airbyte/scheduler/worker_run_factories/GetSpecWorkerRunFactory.java

michel-tricot · 2021-03-02T06:03:30Z

...cheduler/src/main/java/io/airbyte/scheduler/worker_run_factories/SyncWorkerRunFactories.java

+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SyncWorkerRunFactories {


Why did you decide to group them together instead of splitting them up in separate classes?

because they are essentially just the same logic with very slight changes in configuration. i wanted the code close together.

...yte-scheduler/src/main/java/io/airbyte/scheduler/worker_run_factories/WorkerRunAssembly.java

jrhizor · 2021-03-02T22:34:44Z

...integrations/connectors/source-facebook-marketing/source_facebook_marketing/client/client.py

@@ -26,6 +26,7 @@

 import pendulum as pendulum
 from base_python import BaseClient
+


jrhizor · 2021-03-02T22:46:35Z

airbyte-scheduler/src/main/java/io/airbyte/scheduler/worker_run/WorkerRunFactoryUtils.java

+
+public class WorkerRunFactoryUtils {
+
+  public static IntegrationLauncher createLauncher(long jobId, int attempt, final String image, ProcessBuilderFactory pbf) {


Could put this as a static method AirbyteIntegrationLauncher.create(...) instead of having a new utils file.

Good point!

airbyte-scheduler/src/main/java/io/airbyte/scheduler/worker_run/WorkerRunFactory.java

jrhizor · 2021-03-02T22:55:03Z

.../src/main/java/io/airbyte/scheduler/worker_run/SchedulerWorkerRunWithEnvironmentFactory.java

+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SchedulerWorkerRunWithEnvironmentFactory {


Is the only value of this class vs just including it inside of JobSubmitter for testing?

I wish the name just didn't resemble WorkerRunWithEnvironmentFactory. JobToWorkerRunConverter or anything really feels better so they have clearer roles.

Agree, the naming really has a bad code smell right, but if I understand from your PR desc, only one will remain right?

exactly. this is to get us over the hump from scheduler doing stuff to temporal doing it. will go away.

michel-tricot · 2021-03-03T00:19:27Z

airbyte-scheduler/src/main/java/io/airbyte/scheduler/worker_run/SyncWorkerRunFactories.java

+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SyncWorkerRunFactories {


We don't have that pattern anywhere in the code and it doesn't match the pattern of the other factories. you're basically doing "modules" with static classes. If the logic is the same, you can use inheritance or your can use helpers.

michel-tricot · 2021-03-03T00:24:39Z

airbyte-scheduler/src/main/java/io/airbyte/scheduler/worker_run/WorkerRunFactory.java

+// T must be serializable. The generated json pojos do not implement serializable (but they are
+// serializable). It means we can't force Serializable as a constraint in this interface
+// unfortunately.
+public interface WorkerRunFactory<T> {


you can enforce serializable on T and you can make sure the generated classes implement Serializable. jsonschema2pojo supports it.

https://github.com/joelittlejohn/jsonschema2pojo/tree/master/jsonschema2pojo-gradle-plugin
serializable = true

🙌 thanks!

michel-tricot · 2021-03-03T00:25:24Z

airbyte-scheduler/src/main/java/io/airbyte/scheduler/worker_run/WorkerRunFactoryUtils.java

+
+public class WorkerRunFactoryUtils {
+
+  public static IntegrationLauncher createLauncher(long jobId, int attempt, final String image, ProcessBuilderFactory pbf) {


Good point!

michel-tricot · 2021-03-03T00:28:47Z

airbyte-scheduler/src/main/java/io/airbyte/scheduler/SchedulerApp.java

@@ -93,11 +94,12 @@ public SchedulerApp(Path workspaceRoot,
  public void start() throws IOException {
    final ExecutorService workerThreadPool = Executors.newFixedThreadPool(MAX_WORKERS, THREAD_FACTORY);
    final ScheduledExecutorService scheduledPool = Executors.newSingleThreadScheduledExecutor();
-    final WorkerRunFactory workerRunFactory = new WorkerRunFactory(workspaceRoot, pbf);
+    final SchedulerWorkerRunWithEnvironmentFactory schedulerWorkerRunWithEnvironmentFactory =


NIT: You can probably shorten the var name to avoid the multiline statement. I think it is still ok to call it workerRunFactory

michel-tricot · 2021-03-03T00:31:02Z

.../src/main/java/io/airbyte/scheduler/worker_run/SchedulerWorkerRunWithEnvironmentFactory.java

+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SchedulerWorkerRunWithEnvironmentFactory {


Agree, the naming really has a bad code smell right, but if I understand from your PR desc, only one will remain right?

michel-tricot · 2021-03-03T00:33:57Z

...scheduler/src/main/java/io/airbyte/scheduler/worker_run/WorkerRunWithEnvironmentFactory.java

+    this.workerRunFactory = workerRunFactory;
+  }
+
+  public WorkerRun create(final long jobId, final int attempt, final T config) {


I assume this logic of creating path can at some point belong to the workerrun right (once we are 100% on temporal)? it is a bit overkill at the moment to have it.

definitely.

i think we can redefine the worker run concept a bit going forward to simplify this, but i think figuring that out will be easier once we're on temporal because we'll know exactly what the constrains are.

…n/WorkerRunFactory.java Co-authored-by: Jared Rhizor <[email protected]>

cgardens requested a review from jrhizor February 27, 2021 01:52

jrhizor approved these changes Mar 1, 2021

View reviewed changes

michel-tricot requested changes Mar 2, 2021

View reviewed changes

cgardens requested a review from jrhizor March 2, 2021 22:18

jrhizor approved these changes Mar 2, 2021

View reviewed changes

michel-tricot approved these changes Mar 3, 2021

View reviewed changes

michel-tricot reviewed Mar 3, 2021

View reviewed changes

cgardens force-pushed the cgardens/worker_run_assembly branch 2 times, most recently from 4115f46 to 7749367 Compare March 3, 2021 01:58

cgardens and others added 11 commits March 3, 2021 10:13

start somewhere

08d33ab

rename

23d115d

test 1 of 80000000 passing

9d0e896

worker_run

9ac112d

clean

1d0380d

add tests for each worker run factory

4c1b6f7

Update airbyte-scheduler/src/main/java/io/airbyte/scheduler/worker_ru…

d133eb6

…n/WorkerRunFactory.java Co-authored-by: Jared Rhizor <[email protected]>

Update airbyte-scheduler/src/main/java/io/airbyte/scheduler/worker_ru…

c053974

…n/WorkerRunFactory.java Co-authored-by: Jared Rhizor <[email protected]>

typo

bf84301

code review feedback

bb15ffb

clean

d83b6e0

cgardens force-pushed the cgardens/worker_run_assembly branch from 418e9ce to d83b6e0 Compare March 3, 2021 18:13

cgardens merged commit 8503b7e into master Mar 3, 2021

cgardens deleted the cgardens/worker_run_assembly branch March 3, 2021 18:37

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Provide Serializable Interface for Creating WorkerRuns #2232

Provide Serializable Interface for Creating WorkerRuns #2232

cgardens commented Feb 27, 2021 •

edited

Loading

michel-tricot left a comment

michel-tricot Mar 2, 2021

michel-tricot Mar 2, 2021

cgardens Mar 2, 2021 •

edited

Loading

jrhizor Mar 2, 2021

jrhizor Mar 2, 2021

michel-tricot Mar 3, 2021

jrhizor Mar 2, 2021

jrhizor Mar 2, 2021

michel-tricot Mar 3, 2021

cgardens Mar 3, 2021

michel-tricot Mar 3, 2021

michel-tricot Mar 3, 2021

cgardens Mar 3, 2021

michel-tricot Mar 3, 2021

michel-tricot Mar 3, 2021

michel-tricot Mar 3, 2021

michel-tricot Mar 3, 2021

cgardens Mar 3, 2021

cgardens Mar 3, 2021

		@@ -26,6 +26,7 @@

		import pendulum as pendulum
		from base_python import BaseClient


		public class WorkerRunFactoryUtils {

		public static IntegrationLauncher createLauncher(long jobId, int attempt, final String image, ProcessBuilderFactory pbf) {

Provide Serializable Interface for Creating WorkerRuns #2232

Provide Serializable Interface for Creating WorkerRuns #2232

Conversation

cgardens commented Feb 27, 2021 • edited Loading

What

How

Pre-merge Checklist

Recommended reading order

michel-tricot left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cgardens Mar 2, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cgardens commented Feb 27, 2021 •

edited

Loading

cgardens Mar 2, 2021 •

edited

Loading