separate tap and target in sync worker 2 #113

cgardens · 2020-08-26T07:24:59Z

Note: I will NOT merge this until the final two items in the checklist below are complete. I think this is in a good state, however, to get feedback now. Really looking for a yay or nay on whether we are happy with this approach and it's worth bringing across the finish line.

What

Per our discussion on RFC PR, this PR attempts to implement the sync worker in such a way that separate tap and target from each other. This opens up in the future the ability to 1. use non-singer taps / targets 2) introduce our own protocol 3) provide a well-defined interface to build new integrations against.

How

Separate tap and target behind two new interface. For now they are just implemented specifically for singer.

Checklist

try tap and target as try-as-resource
redo all exception handling within the workers (NOT GOING TO DO THIS IN THIS PR. Need to look at all workers holistically next to get this right.)
reimplement the test coverage of SingerSyncWorkerTest for DefaultSyncWorker, SingerTap, and SingerTarget.
write tests in general

Recommended reading order

WorkerRunner.java
DefaultSyncWorker
SyncTap
SyncTarget
SingerTap
SingerTarget
the rest

closes #90

cgardens · 2020-08-26T08:02:25Z

dataline-workers/src/main/java/io/dataline/workers/singer/SingerSyncWorker.java

@@ -1,239 +0,0 @@
-/*


replaced by DefaultSyncWorker.

cgardens · 2020-08-26T08:03:07Z

dataline-workers/src/test/java/io/dataline/workers/singer/SingerSyncWorkerTest.java

@@ -1,208 +0,0 @@
-/*


Still need to reimplement these tests for DefaultSyncWorker, SingerTap, and SingerTarget.

dataline-workers/src/main/java/io/dataline/workers/singer/SingerTap.java

cgardens · 2020-08-26T18:11:25Z

dataline-workers/src/main/java/io/dataline/workers/ConsumerIterator.java

+import java.util.Iterator;
+import java.util.function.Consumer;
+
+/**


i tried to find a guava iterator that did this, but could not find one. feels like something that should exist. i guess the counterargument here is we just used a stream we could use peek instead.

A stream with peek sounds better for this.

cgardens · 2020-08-26T18:15:55Z

dataline-workers/src/main/java/io/dataline/workers/SyncTap.java

+import java.util.Iterator;
+
+public interface SyncTap<T> extends AutoCloseable {
+  Iterator<T> run(StandardTapConfig tapConfig, Path workspacePath)


do we have an opinion here on Iterator versus Stream?

cgardens · 2020-08-26T19:08:45Z

dataline-workers/src/main/java/io/dataline/workers/DefaultSyncWorker.java

+    final State state;
+    try (SyncTap<SingerProtocol> tap = new SingerTap(tapDockerImage)) {
+      final Iterator<SingerProtocol> iterator = tap.run(tapConfig, workspacePath);
+      tapCloser = tap::close;


this is a little weird, but it was the only way i could figure out how to do this while still instantiating the tap in the try-with-resource block.

jrhizor

Generally think this is in the right direction, but the implementation DefaultSyncWorker and the other interfaces would be cleaner with Stream-ing.

dataline-workers/src/main/java/io/dataline/workers/DefaultSyncWorker.java

jrhizor · 2020-08-26T22:04:00Z

dataline-workers/src/main/java/io/dataline/workers/SyncTap.java

+import java.nio.file.Path;
+import java.util.Iterator;
+
+public interface SyncTap<T> extends AutoCloseable {


This doesn't actually seem tied to the Sync concept. Maybe TapStream? If it extended Stream, it would already be autocloseable.

It seems like you separately want a factory from (StandardTapConfig tapConfig, Path workspacePath) -> TapStream.

Same comment for SyncTarget. I'm also disliking how we're using Tap/Target in some places that isn't Singer-specific instead of Source/Destination.

Same comment for SyncTarget. I'm also disliking how we're using Tap/Target in some places that isn't Singer-specific instead of Source/Destination.

good call.

haven't done this name change yet. but will do it.

dataline-workers/src/main/java/io/dataline/workers/SyncTap.java

dataline-workers/src/main/java/io/dataline/workers/protocol/SingerJsonIterator.java

jrhizor · 2020-08-26T22:14:55Z

dataline-workers/src/main/java/io/dataline/workers/singer/SingerTap.java

+
+  @Override
+  public void cancel() {
+    WorkerUtils.cancelHelper(tapProcess);


How does this manifest in the Iterator? Does it throw an IOException or InterruptedException or something else?

jrhizor · 2020-08-26T22:48:46Z

dataline-workers/src/main/java/io/dataline/workers/ConsumerIterator.java

+import java.util.Iterator;
+import java.util.function.Consumer;
+
+/**


A stream with peek sounds better for this.

jrhizor · 2020-08-26T22:50:09Z

dataline-workers/src/main/java/io/dataline/workers/DefaultSyncWorker.java

+
+    StandardSyncSummary summary = new StandardSyncSummary();
+    summary.setRecordsSynced(recordCount.getValue());
+    summary.setStartTime(startTime);


Would we show this time, the Job timestamps, or both in the UI?

i guess this one? don't feel strongly. this timestamp feels boarding on redundant.

jrhizor · 2020-08-26T22:57:47Z

dataline-workers/src/main/java/io/dataline/workers/singer/SingerTarget.java

+  }
+
+  @Override
+  public State run(


So this would basically be a stateful consumer of the tap stream if this is reimplemented using streams?

SingerTarget st = new SingerTarget(...); TapFactory.create(...) .peek(statsPeeker) .forEach(st); State state = st.getState();

yeah. check it out.

jrhizor · 2020-08-26T22:58:16Z

dataline-workers/src/main/java/io/dataline/workers/singer/SingerTarget.java

+
+  @Override
+  public void cancel() {
+    WorkerUtils.cancelHelper(targetProcess);


Also a little confused about the cancellation lifecycle for targets.

i think i was confused too. lmk if new version clarifies it.

jrhizor · 2020-08-26T23:01:06Z

dataline-workers/src/main/java/io/dataline/workers/DefaultSyncWorker.java

+  @Override
+  public void cancel() {
+    try {
+      if (tapCloser != null) {


Maybe

Optional.ofNullable(tapCloser).ifPresent(closer -> closer.close()); Optional.ofNullable(targetCloser).ifPresent(closer -> closer.close());

this ends up uglier because you need to handle the Exception inside the lambda which explodes the lines of the code.

okay. ended up having to handle the exceptions in lambda for other reasons so went with this now :D

cgardens · 2020-08-26T23:26:32Z

I got some really useful feedback from Jared. Switching this back to draft while I incorporate it. Always happy to get more feedback if you read it anyway, but feel free to hold off until I un-draft it again.

cgardens · 2020-08-27T01:10:50Z

dataline-workers/src/main/java/io/dataline/workers/singer/SingerTapFactory.java

+    }
+
+    stdout = tapProcess.getInputStream();
+    return Streams.stream(new SingerJsonIterator(stdout)).onClose(getCloseFunction());


@jrhizor - would be curious to hear what you think about this. i'm a little wary because 1. people don't usually call close on streams in java 2. it doesn't codify close in the interface so it's a little harder to enforce the close step in the lifecycle. don't hate it. don't love it. what do you think?

Seems fine to me. The fact that the close can be called separately in the cancellation project is the part that feels clunky, so I'm not sure changing this makes the overall usage "nicer".

cgardens · 2020-08-27T01:11:27Z

Okay, I think I have now integrated @jrhizor 's feedback and we are now all set for another round of review.

jrhizor · 2020-08-27T16:39:13Z

dataline-workers/src/main/java/io/dataline/workers/DefaultSyncWorker.java

+      try (Target<SingerMessage> target = new SingerTarget(targetDockerImage)) {
+        targetCloser = getTargetCloser(target);
+
+        target.consume(peakedTapStream, targetConfig, workspacePath);


Feels a bit strange that the targetConfig and workspacePath are inputs to the consumer instead of the consumer actually being a Consumer and being called on the stream's forEach. They are always going to be the same for the target's lifespan.

jrhizor · 2020-08-27T16:40:49Z

dataline-workers/src/main/java/io/dataline/workers/Target.java

+import java.util.stream.Stream;
+
+public interface Target<T> extends AutoCloseable {
+  void consume(Stream<T> data, StandardTargetConfig targetConfig, Path workspacePath)


See above comment, feels like this should implement void accept(T t) and be an AutoCloseable Consumer, with a TargetFactory that takes in the config and path.

the interface is written from the point of view of what interface could we given an OSS developer to write an integration against. i think this interface does that part well, but i think internal to worker, your critique makes sense.

I think I have an idea on how to do do both the interface thing that i set out to do and fix the part that you think is weird too.

jrhizor · 2020-08-27T16:44:00Z

dataline-workers/src/main/java/io/dataline/workers/singer/SingerTapFactory.java

+    }
+
+    stdout = tapProcess.getInputStream();
+    return Streams.stream(new SingerJsonIterator(stdout)).onClose(getCloseFunction());


Seems fine to me. The fact that the close can be called separately in the cancellation project is the part that feels clunky, so I'm not sure changing this makes the overall usage "nicer".

cgardens · 2020-08-27T18:22:46Z

dataline-workers/src/main/java/io/dataline/workers/DefaultSyncWorker.java

+    final SingerTapFactory singerTapFactory = new SingerTapFactory(tapDockerImage);
+    final SingerTargetFactory singerTargetFactory = new SingerTargetFactory(targetDockerImage);
+
+    try (Stream<SingerMessage> tap = singerTapFactory.create(tapConfig, workspacePath)) {


going to dependency inject the tapfactory and targetfactory.

michel-tricot

Really cool PR

michel-tricot · 2020-08-27T18:06:53Z

dataline-workers/src/main/java/io/dataline/workers/DefaultSyncWorker.java


-        target.consume(peakedTapStream, targetConfig, workspacePath);
+      try (CloseableConsumer<SingerMessage> consumer =


you can create both resources in the same try:
try (tap = xxx; consumer = yyy)

michel-tricot · 2020-08-27T18:10:53Z

dataline-workers/src/main/java/io/dataline/workers/CloseableConsumer.java

+
+import java.util.function.Consumer;
+
+public interface CloseableConsumer<T> extends Consumer<T>, AutoCloseable {}


this belongs to commons maybe add a functional package.
add the @ FunctionalInterface annotation

will move. i think not technically functional interface since it has consume and close, right?

hum right... :)

michel-tricot · 2020-08-27T18:11:45Z

dataline-scheduler/src/main/java/io/dataline/scheduler/WorkerRunner.java

@@ -89,7 +89,11 @@ public void run() {
        new WorkerRun<>(
                jobId,
                syncInput,
-                new SingerSyncWorker(
+                // todo (cgardens) - still locked into only using SingerTaps and Targets. Next step


I see nothing related to singer in that code. I think this comment is mis-leading

i'll fix this. going to dependency inject the factories and then this comment will make sense again.

michel-tricot · 2020-08-27T18:12:28Z

dataline-workers/src/main/java/io/dataline/workers/DefaultSyncWorker.java

+  private final String tapDockerImage;
+  private final String targetDockerImage;
+
+  Runnable tapCloser = null;


why is this not private?

michel-tricot · 2020-08-27T18:14:59Z

dataline-workers/src/main/java/io/dataline/workers/DefaultSyncWorker.java

+    final SingerTargetFactory singerTargetFactory = new SingerTargetFactory(targetDockerImage);
+
+    try (Stream<SingerMessage> tap = singerTapFactory.create(tapConfig, workspacePath)) {
+      tapCloser = tap::close;


that's a nasty state. My suggestion: don't use forEach instead loop with an iterator and have a boolean that gets set when it is time to stop.

michel-tricot · 2020-08-27T18:17:32Z

dataline-workers/src/main/java/io/dataline/workers/DefaultSyncWorker.java

+
+  @Override
+  public void cancel() {
+    Optional.ofNullable(tapCloser).ifPresent(Runnable::run);


set a boolean instead. Make sure you set the proper JobStatus in the run.

michel-tricot · 2020-08-27T18:19:45Z

dataline-workers/src/main/java/io/dataline/workers/WorkerUtils.java

@@ -56,6 +81,9 @@ public static String readFileFromWorkspace(Path workspaceRoot, String fileName)
  }

  public static void cancelHelper(Process workerProcess) {


That's not a great name for a function what about cancelProcess?

michel-tricot · 2020-08-27T18:24:24Z

dataline-workers/src/main/java/io/dataline/workers/protocol/SingerJsonIterator.java

+import java.io.InputStream;
+import java.util.Iterator;
+
+public class SingerJsonIterator implements Iterator<SingerMessage> {


I suggest you extend Guava's AbstractIterator to build iterators. They make writing iterators less error prone and more readable (you don't have to put tons of code in your ctor.

happy to use the AbstractIterator. I don't think it effects the constructor code though. There's still something we want to check when we first start the iterator that seems like it only fits in the constructor. just pushed what the version with AbstractIterator looks like. Actually not sure that it's better.

wip remove old version of singer protocol clean up add comment clean up remove SingerSyncWorker attempt to clean up the resources improve comment re-used schema from catalog clean up gradle style new way of handling closing tap and target; still some jenk to it share docker command style Update dataline-workers/src/main/java/io/dataline/workers/DefaultSyncWorker.java Co-authored-by: Jared Rhizor <jared@dataline.io> Update dataline-workers/src/main/java/io/dataline/workers/singer/SingerTap.java Co-authored-by: Jared Rhizor <jared@dataline.io> do it remove Docker*Workers for sake of not overcrowding the PR target factory (#118) fix object mapper properties unused import code review feedback dependency inject take while job status iteratore

cgardens force-pushed the cgardens/abstract_this branch from 18112e4 to 8026988 Compare August 26, 2020 08:00

cgardens commented Aug 26, 2020

View reviewed changes

dataline-workers/src/main/java/io/dataline/workers/singer/SingerTap.java Outdated Show resolved Hide resolved

cgardens commented Aug 26, 2020

View reviewed changes

cgardens marked this pull request as ready for review August 26, 2020 19:09

cgardens requested review from jrhizor, michel-tricot and sherifnada August 26, 2020 19:09

cgardens force-pushed the cgardens/abstract_this branch from 7f21c69 to 54392eb Compare August 26, 2020 20:48

jrhizor suggested changes Aug 26, 2020

View reviewed changes

cgardens marked this pull request as draft August 26, 2020 23:25

cgardens commented Aug 27, 2020

View reviewed changes

cgardens marked this pull request as ready for review August 27, 2020 01:11

cgardens requested a review from jrhizor August 27, 2020 01:11

jrhizor approved these changes Aug 27, 2020

View reviewed changes

cgardens commented Aug 27, 2020

View reviewed changes

michel-tricot approved these changes Aug 27, 2020

View reviewed changes

cgardens force-pushed the cgardens/abstract_this branch 2 times, most recently from 8c772c6 to e759cfe Compare August 27, 2020 20:44

cgardens force-pushed the cgardens/abstract_this branch from ce147f7 to 3e42bab Compare August 27, 2020 21:27

cgardens merged commit e53c0e1 into master Aug 27, 2020

cgardens deleted the cgardens/abstract_this branch August 27, 2020 21:32

cgardens mentioned this pull request Aug 28, 2020

Create sync worker interface that isolates tap and target #129

Closed


		target.consume(peakedTapStream, targetConfig, workspacePath);
		try (CloseableConsumer<SingerMessage> consumer =


		import java.util.function.Consumer;

		public interface CloseableConsumer<T> extends Consumer<T>, AutoCloseable {}

		@@ -56,6 +81,9 @@ public static String readFileFromWorkspace(Path workspaceRoot, String fileName)
		}

		public static void cancelHelper(Process workerProcess) {

separate tap and target in sync worker 2 #113

separate tap and target in sync worker 2 #113

Conversation

cgardens commented Aug 26, 2020 • edited

What

How

Checklist

Recommended reading order

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

jrhizor left a 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

jrhizor Aug 26, 2020 • edited

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

jrhizor Aug 26, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cgardens commented Aug 26, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cgardens commented Aug 27, 2020

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

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

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 Aug 27, 2020 • edited

Choose a reason for hiding this comment

cgardens commented Aug 26, 2020 •

edited

jrhizor Aug 26, 2020 •

edited

jrhizor Aug 26, 2020 •

edited

cgardens Aug 27, 2020 •

edited