[FLINK-21362][coordination] Move State.onEnter() logic into constructor #14963

rmetzger · 2021-02-18T15:33:05Z

What is the purpose of the change

Increase robustness by moving the onEnter into the constructor of the states.

flinkbot · 2021-02-18T15:35:56Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 80874c3 (Thu Feb 18 15:35:56 UTC 2021)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2021-02-18T16:07:51Z

CI report:

80874c3 UNKNOWN
d6bbdea Azure: SUCCESS
4a79199 UNKNOWN
eae392b Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

zentol · 2021-02-18T18:23:19Z

...ntime/src/main/java/org/apache/flink/runtime/scheduler/declarative/DeclarativeScheduler.java

-        }
+    <S extends State> void transitionToState(StateFactory<S> targetState) {
+        Preconditions.checkState(
+                state != null, "State transitions are now allowed while construcing a state.");


Suggested change

state != null, "State transitions are now allowed while construcing a state.");

state != null, "State transitions are not allowed while constructing a state.");

zentol · 2021-02-18T18:28:12Z

...me/src/test/java/org/apache/flink/runtime/scheduler/declarative/WaitingForResourcesTest.java

+            new WaitingForResources(ctx, log, RESOURCE_COUNTER, Duration.ZERO);
+            // run delayed actions
+            for (ScheduledRunnable scheduledRunnable : ctx.getScheduledRunnables()) {
+                if (!ctx.hasStateTransition()) {


so basically you want to run actions until you hit a state transition?
I think you could make this more obvious by checking for hasStateTransition after running the action and existing the loop

zentol · 2021-02-18T18:29:44Z

...e/src/test/java/org/apache/flink/runtime/scheduler/declarative/DeclarativeSchedulerTest.java

-        final LifecycleMethodCapturingState firstState = new LifecycleMethodCapturingState();
-        final DummyState secondState = new DummyState();
+        final LifecycleMethodCapturingState firstState =
+                (LifecycleMethodCapturingState) scheduler.getState();


you could simplify this bit by having a test factory that accepts an constructed state. That's usually a nicer way to deal with factories.

zentol · 2021-02-18T18:30:40Z

...e/src/test/java/org/apache/flink/runtime/scheduler/declarative/DeclarativeSchedulerTest.java

+        transitionToState(scheduler, new Created.Factory(scheduler, log));
+    }
+
+    private static <S extends State> void transitionToState(


do we really need this method?

zentol · 2021-02-18T18:39:23Z

...ntime/src/main/java/org/apache/flink/runtime/scheduler/declarative/DeclarativeScheduler.java

+        // transitionToState(WaitingForResources) call, where we overwrite Executing with
+        // WaitingForResources. And there we have it, a deployed execution graph, and a scheduler
+        // that is in WaitingForResources.
+        state = null;


hmm well this is annoying; these kind of loops are driving me crazy in the slot procotol...

what if all goToX methods would schedule the state transition instead, or if transitionToState just schedules it? well that would be annoying during tests wouldn't it...

It would probably work with the tests, since we collect and manually trigger all scheduled executions in the close() method, before checking if the expected state transitions happened.

But I'm not sure if it's worth introducing such behavior for the sake of removing the onEnter() method.
One of the reasons why we agreed to remove onEnter() was because the change revealed a few test inaccuracies. Basically all of these inaccuracies have been found and fixed during the review of the respective tests. The onEnter() prototype was just done on an unreviewed version of the tests.

Right now, WaitingForResources is the only state that triggers state transitions in the constructor.

I would actually propose to revisit the decision to remove the onEnter() method, given the compromises we have to do to make it work (CC @tillrohrmann )

Theoretically, we could remember in the transitionToState method whether we are currently doing a state transition. If this is the case, then we enqueue the targetState in some queue. If not, then we set the flag that we are transitioning into a new state and do the state transition. Once we are complete with a state transition we check the queue for enqueued state transitions. Only if all state transitions are dequeued, we switch the flag back to not doing a state transition.

"The issues are fixed now; we don't need safeguards anymore" could just as well be used as an argument to keep the PR as is and even remove state transitions check. We fixed the one problematic case and could call it a day.

Overall, my impression is that we should not allow immediate state transitions in the constructor, onEnter, or onLeave, in any case. Because all of these result in weird loops/interleaving of state transitions that can lead to subtle issues.
IOW, transitionToState should be an atomic operation that fully completes before another transition can be triggered. Any attempt at triggering a state transition will fail hard.

Hence, whether onEnter exists or not is actually not relevant in this consideration. While it was the sole case where this issue occurred, the underlying issues are unclear and unenforced contracts as to what a State is allowed to do in which methods.

As you have shown in the PR it is pretty easy to safeguard against such occurrences; you'd just need to null the state before calling onLeave.

Alternatively, this would also work, and would be a bit more consolidated:

oldState = state state = null; oldState.onLeave() newState = targetState.getState() newState.onEnter() Preconditions.checkState(state == null); // this will fail if any other state transition occurred in the mean time state = newState;

And as far as I'm concerned that's a pretty tiny cost compared to the risk of testing entirely theoretical scenarios or breaking the state machine.

Maybe we could even make it a bit more explicit by having a dedicated flag which we check before doing the state transition. That way the program would directly fail when a new state transition is triggered and not later when the outer state transition will be continued.

Wouldn't we achieve this "fail fast" behavior if we'd move the Precondition up in Chesnay's proposal?

Preconditions.checkState(state == null); // this will fail if a new state transition is triggered oldState = state state = null; oldState.onLeave() newState = targetState.getState() newState.onEnter() state = newState;

yes that I think that would also work.

maybe a flag is easier to read though

this.isTransitioningState = true this.state.onLeave() this.state = targetState.getState() this.state = onEnter() this.isTransitioningState = false; // maybe this should be done in a finally block...what actually happens if a transition throws an exception?

flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/declarative/RestartingTest.java

tillrohrmann

Two minor comments.

tillrohrmann · 2021-02-19T11:07:56Z

...ntime/src/main/java/org/apache/flink/runtime/scheduler/declarative/DeclarativeScheduler.java

-            state = newState;
-            newState.onEnter();
-        }
+    <S extends State> void transitionToState(StateFactory<S> targetState) {


Can't S extends State be part of the StateFactory?

tillrohrmann · 2021-02-19T11:08:17Z

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/declarative/StateFactory.java

+ *
+ * @param <T> Type of the state.
+ */
+public interface StateFactory<T> {


Should we limit T extends State?

rmetzger · 2021-02-22T13:58:00Z

Thanks for your comments. I believe I addressed all of them.

zentol · 2021-02-22T14:38:26Z

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java

@@ -911,19 +914,26 @@ public void runIfState(State expectedState, Runnable action, Duration delay) {

    // ----------------------------------------------------------------

+    /** Note: Do not call this method from a State constructor. */


Suggested change

/** Note: Do not call this method from a State constructor. */

/** Note: Do not call this method from a State constructor or State#onLeave. */

zentol · 2021-02-22T14:40:01Z

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java

-        if (state != newState) {
+    void transitionToState(StateFactory<?> targetState) {
+        Preconditions.checkState(
+                !isTransitioningState, "Concurrent state transitions are not allowed");


I'm wondering if "concurrent" is the right word; to me it usually implies threading issues and such.

How about "State transitions must not be triggered while another state transition is in progress."

rmetzger · 2021-02-22T15:07:46Z

Thanks a lot for your approval. I'll merge this now!

rmetzger added the review=description? label Feb 18, 2021

rmetzger force-pushed the FLINK-21362-onEnter branch from 80874c3 to d6bbdea Compare February 18, 2021 15:44

rmetzger added the component=Runtime/Coordination label Feb 18, 2021

zentol self-assigned this Feb 18, 2021

zentol reviewed Feb 18, 2021

View reviewed changes

flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/declarative/RestartingTest.java Outdated Show resolved Hide resolved

tillrohrmann reviewed Feb 19, 2021

View reviewed changes

rmetzger added 2 commits February 22, 2021 13:57

[FLINK-21362][coordination] Move State.onEnter() logic into constructor

1fc7028

wip

26b3d4d

rmetzger force-pushed the FLINK-21362-onEnter branch from d6bbdea to 4a79199 Compare February 22, 2021 13:52

addressed all comments

eae392b

rmetzger force-pushed the FLINK-21362-onEnter branch from 4a79199 to eae392b Compare February 22, 2021 13:57

zentol reviewed Feb 22, 2021

View reviewed changes

zentol approved these changes Feb 22, 2021

View reviewed changes

rmetzger closed this in 85879bf Feb 22, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-21362][coordination] Move State.onEnter() logic into constructor #14963

[FLINK-21362][coordination] Move State.onEnter() logic into constructor #14963

rmetzger commented Feb 18, 2021

flinkbot commented Feb 18, 2021

flinkbot commented Feb 18, 2021 •

edited

zentol Feb 18, 2021

zentol Feb 18, 2021

zentol Feb 18, 2021

zentol Feb 18, 2021

zentol Feb 18, 2021

rmetzger Feb 19, 2021

tillrohrmann Feb 19, 2021

zentol Feb 19, 2021 •

edited

tillrohrmann Feb 19, 2021

rmetzger Feb 19, 2021

zentol Feb 19, 2021

zentol Feb 19, 2021

tillrohrmann left a comment

tillrohrmann Feb 19, 2021

tillrohrmann Feb 19, 2021

rmetzger commented Feb 22, 2021

zentol Feb 22, 2021

zentol Feb 22, 2021

rmetzger commented Feb 22, 2021

	state != null, "State transitions are now allowed while construcing a state.");
	state != null, "State transitions are not allowed while constructing a state.");

		@@ -911,19 +914,26 @@ public void runIfState(State expectedState, Runnable action, Duration delay) {

		// ----------------------------------------------------------------

		/** Note: Do not call this method from a State constructor. */

[FLINK-21362][coordination] Move State.onEnter() logic into constructor #14963

[FLINK-21362][coordination] Move State.onEnter() logic into constructor #14963

Conversation

rmetzger commented Feb 18, 2021

What is the purpose of the change

flinkbot commented Feb 18, 2021

Automated Checks

Review Progress

flinkbot commented Feb 18, 2021 • edited

CI report:

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

zentol Feb 19, 2021 • 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

Choose a reason for hiding this comment

tillrohrmann left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rmetzger commented Feb 22, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rmetzger commented Feb 22, 2021

flinkbot commented Feb 18, 2021 •

edited

zentol Feb 19, 2021 •

edited