-
Notifications
You must be signed in to change notification settings - Fork 13.3k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[FLINK-21255] Add WaitingForResources state for DeclarativeScheduler #14852
[FLINK-21255] Add WaitingForResources state for DeclarativeScheduler #14852
Conversation
d4d6eae
to
93b24f6
Compare
@tillrohrmann I split this change into two commits to attribute the scheduler changes to you. Please let me know if you disagree with this. |
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community Automated ChecksLast check on commit b8452a6 (Fri May 28 08:17:51 UTC 2021) Warnings:
Mention the bot in a comment to re-run the automated checks. Review Progress
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 commandsThe @flinkbot bot supports the following commands:
|
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/declarative/StateValidator.java
Outdated
Show resolved
Hide resolved
93b24f6
to
d319b86
Compare
...untime/src/main/java/org/apache/flink/runtime/scheduler/declarative/WaitingForResources.java
Show resolved
Hide resolved
* | ||
* @param newState newState is the state into which the scheduler transitions | ||
*/ | ||
default void onLeave(State newState) {} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I thought we no longer need this?
* State abstraction of the {@link DeclarativeScheduler}. This interface contains all methods every | ||
* state implementation must support. | ||
*/ | ||
interface State { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
note: Had shared classes such as this been added first in a separate PR we could review all states in isolation without having to be based on one another.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good feedback, I didn't consider this. I didn't expect that much review on classes outside the test + I have never migrated a prototype to Flink, thus I need feedback like this (and I need to think harder beforehand myself ;) )
|
||
private final Context context; | ||
|
||
private final Logger logger; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
private final Logger logger; | |
private final Logger log; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is this your personal taste, or is there consensus on the name? (I have the feeling Till prefers logger, you prefer log)
From this anecdotal evidence, there seems to be no clear winner:
≻ git grep "final Logger log;" | wc -l
11
≻ git grep "final Logger logger;" | wc -l
9
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I will change it anyways, because you are reviewing this ;)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
$ git grep "final Logger log =" | wc -l
30
$ git grep "final Logger logger =" | wc -l
9
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Let the grep-games begin ;)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Let's not fight about stuff like this. At the end of the day time spent for these kind of discussions is wasted.
|
||
context.goToExecuting(executionGraph); | ||
} catch (Exception exception) { | ||
logger.error("handling initialization failure", exception); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We should clean this up a bit. I only added it back then because it seems like this transition was never logged, but ideally the context logs all state transitions with appropriate the exception.
|
||
@Override | ||
public void onEnter() { | ||
context.runIfState(this, this::resourceTimeout, Duration.ofSeconds(10L)); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If we merge this as is please file a ticket for making the timeout configurable and increasing the default to a higher value.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can we make it configurable right away? I think this would be good because it is also described like this in the latest state of the FLIP.
ctx.setExpectExecuting(assertNonNull()); | ||
ctx.setHasEnoughResources(() -> true); // make resources available |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ctx.setExpectExecuting(assertNonNull()); | |
ctx.setHasEnoughResources(() -> true); // make resources available | |
ctx.setHasEnoughResources(() -> true); // make resources available | |
ctx.setExpectExecuting(assertNonNull()); |
ctx.setExpectExecuting(assertNonNull()); | ||
wfr.onEnter(); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ctx.setExpectExecuting(assertNonNull()); | |
wfr.onEnter(); | |
wfr.onEnter(); | |
ctx.setExpectExecuting(assertNonNull()); |
ctx.setExpectFinished( | ||
archivedExecutionGraph -> { | ||
assertThat(archivedExecutionGraph.getState(), is(JobStatus.FAILED)); | ||
assertThat(archivedExecutionGraph.getFailureInfo(), notNullValue()); | ||
assertTrue( | ||
archivedExecutionGraph | ||
.getFailureInfo() | ||
.getExceptionAsString() | ||
.contains(testExceptionString)); | ||
}); | ||
wfr.onEnter(); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ctx.setExpectFinished( | |
archivedExecutionGraph -> { | |
assertThat(archivedExecutionGraph.getState(), is(JobStatus.FAILED)); | |
assertThat(archivedExecutionGraph.getFailureInfo(), notNullValue()); | |
assertTrue( | |
archivedExecutionGraph | |
.getFailureInfo() | |
.getExceptionAsString() | |
.contains(testExceptionString)); | |
}); | |
wfr.onEnter(); | |
wfr.onEnter(); | |
ctx.setExpectFinished( | |
archivedExecutionGraph -> { | |
assertThat(archivedExecutionGraph.getState(), is(JobStatus.FAILED)); | |
assertThat(archivedExecutionGraph.getFailureInfo(), notNullValue()); | |
assertTrue( | |
archivedExecutionGraph | |
.getFailureInfo() | |
.getExceptionAsString() | |
.contains(testExceptionString)); | |
}); |
ctx.setExpectFinished( | ||
(archivedExecutionGraph -> { | ||
assertThat(archivedExecutionGraph.getState(), is(JobStatus.CANCELED)); | ||
})); | ||
wfr.onEnter(); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ctx.setExpectFinished( | |
(archivedExecutionGraph -> { | |
assertThat(archivedExecutionGraph.getState(), is(JobStatus.CANCELED)); | |
})); | |
wfr.onEnter(); | |
wfr.onEnter(); | |
ctx.setExpectFinished( | |
(archivedExecutionGraph -> { | |
assertThat(archivedExecutionGraph.getState(), is(JobStatus.CANCELED)); | |
})); |
ctx.setExpectFinished( | ||
(archivedExecutionGraph -> { | ||
assertThat(archivedExecutionGraph.getState(), is(JobStatus.SUSPENDED)); | ||
assertThat(archivedExecutionGraph.getFailureInfo(), notNullValue()); | ||
})); | ||
wfr.onEnter(); | ||
wfr.suspend(new RuntimeException("suspend")); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ctx.setExpectFinished( | |
(archivedExecutionGraph -> { | |
assertThat(archivedExecutionGraph.getState(), is(JobStatus.SUSPENDED)); | |
assertThat(archivedExecutionGraph.getFailureInfo(), notNullValue()); | |
})); | |
wfr.onEnter(); | |
wfr.suspend(new RuntimeException("suspend")); | |
wfr.onEnter(); | |
ctx.setExpectFinished( | |
(archivedExecutionGraph -> { | |
assertThat(archivedExecutionGraph.getState(), is(JobStatus.SUSPENDED)); | |
assertThat(archivedExecutionGraph.getFailureInfo(), notNullValue()); | |
})); | |
wfr.suspend(new RuntimeException("suspend")); |
Thanks a lot for your review. I addressed all comments! |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
+1
Thanks for your review. Will merge now! |
What is the purpose of the change
Declarative Scheduler consists of a number of internal states. This PR is adding the first state of the new scheduler to Flink.
Note that this change is currently not usable as-is, as the other parts of declarative scheduler are not merged yet (See for the prototype this PR is based on: https://github.com/tillrohrmann/flink/tree/declarative-scheduler)
Verifying this change
Does this pull request potentially affect one of the following parts:
@Public(Evolving)
: (yes / no)Documentation
Will be handled in a separate PR.