Fix unexpected sink side unsubscribe behavior #562

Andyz26 · 2023-10-06T23:31:24Z

Context

There is a race condition on the subscription handler where the current state can be empty if the subscription is established before the handler service is started, which causes non-perpetual jobs to silently fail at startup.

Added handling + logs to this case.
Added extra test in integration test.
Removed sinkSubscriptionHandlerFactory on TaskExecutor level (not used anymore).

Checklist

./gradlew build compiles code correctly
Added new tests where applicable
./gradlew test passes all tests
Extended README or added javadocs where applicable

github-actions · 2023-10-06T23:45:01Z

Test Results

545 tests +1 537 ✔️ +1 6m 46s ⏱️ - 1m 8s
128 suites ±0     8 💤 ±0
128 files ±0     0 ❌ ±0

Results for commit 7210270. ± Comparison against base commit 49cd280.

This pull request removes 1 and adds 2 tests. Note that renamed tests count towards both.

TestContainerHelloWorld ‑ helloWorld

TestContainerHelloWorld ‑ testQuickSubmitJob
TestContainerHelloWorld ‑ testRegularSubmitJob

♻️ This comment has been updated with latest results.

liuml07

LGTM but a second review is appreciated. I have limited context.

liuml07 · 2023-10-06T23:58:23Z

...is-runtime-loader/src/main/java/io/mantisrx/runtime/loader/SubscriptionStateHandlerImpl.java

@@ -53,7 +53,9 @@ class SubscriptionStateHandlerImpl extends AbstractScheduledService implements S

    @Override
    public void startUp() {
-        currentState.set(SubscriptionState.of(clock));
+        if (currentState.get() == null) {


is this if check needed?

seems like you are expecting scenarios where startUp can be called multiple times?

The currentState can be set at startUp or the first update from subscribe event (currently there is no set at subscribe so if subscribe happens before startup it hits NPE).

We need to ensure that none of the members of the class are called before startUp, as this would violate the semantics of a Service class. To prevent this, I suggest setting up preconditions in the listener methods rather than working around the issue.

I think it might be a better idea to start the subscription handler in

private void setupSubscriptionStateHandler(ExecuteStageRequest executeStageRequest) {

liuml07 · 2023-10-06T23:59:35Z

...is-runtime-loader/src/main/java/io/mantisrx/runtime/loader/SubscriptionStateHandlerImpl.java

@@ -81,11 +83,20 @@ protected void runOneIteration() {

    @Override
    public void onSinkUnsubscribed() {
+        if (currentState.get() == null) {


These two operations in the method i.e. get-then-update are not atomic, is it ok?

currentState should never be unset back to null in this instance, thus no race on getting non-null then reading null.

liuml07 · 2023-10-07T00:01:50Z

...is-runtime-loader/src/main/java/io/mantisrx/runtime/loader/SubscriptionStateHandlerImpl.java

        currentState.updateAndGet(SubscriptionState::onSinkUnsubscribed);
    }

    @Override
    public void onSinkSubscribed() {
+        // sink subscription can happen before the startup of this service.
+        if (currentState.get() == null) {
+            currentState.compareAndSet(null, SubscriptionState.of(clock));


same question, I was thinking the if-check is not necessary. currentState.compareAndSet(null, SubscriptionState.of(clock)); works just fine.

yeah it's not for thread safety but just a minor save to avoid the extra SubscriptionState.of(clock) creation.

liuml07 · 2023-10-07T00:03:29Z

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

-                        if (onUnsubscribeAction != null)
-                            onUnsubscribeAction.call();
-                    }
+                .doOnCompleted(() -> logger.info("Sink observable subscription completed."))


nit: is other information useful in the log like subscribe etc.

sundargates · 2023-10-09T19:56:52Z

...is-runtime-loader/src/main/java/io/mantisrx/runtime/loader/SubscriptionStateHandlerImpl.java

@@ -53,7 +53,10 @@ class SubscriptionStateHandlerImpl extends AbstractScheduledService implements S

    @Override
    public void startUp() {
-        currentState.set(SubscriptionState.of(clock));
+        if (currentState.get() == null) {


I would convert this into a Precondition instead.

I don't think it has to fail if the service got started again. Let me remove the check here directly (atomic set with null should be sufficient.)

calvin681 · 2023-10-09T20:02:32Z

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

@@ -478,6 +478,12 @@ private void setupSubscriptionStateHandler(ExecuteStageRequest executeStageReque
        };

        this.subscriptionStateHandler = subscriptionStateHandler;
+        try {
+            this.subscriptionStateHandler.startAsync().awaitRunning(Duration.of(5, ChronoUnit.SECONDS));


This should be TimeUnit no?

TimeUnit is in java.util.concurrent and this Duration.of requires the temporal unit interface which is from java.time.

fix sink unsubscrib

3643ed7

Andyz26 requested review from nickmahilani, jeffchao, piygoyal, calvin681, sundargates, hmitnflx, markcho, liuml07 and fdc-ntflx as code owners October 6, 2023 23:31

Andyz26 had a problem deploying to Integrate Pull Request October 6, 2023 23:31 — with GitHub Actions Failure

Andyz26 changed the title ~~fix sink unsubscrib~~ Fix unexpected sink side un-subscription Oct 6, 2023

Andyz26 changed the title ~~Fix unexpected sink side un-subscription~~ Fix unexpected sink side unsubscribe behavior Oct 6, 2023

liuml07 approved these changes Oct 7, 2023

View reviewed changes

comments

76c22dc

Andyz26 had a problem deploying to Integrate Pull Request October 9, 2023 19:49 — with GitHub Actions Failure

sundargates reviewed Oct 9, 2023

View reviewed changes

calvin681 reviewed Oct 9, 2023

View reviewed changes

comment

7210270

Andyz26 had a problem deploying to Integrate Pull Request October 9, 2023 21:41 — with GitHub Actions Failure

Andyz26 merged commit d94a7b0 into master Oct 9, 2023
1 of 2 checks passed

Andyz26 deleted the andyz/fixSinkUnsubError branch October 9, 2023 23:07

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix unexpected sink side unsubscribe behavior #562

Fix unexpected sink side unsubscribe behavior #562

Andyz26 commented Oct 6, 2023

github-actions bot commented Oct 6, 2023 •

edited

Loading

liuml07 left a comment

liuml07 Oct 6, 2023

calvin681 Oct 7, 2023

Andyz26 Oct 8, 2023

sundargates Oct 8, 2023

sundargates Oct 8, 2023

liuml07 Oct 6, 2023

Andyz26 Oct 8, 2023

liuml07 Oct 7, 2023

Andyz26 Oct 8, 2023

liuml07 Oct 7, 2023

Andyz26 Oct 8, 2023

sundargates Oct 9, 2023

Andyz26 Oct 9, 2023

calvin681 Oct 9, 2023

Andyz26 Oct 9, 2023

Fix unexpected sink side unsubscribe behavior #562

Fix unexpected sink side unsubscribe behavior #562

Conversation

Andyz26 commented Oct 6, 2023

Context

Checklist

github-actions bot commented Oct 6, 2023 • edited Loading

Test Results

liuml07 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

github-actions bot commented Oct 6, 2023 •

edited

Loading