[Part 12] Calling Source::close, Sink::close as part of stage execution #163

sundargates · 2022-04-05T23:51:44Z

Context

Before both the APIs were not available, we would get rid of the container as part of shutting down the workload. In the new Titus world, we want to gracefully shut down all the resources associated with the stage execution.

Checklist

./gradlew build compiles code correctly
Added new tests where applicable
./gradlew test passes all tests
Extended README or added javadocs where applicable
Added copyright headers for new files from CONTRIBUTING.md

sundargates · 2022-04-06T00:00:30Z

Only the last commit in this diff needs to be reviewed. It's stacked on top of Part 11.

hmitnflx · 2022-04-06T01:04:36Z

mantis-runtime/src/main/java/io/mantisrx/runtime/executor/WorkerPublisherRemoteObservable.java

        this.propService = ServiceRegistry.INSTANCE.getPropertiesService();
        this.jobName = jobName;
    }

    @SuppressWarnings( {"rawtypes", "unchecked"})
    @Override
-    public void start(final StageConfig<T, R> stage, Observable<Observable<R>> toServe) {
+    public void start(final StageConfig<?, T> stage, Observable<Observable<T>> toServe) {


This is an interesting change, wasn't expecting it.
LGTM but would appreciate a closer look from one more person here. perhaps @codyrioux or @Andyz26

I just took a look through and this appears to be fine. I like that it is more explicit about the fact that we are not using the first type parameter in the StageConfig in these locations.

codyrioux

I appreciate that you took the time to leverage functional features and get rid of the ad-hoc interface implementations.

Before both the APIs were not available, we would get rid of the container as part of shutting down the workload. In the new Titus world, we want to gracefully shut down all the resources associated with the stage execution.

sundargates · 2022-04-06T18:06:53Z

mantis-runtime/src/main/java/io/mantisrx/runtime/executor/SinkPublisher.java

-public class SinkPublisher<T, R> implements WorkerPublisher<T, R> {
+/**
+ * Implementation that publishes the results of a stage to a sink such as an SSE port.
+ * @param <T>


nit: add documentation.

sundargates · 2022-04-06T18:07:17Z

mantis-runtime/src/main/java/io/mantisrx/runtime/executor/SinkPublisher.java

+    private final Action0 onUnsubscribeAction;
+    private final Action0 observableOnCompleteCallback;
+    private final Action1<Throwable> observableOnErrorCallback;
+    private Subscription eagerSubscription;


nit: separate the state from the rest of the fields.

sundargates · 2022-04-06T18:08:12Z

mantis-runtime/src/main/java/io/mantisrx/runtime/executor/SinkPublisher.java

+        } finally {
+            if (eagerSubscription != null) {
+                eagerSubscription.unsubscribe();
+                eagerSubscription = null;


should sink also be set to null eventually?

sundargates · 2022-04-06T18:10:08Z

mantis-runtime/src/main/java/io/mantisrx/runtime/executor/WorkerConsumerRemoteObservable.java

@@ -100,6 +100,7 @@ private void registerMetrics(Metrics metrics) {
    }

    @Override
-    public void stop() {}
+    public void close() throws IOException {


why is there nothing to close for this executor?

I'm guessing the Observable handles most of the behavior under the hood when the unsubscribe occurs. Or just depends on the whole JVM being blown away.

sundargates · 2022-04-06T18:10:25Z

mantis-runtime/src/main/java/io/mantisrx/runtime/executor/WorkerPublisherRemoteObservable.java

-    private String name;
-    private int serverPort;
+    private final String name;
+    private final int serverPort;
    private RemoteRxServer server;


separate state from the fields

sundargates · 2022-04-06T18:11:52Z

...r/mantis-server-worker/src/main/java/io/mantisrx/server/worker/DataDroppedPayloadSetter.java

+
+    @Override
+    public void close() throws IOException {
+        if (future != null) {


reset the state in the finally block to make close() idempotent

sundargates · 2022-04-06T18:12:17Z

...mantis-server-worker/src/main/java/io/mantisrx/server/worker/ResourceUsagePayloadSetter.java

+
+    @Override
+    public void close() throws IOException {
+        executor.shutdownNow();


Set executor to null

sundargates · 2022-04-06T18:13:39Z

...er-worker/src/main/java/io/mantisrx/server/worker/WorkerExecutionOperationsNetworkStage.java

@@ -129,6 +136,7 @@ public WorkerExecutionOperationsNetworkStage(
        String locateSpectatorRegistry =
                ServiceRegistry.INSTANCE.getPropertiesService().getStringValue("mantis.worker.locate.spectator.registry", "true");
        lookupSpectatorRegistry = Boolean.valueOf(locateSpectatorRegistry);
+        scheduledExecutorService = new ScheduledThreadPoolExecutor(1);


name the thread pool factory please.

sundargates · 2022-04-06T18:15:33Z

...er-worker/src/main/java/io/mantisrx/server/worker/WorkerExecutionOperationsNetworkStage.java

@@ -640,7 +653,7 @@ private WorkerConsumer connectToObservableAtPreviousStages(Observable<JobSchedul
    }

    @Override
-    public void shutdownStage() {
+    public void shutdownStage() throws IOException  {


make sure shutdownStage cannot be called when executeStage is being executed.

sundargates · 2022-04-06T18:15:53Z

...er-worker/src/main/java/io/mantisrx/server/worker/WorkerExecutionOperationsNetworkStage.java

@@ -650,6 +663,7 @@ public void shutdownStage() {
            }
        }

-        System.exit(0);
+        Closeables.combine(closeables).close();


clean the closeables to make this idempotent.

codyrioux · 2022-04-06T20:06:37Z

mantis-runtime/src/main/java/io/mantisrx/runtime/executor/WorkerConsumerRemoteObservable.java

@@ -100,6 +100,7 @@ private void registerMetrics(Metrics metrics) {
    }

    @Override
-    public void stop() {}
+    public void close() throws IOException {


I'm guessing the Observable handles most of the behavior under the hood when the unsubscribe occurs. Or just depends on the whole JVM being blown away.

sundargates requested review from codyrioux, nickmahilani, jeffchao, piygoyal, calvin681, Andyz26, hmitnflx, markcho and liuml07 as code owners April 5, 2022 23:51

hmitnflx approved these changes Apr 6, 2022

View reviewed changes

calvin681 approved these changes Apr 6, 2022

View reviewed changes

codyrioux approved these changes Apr 6, 2022

View reviewed changes

sundargates force-pushed the sundaram/titus_part12 branch from 9d9819e to 5d380c1 Compare April 6, 2022 17:55

sundargates commented Apr 6, 2022

View reviewed changes

Addressing comments

d83e721

codyrioux approved these changes Apr 6, 2022

View reviewed changes

calvin681 approved these changes Apr 10, 2022

View reviewed changes

sundargates merged commit d8c82e1 into Netflix:master Apr 13, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Part 12] Calling Source::close, Sink::close as part of stage execution #163

[Part 12] Calling Source::close, Sink::close as part of stage execution #163

sundargates commented Apr 5, 2022

sundargates commented Apr 6, 2022

hmitnflx Apr 6, 2022

codyrioux Apr 6, 2022

codyrioux left a comment

sundargates Apr 6, 2022

sundargates Apr 6, 2022

sundargates Apr 6, 2022

sundargates Apr 6, 2022

sundargates Apr 6, 2022

sundargates Apr 6, 2022

sundargates Apr 6, 2022

codyrioux Apr 6, 2022

sundargates Apr 6, 2022

sundargates Apr 6, 2022

sundargates Apr 6, 2022

sundargates Apr 6, 2022

sundargates Apr 6, 2022

sundargates Apr 6, 2022

codyrioux Apr 6, 2022

[Part 12] Calling Source::close, Sink::close as part of stage execution #163

[Part 12] Calling Source::close, Sink::close as part of stage execution #163

Conversation

sundargates commented Apr 5, 2022

Context

Checklist

sundargates commented Apr 6, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codyrioux 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

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