You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/03/11 19:27:35 UTC
[GitHub] [flink] rmetzger opened a new pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
rmetzger opened a new pull request #15159:
URL: https://github.com/apache/flink/pull/15159
## What is the purpose of the change
Adjust timeout behavior of adaptive scheduler according to the discussion in Jira.
## Verifying this change
Covered by new and existing tests.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499)
* 001d7c8625cf8ac8574fab93ee122d73d2722cd7 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #15159:
URL: https://github.com/apache/flink/pull/15159#discussion_r598722325
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java
##########
@@ -178,12 +344,24 @@ void setExpectCreatingExecutionGraph() {
creatingExecutionGraphStateValidator.expectInput(none -> {});
}
- void runScheduledTasks() {
- for (ScheduledTask<Void> scheduledTask : scheduledTasks) {
- scheduledTask.execute();
+ void runScheduledTasks(long untilDelay) {
+ Iterable<ScheduledTask<Void>> copyOfScheduledTasks = new ArrayList<>(scheduledTasks);
+ Iterator<ScheduledTask<Void>> scheduledTaskIterator = copyOfScheduledTasks.iterator();
+ while (scheduledTaskIterator.hasNext()) {
+ ScheduledTask<Void> scheduledTask = scheduledTaskIterator.next();
+ if (scheduledTask.getDelay(TimeUnit.MILLISECONDS) <= untilDelay) {
+ scheduledTask.execute();
+ if (!scheduledTask.isPeriodic()) {
+ scheduledTaskIterator.remove();
+ }
+ }
Review comment:
But they will be executed with the next call to runScheduledTasks(). I want to only execute tasks which are scheduled up to now.
Immediately executing newly scheduled tasks can lead to loops, for example WaitingForResources.checkDesiredOrSufficientResourcesAvailable() schedules itself in some circumstances.
But there was actually a flaw in my code: The removal mechanism doesn't work as intended: It removes elements from the copy. My intention was actually to execute non periodic tasks only once. I've addressed this now in the code.
If you disagree with any of these changes, I'm happy to address this in a hotfix alongside another pull request or a separate PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 001d7c8625cf8ac8574fab93ee122d73d2722cd7 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732)
* aab6b14a76ca84b1fab62d698c2c5914a8f37807 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on a change in pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #15159:
URL: https://github.com/apache/flink/pull/15159#discussion_r598740164
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java
##########
@@ -178,12 +344,24 @@ void setExpectCreatingExecutionGraph() {
creatingExecutionGraphStateValidator.expectInput(none -> {});
}
- void runScheduledTasks() {
- for (ScheduledTask<Void> scheduledTask : scheduledTasks) {
- scheduledTask.execute();
+ void runScheduledTasks(long untilDelay) {
+ Iterable<ScheduledTask<Void>> copyOfScheduledTasks = new ArrayList<>(scheduledTasks);
+ Iterator<ScheduledTask<Void>> scheduledTaskIterator = copyOfScheduledTasks.iterator();
+ while (scheduledTaskIterator.hasNext()) {
+ ScheduledTask<Void> scheduledTask = scheduledTaskIterator.next();
+ if (scheduledTask.getDelay(TimeUnit.MILLISECONDS) <= untilDelay) {
+ scheduledTask.execute();
+ if (!scheduledTask.isPeriodic()) {
+ scheduledTaskIterator.remove();
+ }
+ }
Review comment:
`untilDelay` should define an upper bound for the tasks to execute if there is no tasks with a delay of `0` which enqueues itself over and over again.
The problem with this implementation is that this might simulate an interleaving which cannot happen with a proper executor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
}, {
"hash" : "30724c1ba85f417b86d00337257b2180f03d376a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792",
"triggerID" : "30724c1ba85f417b86d00337257b2180f03d376a",
"triggerType" : "PUSH"
}, {
"hash" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818",
"triggerID" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"triggerType" : "PUSH"
}, {
"hash" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15104",
"triggerID" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"triggerType" : "PUSH"
}, {
"hash" : "989abd48b62ba85081f9b4cb4492dbff97fb7639",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15147",
"triggerID" : "989abd48b62ba85081f9b4cb4492dbff97fb7639",
"triggerType" : "PUSH"
}, {
"hash" : "6651b8a6ba48a81572de8d22e43069a353cabf2e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15154",
"triggerID" : "6651b8a6ba48a81572de8d22e43069a353cabf2e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6651b8a6ba48a81572de8d22e43069a353cabf2e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15154)
<details>
<summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
}, {
"hash" : "30724c1ba85f417b86d00337257b2180f03d376a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "30724c1ba85f417b86d00337257b2180f03d376a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* aab6b14a76ca84b1fab62d698c2c5914a8f37807 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735)
* 30724c1ba85f417b86d00337257b2180f03d376a UNKNOWN
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
rmetzger commented on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-799691648
Thanks a lot for your feedback. I believe I've addressed all items!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
}, {
"hash" : "30724c1ba85f417b86d00337257b2180f03d376a",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792",
"triggerID" : "30724c1ba85f417b86d00337257b2180f03d376a",
"triggerType" : "PUSH"
}, {
"hash" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818",
"triggerID" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 30724c1ba85f417b86d00337257b2180f03d376a Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792)
* 57d9b95fac95a649bd8295c294b77aeee5df12fc Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
}, {
"hash" : "30724c1ba85f417b86d00337257b2180f03d376a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792",
"triggerID" : "30724c1ba85f417b86d00337257b2180f03d376a",
"triggerType" : "PUSH"
}, {
"hash" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818",
"triggerID" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"triggerType" : "PUSH"
}, {
"hash" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15104",
"triggerID" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 57d9b95fac95a649bd8295c294b77aeee5df12fc Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818)
* 28bf753c82199c68f31a19234bc85b7fc03dc9cb Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15104)
<details>
<summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499)
* 001d7c8625cf8ac8574fab93ee122d73d2722cd7 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732)
* aab6b14a76ca84b1fab62d698c2c5914a8f37807 UNKNOWN
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
rmetzger commented on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-803846663
Thanks a lot for your review! I addressed all your comments (however skipping optional comments for the sake of time). I also rebased this change onto the latest changes on master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
}, {
"hash" : "30724c1ba85f417b86d00337257b2180f03d376a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792",
"triggerID" : "30724c1ba85f417b86d00337257b2180f03d376a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* aab6b14a76ca84b1fab62d698c2c5914a8f37807 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735)
* 30724c1ba85f417b86d00337257b2180f03d376a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rmetzger closed pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
rmetzger closed pull request #15159:
URL: https://github.com/apache/flink/pull/15159
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f UNKNOWN
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on a change in pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #15159:
URL: https://github.com/apache/flink/pull/15159#discussion_r597613334
##########
File path: docs/layouts/shortcodes/generated/all_jobmanager_section.html
##########
@@ -14,11 +14,17 @@
<td>Integer</td>
<td>Configure the minimum increase in parallelism for a job to scale up.</td>
</tr>
+ <tr>
+ <td><h5>jobmanager.adaptive-scheduler.resource-stabilization-timeout</h5></td>
+ <td style="word-wrap: break-word;">10 s</td>
+ <td>Duration</td>
+ <td>The resource stabilization timeout defines the time the JobManager will wait if fewer than the required resources are available, but sufficient resources for execution are there. Once this timeout has passed, the job will start executing with the available resources, or fail, if the resources are not sufficient. The timeout starts as soon as there are sufficient resources available for execution.<br />If <span markdown="span">`scheduler-mode`</span> is configured to <span markdown="span">`REACTIVE`</span>, this configuration value will default to 0, so that jobs are starting immediately with the available resources.</td>
Review comment:
Yes, this can happen but to me the description reads a bit like we have seen once that the resources are sufficient and then we trigger the timeout and no matter what when the timeout triggers we try to schedule the job. Instead, the system should still check whether it has sufficient resources and only try to schedule the job if this is still true.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on a change in pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #15159:
URL: https://github.com/apache/flink/pull/15159#discussion_r596773120
##########
File path: docs/layouts/shortcodes/generated/all_jobmanager_section.html
##########
@@ -14,11 +14,17 @@
<td>Integer</td>
<td>Configure the minimum increase in parallelism for a job to scale up.</td>
</tr>
+ <tr>
+ <td><h5>jobmanager.adaptive-scheduler.resource-stabilization-timeout</h5></td>
+ <td style="word-wrap: break-word;">10 s</td>
+ <td>Duration</td>
+ <td>The resource stabilization timeout defines the time the JobManager will wait if fewer than the required resources are available, but sufficient resources for execution are there. Once this timeout has passed, the job will start executing with the available resources, or fail, if the resources are not sufficient. The timeout starts as soon as there are sufficient resources available for execution.<br />If <span markdown="span">`scheduler-mode`</span> is configured to <span markdown="span">`REACTIVE`</span>, this configuration value will default to 0, so that jobs are starting immediately with the available resources.</td>
+ </tr>
<tr>
<td><h5>jobmanager.adaptive-scheduler.resource-wait-timeout</h5></td>
<td style="word-wrap: break-word;">10 s</td>
<td>Duration</td>
- <td>The maximum time the JobManager will wait to acquire all required resources after a job submission or restart. Once elapsed it will try to run the job with a lower parallelism, or fail if the minimum amount of resources could not be acquired.<br />Increasing this value will make the cluster more resilient against temporary resources shortages (e.g., there is more time for a failed TaskManager to be restarted), while decreasing this value reduces downtime of a job (provided that enough slots are available to still run the job).</td>
+ <td>The maximum time the JobManager will wait to acquire all required resources after a job submission or restart. Once elapsed it will try to run the job with a lower parallelism, or fail if the minimum amount of resources could not be acquired.<br />Increasing this value will make the cluster more resilient against temporary resources shortages (e.g., there is more time for a failed TaskManager to be restarted), while decreasing this value reduces downtime of a job (provided that enough slots are available to still run the job).<br />Setting a negative duration will disable the resource timeout: The JobManager will wait indefinitely for resources to appear.<br />If <span markdown="span">`scheduler-mode`</span> is configured to <span markdown="span">`REACTIVE`</span>, this configuration value will default to a negative value to disable the resource timeout.</td>
Review comment:
Is it still correct to say that decreasing this value will reduce the downtime of jobs? If we have sufficient resources, then `jobmanager.adaptive-scheduler.resource-stabilization-timeout` should be the decisive factor how long the system waits to restart the job.
##########
File path: docs/layouts/shortcodes/generated/all_jobmanager_section.html
##########
@@ -14,11 +14,17 @@
<td>Integer</td>
<td>Configure the minimum increase in parallelism for a job to scale up.</td>
</tr>
+ <tr>
+ <td><h5>jobmanager.adaptive-scheduler.resource-stabilization-timeout</h5></td>
+ <td style="word-wrap: break-word;">10 s</td>
+ <td>Duration</td>
+ <td>The resource stabilization timeout defines the time the JobManager will wait if fewer than the required resources are available, but sufficient resources for execution are there. Once this timeout has passed, the job will start executing with the available resources, or fail, if the resources are not sufficient. The timeout starts as soon as there are sufficient resources available for execution.<br />If <span markdown="span">`scheduler-mode`</span> is configured to <span markdown="span">`REACTIVE`</span>, this configuration value will default to 0, so that jobs are starting immediately with the available resources.</td>
Review comment:
Maybe: `The resource stabilization timeout defines the time the JobManager will wait if fewer than the desired but sufficient resources are available.`
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java
##########
@@ -90,13 +107,40 @@ public Logger getLogger() {
@Override
public void notifyNewResourcesAvailable() {
- if (context.hasEnoughResources(desiredResources)) {
+ checkDesiredOrSufficientResourcesAvailable();
+ }
+
+ private void checkDesiredOrSufficientResourcesAvailable() {
+ if (context.hasDesiredResources(desiredResources)) {
createExecutionGraphWithAvailableResources();
+ return;
}
+
+ if (context.hasSufficientResources()) {
+ Deadline deadline = initializeOrGetResourceStabilizationDeadline();
+ if (deadline.isOverdue()) {
+ createExecutionGraphWithAvailableResources();
+ } else {
+ // schedule next resource check
+ context.runIfState(
+ this,
+ this::checkDesiredOrSufficientResourcesAvailable,
+ deadline.timeLeft());
+ }
+ }
Review comment:
Shouldn't we clear the resource stabilization deadline if we no longer have sufficient resources so that we can start a new deadline once we have sufficient resources again?
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java
##########
@@ -110,41 +117,187 @@ public void testNotEnoughResources() throws Exception {
@Test
public void testNotifyNewResourcesAvailable() throws Exception {
try (MockContext ctx = new MockContext()) {
- ctx.setHasEnoughResources(() -> false); // initially, not enough resources
+ ctx.setHasDesiredResources(() -> false); // initially, not enough resources
WaitingForResources wfr =
- new WaitingForResources(ctx, log, RESOURCE_COUNTER, Duration.ZERO);
- ctx.setHasEnoughResources(() -> true); // make resources available
+ new WaitingForResources(
+ ctx, log, RESOURCE_COUNTER, Duration.ZERO, STABILIZATION_TIMEOUT);
+ ctx.setHasDesiredResources(() -> true); // make resources available
ctx.setExpectExecuting(assertNonNull());
wfr.notifyNewResourcesAvailable(); // .. and notify
}
}
@Test
- public void testResourceTimeout() throws Exception {
+ public void testSchedulingWithSufficientResourcesAndNoStabilizationTimeout() throws Exception {
+ try (MockContext ctx = new MockContext()) {
+
+ Duration noStabilizationTimeout = Duration.ofMillis(0);
+ WaitingForResources wfr =
+ new WaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ Duration.ofSeconds(1000),
+ noStabilizationTimeout);
+
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> true);
+ ctx.setExpectExecuting(assertNonNull());
+ wfr.notifyNewResourcesAvailable();
+ }
+ }
+
+ @Test
+ public void testNoSchedulingIfStabilizationTimeoutIsConfigured() throws Exception {
try (MockContext ctx = new MockContext()) {
- ctx.setHasEnoughResources(() -> false);
+
+ Duration stabilizationTimeout = Duration.ofMillis(50000);
+
WaitingForResources wfr =
- new WaitingForResources(ctx, log, RESOURCE_COUNTER, Duration.ZERO);
+ new WaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ Duration.ofSeconds(1000),
+ stabilizationTimeout);
+
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> true);
+ wfr.notifyNewResourcesAvailable();
+ // we are not triggering the scheduled tasks, to simulate a long stabilization timeout
+
+ assertThat(ctx.hasStateTransition(), is(false));
+ }
+ }
+
+ @Test
+ public void testSchedulingIfStabilizationTimeoutIsConfigured() throws Exception {
+ try (MockContext ctx = new MockContext()) {
+
+ Duration initialResourceTimeout = Duration.ofMillis(120948);
+ Duration stabilizationTimeout = Duration.ofMillis(50000);
+
+ TestingWaitingForResources wfr =
+ new TestingWaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ initialResourceTimeout,
+ stabilizationTimeout);
+
+ // not enough resources available
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> false);
+
+ assertNoStateTransitionsAfterExecutingRunnables(ctx, wfr);
+
+ // sufficient resources available
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> true);
+ wfr.setTestDeadline(Deadline.fromNow(Duration.ofDays(600)));
ctx.setExpectExecuting(assertNonNull());
+ // start countdown
+ wfr.notifyNewResourcesAvailable();
+
+ assertNoStateTransitionsAfterExecutingRunnables(ctx, wfr);
+
+ // advance time to now
+ wfr.setTestDeadline(Deadline.now());
- // immediately execute all scheduled runnables
- assertThat(ctx.getScheduledRunnables().size(), greaterThan(0));
+ // execute all runnables and trigger expected state transition
for (ScheduledRunnable scheduledRunnable : ctx.getScheduledRunnables()) {
- if (scheduledRunnable.getExpectedState() == wfr) {
+ if (scheduledRunnable.getExpectedState() == wfr
+ && !scheduledRunnable.getDelay().equals(initialResourceTimeout)
+ && !scheduledRunnable.getDelay().equals(Duration.ZERO)) {
scheduledRunnable.runAction();
}
}
}
}
+ private static void assertNoStateTransitionsAfterExecutingRunnables(
+ MockContext ctx, WaitingForResources wfr) {
+ Iterator<ScheduledRunnable> runnableIterator = ctx.getScheduledRunnables().iterator();
+ while (runnableIterator.hasNext()) {
+ ScheduledRunnable scheduledRunnable = runnableIterator.next();
+ if (scheduledRunnable.getExpectedState() == wfr
+ && scheduledRunnable.getDeadline().isOverdue()) {
+ scheduledRunnable.runAction();
+ runnableIterator.remove();
+ }
+ }
+ assertThat(ctx.hasStateTransition, is(false));
+ }
+
+ private static class TestingWaitingForResources extends WaitingForResources {
+
+ private Deadline testDeadline;
+
+ TestingWaitingForResources(
+ Context context,
+ Logger log,
+ ResourceCounter desiredResources,
+ Duration initialResourceAllocationTimeout,
+ Duration resourceStabilizationTimeout) {
+ super(
+ context,
+ log,
+ desiredResources,
+ initialResourceAllocationTimeout,
+ resourceStabilizationTimeout);
+ }
+
+ @Override
+ protected Deadline initializeOrGetResourceStabilizationDeadline() {
+ return testDeadline;
+ }
+
+ public void setTestDeadline(Deadline testDeadline) {
+ this.testDeadline = testDeadline;
+ }
+ }
+
+ @Test
+ public void testNoStateTransitionOnNoResourceTimeout() throws Exception {
+ try (MockContext ctx = new MockContext()) {
+ ctx.setHasDesiredResources(() -> false);
+ WaitingForResources wfr =
+ new WaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ Duration.ofMillis(-1),
+ STABILIZATION_TIMEOUT);
+
+ executeAllScheduledRunnables(ctx, wfr);
+
+ assertThat(ctx.hasStateTransition(), is(false));
+ }
+ }
+
+ @Test
+ public void testStateTransitionOnResourceTimeout() throws Exception {
+ try (MockContext ctx = new MockContext()) {
+ ctx.setHasDesiredResources(() -> false);
+ WaitingForResources wfr =
+ new WaitingForResources(
+ ctx, log, RESOURCE_COUNTER, Duration.ZERO, STABILIZATION_TIMEOUT);
+
+ ctx.setExpectExecuting(assertNonNull());
+
+ executeAllScheduledRunnables(ctx, wfr);
+ }
+ }
+
Review comment:
I think we are lacking a test where we have sufficient resources, then again no sufficient resources and then restart the timeout once we have sufficient resources again (e.g. testing that the sufficient resource timeout can be restarted).
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java
##########
@@ -90,13 +107,40 @@ public Logger getLogger() {
@Override
public void notifyNewResourcesAvailable() {
- if (context.hasEnoughResources(desiredResources)) {
+ checkDesiredOrSufficientResourcesAvailable();
+ }
+
+ private void checkDesiredOrSufficientResourcesAvailable() {
+ if (context.hasDesiredResources(desiredResources)) {
createExecutionGraphWithAvailableResources();
+ return;
}
+
+ if (context.hasSufficientResources()) {
Review comment:
One minor idea could be to unify `hasDesiredResources` and `hasSufficientResources`. I think both methods do effectively the same and could signal the different results with a more expressive return type.
##########
File path: flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
##########
@@ -411,6 +411,35 @@
.text(
"Increasing this value will make the cluster more resilient against temporary resources shortages (e.g., there is more time for a failed TaskManager to be restarted), "
+ "while decreasing this value reduces downtime of a job (provided that enough slots are available to still run the job).")
+ .linebreak()
+ .text(
+ "Setting a negative duration will disable the resource timeout: The JobManager will wait indefinitely for resources to appear.")
+ .linebreak()
+ .text(
+ "If %s is configured to %s, this configuration value will default to a negative value to disable the resource timeout.",
+ code(SCHEDULER_MODE.key()),
+ code(SchedulerExecutionMode.REACTIVE.name()))
+ .build());
+
+ @Documentation.Section({
+ Documentation.Sections.EXPERT_SCHEDULING,
+ Documentation.Sections.ALL_JOB_MANAGER
+ })
+ public static final ConfigOption<Duration> RESOURCE_STABILIZATION_TIMEOUT =
+ key("jobmanager.adaptive-scheduler.resource-stabilization-timeout")
+ .durationType()
+ .defaultValue(RESOURCE_WAIT_TIMEOUT.defaultValue())
+ .withDescription(
+ Description.builder()
+ .text(
+ "The resource stabilization timeout defines the time the JobManager will wait "
Review comment:
I like Chesnay's proposal.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactory.java
##########
@@ -193,4 +197,37 @@ public static DefaultSlotPoolServiceSchedulerFactory fromConfiguration(
return new DefaultSlotPoolServiceSchedulerFactory(
slotPoolServiceFactory, schedulerNGFactory);
}
+
+ private static AdaptiveSchedulerFactory getAdaptiveSchedulerFactoryFromConfiguration(
+ Configuration configuration) {
+ Duration initialResourceAllocationTimeout =
+ returnValueOrReplaceDefaultIfReactiveMode(
+ configuration,
+ JobManagerOptions.RESOURCE_WAIT_TIMEOUT,
+ Duration.ofMillis(-1));
+ Duration resourceStabilizationTimeout =
+ returnValueOrReplaceDefaultIfReactiveMode(
+ configuration,
+ JobManagerOptions.RESOURCE_STABILIZATION_TIMEOUT,
+ Duration.ZERO);
Review comment:
Would it be simpler to define the defaults based on `configuration.get(JobManagerOptions.SCHEDULER_MODE)` here and then simply call `configuration.getOptional(option).orElseGet(() -> defaultValue)` where `defaultValue` has been specified before?
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java
##########
@@ -110,41 +117,187 @@ public void testNotEnoughResources() throws Exception {
@Test
public void testNotifyNewResourcesAvailable() throws Exception {
try (MockContext ctx = new MockContext()) {
- ctx.setHasEnoughResources(() -> false); // initially, not enough resources
+ ctx.setHasDesiredResources(() -> false); // initially, not enough resources
WaitingForResources wfr =
- new WaitingForResources(ctx, log, RESOURCE_COUNTER, Duration.ZERO);
- ctx.setHasEnoughResources(() -> true); // make resources available
+ new WaitingForResources(
+ ctx, log, RESOURCE_COUNTER, Duration.ZERO, STABILIZATION_TIMEOUT);
+ ctx.setHasDesiredResources(() -> true); // make resources available
ctx.setExpectExecuting(assertNonNull());
wfr.notifyNewResourcesAvailable(); // .. and notify
}
}
@Test
- public void testResourceTimeout() throws Exception {
+ public void testSchedulingWithSufficientResourcesAndNoStabilizationTimeout() throws Exception {
+ try (MockContext ctx = new MockContext()) {
+
+ Duration noStabilizationTimeout = Duration.ofMillis(0);
+ WaitingForResources wfr =
+ new WaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ Duration.ofSeconds(1000),
+ noStabilizationTimeout);
+
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> true);
+ ctx.setExpectExecuting(assertNonNull());
+ wfr.notifyNewResourcesAvailable();
+ }
+ }
+
+ @Test
+ public void testNoSchedulingIfStabilizationTimeoutIsConfigured() throws Exception {
try (MockContext ctx = new MockContext()) {
- ctx.setHasEnoughResources(() -> false);
+
+ Duration stabilizationTimeout = Duration.ofMillis(50000);
+
WaitingForResources wfr =
- new WaitingForResources(ctx, log, RESOURCE_COUNTER, Duration.ZERO);
+ new WaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ Duration.ofSeconds(1000),
+ stabilizationTimeout);
+
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> true);
+ wfr.notifyNewResourcesAvailable();
+ // we are not triggering the scheduled tasks, to simulate a long stabilization timeout
+
+ assertThat(ctx.hasStateTransition(), is(false));
+ }
+ }
+
+ @Test
+ public void testSchedulingIfStabilizationTimeoutIsConfigured() throws Exception {
+ try (MockContext ctx = new MockContext()) {
+
+ Duration initialResourceTimeout = Duration.ofMillis(120948);
+ Duration stabilizationTimeout = Duration.ofMillis(50000);
+
+ TestingWaitingForResources wfr =
+ new TestingWaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ initialResourceTimeout,
+ stabilizationTimeout);
+
+ // not enough resources available
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> false);
+
+ assertNoStateTransitionsAfterExecutingRunnables(ctx, wfr);
+
+ // sufficient resources available
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> true);
+ wfr.setTestDeadline(Deadline.fromNow(Duration.ofDays(600)));
Review comment:
This leaks quite some implementation details. If we need to go to these lengths then this is usually an indicator that the class is not well testable.
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java
##########
@@ -110,41 +117,187 @@ public void testNotEnoughResources() throws Exception {
@Test
public void testNotifyNewResourcesAvailable() throws Exception {
try (MockContext ctx = new MockContext()) {
- ctx.setHasEnoughResources(() -> false); // initially, not enough resources
+ ctx.setHasDesiredResources(() -> false); // initially, not enough resources
WaitingForResources wfr =
- new WaitingForResources(ctx, log, RESOURCE_COUNTER, Duration.ZERO);
- ctx.setHasEnoughResources(() -> true); // make resources available
+ new WaitingForResources(
+ ctx, log, RESOURCE_COUNTER, Duration.ZERO, STABILIZATION_TIMEOUT);
+ ctx.setHasDesiredResources(() -> true); // make resources available
ctx.setExpectExecuting(assertNonNull());
wfr.notifyNewResourcesAvailable(); // .. and notify
}
}
@Test
- public void testResourceTimeout() throws Exception {
+ public void testSchedulingWithSufficientResourcesAndNoStabilizationTimeout() throws Exception {
+ try (MockContext ctx = new MockContext()) {
+
+ Duration noStabilizationTimeout = Duration.ofMillis(0);
+ WaitingForResources wfr =
+ new WaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ Duration.ofSeconds(1000),
+ noStabilizationTimeout);
+
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> true);
+ ctx.setExpectExecuting(assertNonNull());
+ wfr.notifyNewResourcesAvailable();
+ }
+ }
+
+ @Test
+ public void testNoSchedulingIfStabilizationTimeoutIsConfigured() throws Exception {
try (MockContext ctx = new MockContext()) {
- ctx.setHasEnoughResources(() -> false);
+
+ Duration stabilizationTimeout = Duration.ofMillis(50000);
+
WaitingForResources wfr =
- new WaitingForResources(ctx, log, RESOURCE_COUNTER, Duration.ZERO);
+ new WaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ Duration.ofSeconds(1000),
+ stabilizationTimeout);
+
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> true);
+ wfr.notifyNewResourcesAvailable();
+ // we are not triggering the scheduled tasks, to simulate a long stabilization timeout
+
+ assertThat(ctx.hasStateTransition(), is(false));
+ }
+ }
+
+ @Test
+ public void testSchedulingIfStabilizationTimeoutIsConfigured() throws Exception {
+ try (MockContext ctx = new MockContext()) {
+
+ Duration initialResourceTimeout = Duration.ofMillis(120948);
+ Duration stabilizationTimeout = Duration.ofMillis(50000);
+
+ TestingWaitingForResources wfr =
+ new TestingWaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ initialResourceTimeout,
+ stabilizationTimeout);
+
+ // not enough resources available
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> false);
+
+ assertNoStateTransitionsAfterExecutingRunnables(ctx, wfr);
+
+ // sufficient resources available
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> true);
+ wfr.setTestDeadline(Deadline.fromNow(Duration.ofDays(600)));
ctx.setExpectExecuting(assertNonNull());
+ // start countdown
+ wfr.notifyNewResourcesAvailable();
+
+ assertNoStateTransitionsAfterExecutingRunnables(ctx, wfr);
+
+ // advance time to now
+ wfr.setTestDeadline(Deadline.now());
Review comment:
For these timeout tests, it might make sense to introduce a `Clock` for the global test time. This `Clock` could be used by the state as well as the test context. Then we could say that we can advance the clock which would automatically run all runnables which are eligible now. That way we might be able to express the tests a bit more naturally.
##########
File path: docs/layouts/shortcodes/generated/all_jobmanager_section.html
##########
@@ -14,11 +14,17 @@
<td>Integer</td>
<td>Configure the minimum increase in parallelism for a job to scale up.</td>
</tr>
+ <tr>
+ <td><h5>jobmanager.adaptive-scheduler.resource-stabilization-timeout</h5></td>
+ <td style="word-wrap: break-word;">10 s</td>
+ <td>Duration</td>
+ <td>The resource stabilization timeout defines the time the JobManager will wait if fewer than the required resources are available, but sufficient resources for execution are there. Once this timeout has passed, the job will start executing with the available resources, or fail, if the resources are not sufficient. The timeout starts as soon as there are sufficient resources available for execution.<br />If <span markdown="span">`scheduler-mode`</span> is configured to <span markdown="span">`REACTIVE`</span>, this configuration value will default to 0, so that jobs are starting immediately with the available resources.</td>
Review comment:
Why are we failing if there are not sufficient resources available? I thought that this timeout would only be triggered if we have sufficient resources?
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java
##########
@@ -110,41 +117,187 @@ public void testNotEnoughResources() throws Exception {
@Test
public void testNotifyNewResourcesAvailable() throws Exception {
try (MockContext ctx = new MockContext()) {
- ctx.setHasEnoughResources(() -> false); // initially, not enough resources
+ ctx.setHasDesiredResources(() -> false); // initially, not enough resources
WaitingForResources wfr =
- new WaitingForResources(ctx, log, RESOURCE_COUNTER, Duration.ZERO);
- ctx.setHasEnoughResources(() -> true); // make resources available
+ new WaitingForResources(
+ ctx, log, RESOURCE_COUNTER, Duration.ZERO, STABILIZATION_TIMEOUT);
+ ctx.setHasDesiredResources(() -> true); // make resources available
ctx.setExpectExecuting(assertNonNull());
wfr.notifyNewResourcesAvailable(); // .. and notify
}
}
@Test
- public void testResourceTimeout() throws Exception {
+ public void testSchedulingWithSufficientResourcesAndNoStabilizationTimeout() throws Exception {
+ try (MockContext ctx = new MockContext()) {
+
+ Duration noStabilizationTimeout = Duration.ofMillis(0);
+ WaitingForResources wfr =
+ new WaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ Duration.ofSeconds(1000),
+ noStabilizationTimeout);
+
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> true);
+ ctx.setExpectExecuting(assertNonNull());
+ wfr.notifyNewResourcesAvailable();
+ }
+ }
+
+ @Test
+ public void testNoSchedulingIfStabilizationTimeoutIsConfigured() throws Exception {
try (MockContext ctx = new MockContext()) {
- ctx.setHasEnoughResources(() -> false);
+
+ Duration stabilizationTimeout = Duration.ofMillis(50000);
+
WaitingForResources wfr =
- new WaitingForResources(ctx, log, RESOURCE_COUNTER, Duration.ZERO);
+ new WaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ Duration.ofSeconds(1000),
+ stabilizationTimeout);
+
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> true);
+ wfr.notifyNewResourcesAvailable();
+ // we are not triggering the scheduled tasks, to simulate a long stabilization timeout
+
+ assertThat(ctx.hasStateTransition(), is(false));
+ }
+ }
+
+ @Test
+ public void testSchedulingIfStabilizationTimeoutIsConfigured() throws Exception {
+ try (MockContext ctx = new MockContext()) {
+
+ Duration initialResourceTimeout = Duration.ofMillis(120948);
+ Duration stabilizationTimeout = Duration.ofMillis(50000);
+
+ TestingWaitingForResources wfr =
+ new TestingWaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ initialResourceTimeout,
+ stabilizationTimeout);
+
+ // not enough resources available
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> false);
+
+ assertNoStateTransitionsAfterExecutingRunnables(ctx, wfr);
+
+ // sufficient resources available
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> true);
+ wfr.setTestDeadline(Deadline.fromNow(Duration.ofDays(600)));
ctx.setExpectExecuting(assertNonNull());
+ // start countdown
+ wfr.notifyNewResourcesAvailable();
+
+ assertNoStateTransitionsAfterExecutingRunnables(ctx, wfr);
+
+ // advance time to now
+ wfr.setTestDeadline(Deadline.now());
- // immediately execute all scheduled runnables
- assertThat(ctx.getScheduledRunnables().size(), greaterThan(0));
+ // execute all runnables and trigger expected state transition
for (ScheduledRunnable scheduledRunnable : ctx.getScheduledRunnables()) {
- if (scheduledRunnable.getExpectedState() == wfr) {
+ if (scheduledRunnable.getExpectedState() == wfr
+ && !scheduledRunnable.getDelay().equals(initialResourceTimeout)
+ && !scheduledRunnable.getDelay().equals(Duration.ZERO)) {
Review comment:
Why do we need these filters here?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-796989540
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 ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f (Thu Mar 11 19:31:44 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## 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](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
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 <summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
rmetzger commented on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797272559
Reported CI failure: https://issues.apache.org/jira/browse/FLINK-21728
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on a change in pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #15159:
URL: https://github.com/apache/flink/pull/15159#discussion_r598596260
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java
##########
@@ -178,12 +344,24 @@ void setExpectCreatingExecutionGraph() {
creatingExecutionGraphStateValidator.expectInput(none -> {});
}
- void runScheduledTasks() {
- for (ScheduledTask<Void> scheduledTask : scheduledTasks) {
- scheduledTask.execute();
+ void runScheduledTasks(long untilDelay) {
+ Iterable<ScheduledTask<Void>> copyOfScheduledTasks = new ArrayList<>(scheduledTasks);
+ Iterator<ScheduledTask<Void>> scheduledTaskIterator = copyOfScheduledTasks.iterator();
+ while (scheduledTaskIterator.hasNext()) {
+ ScheduledTask<Void> scheduledTask = scheduledTaskIterator.next();
+ if (scheduledTask.getDelay(TimeUnit.MILLISECONDS) <= untilDelay) {
+ scheduledTask.execute();
+ if (!scheduledTask.isPeriodic()) {
+ scheduledTaskIterator.remove();
+ }
+ }
Review comment:
I think this implementation has the problem that we are forgetting to execute tasks which are created as part of other tasks because we copy `scheduledTasks`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
}, {
"hash" : "30724c1ba85f417b86d00337257b2180f03d376a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792",
"triggerID" : "30724c1ba85f417b86d00337257b2180f03d376a",
"triggerType" : "PUSH"
}, {
"hash" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818",
"triggerID" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"triggerType" : "PUSH"
}, {
"hash" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15104",
"triggerID" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"triggerType" : "PUSH"
}, {
"hash" : "989abd48b62ba85081f9b4cb4492dbff97fb7639",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15147",
"triggerID" : "989abd48b62ba85081f9b4cb4492dbff97fb7639",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 28bf753c82199c68f31a19234bc85b7fc03dc9cb Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15104)
* 989abd48b62ba85081f9b4cb4492dbff97fb7639 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15147)
<details>
<summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* aab6b14a76ca84b1fab62d698c2c5914a8f37807 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499)
* 001d7c8625cf8ac8574fab93ee122d73d2722cd7 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732)
* aab6b14a76ca84b1fab62d698c2c5914a8f37807 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #15159:
URL: https://github.com/apache/flink/pull/15159#discussion_r598462236
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java
##########
@@ -90,13 +107,40 @@ public Logger getLogger() {
@Override
public void notifyNewResourcesAvailable() {
- if (context.hasEnoughResources(desiredResources)) {
+ checkDesiredOrSufficientResourcesAvailable();
+ }
+
+ private void checkDesiredOrSufficientResourcesAvailable() {
+ if (context.hasDesiredResources(desiredResources)) {
createExecutionGraphWithAvailableResources();
+ return;
}
+
+ if (context.hasSufficientResources()) {
Review comment:
If its okay for you, for the sake of time, I would address this later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
}, {
"hash" : "30724c1ba85f417b86d00337257b2180f03d376a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792",
"triggerID" : "30724c1ba85f417b86d00337257b2180f03d376a",
"triggerType" : "PUSH"
}, {
"hash" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818",
"triggerID" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"triggerType" : "PUSH"
}, {
"hash" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 57d9b95fac95a649bd8295c294b77aeee5df12fc Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818)
* 28bf753c82199c68f31a19234bc85b7fc03dc9cb UNKNOWN
<details>
<summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499)
* 001d7c8625cf8ac8574fab93ee122d73d2722cd7 UNKNOWN
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
}, {
"hash" : "30724c1ba85f417b86d00337257b2180f03d376a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792",
"triggerID" : "30724c1ba85f417b86d00337257b2180f03d376a",
"triggerType" : "PUSH"
}, {
"hash" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818",
"triggerID" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"triggerType" : "PUSH"
}, {
"hash" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15104",
"triggerID" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"triggerType" : "PUSH"
}, {
"hash" : "989abd48b62ba85081f9b4cb4492dbff97fb7639",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15147",
"triggerID" : "989abd48b62ba85081f9b4cb4492dbff97fb7639",
"triggerType" : "PUSH"
}, {
"hash" : "6651b8a6ba48a81572de8d22e43069a353cabf2e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6651b8a6ba48a81572de8d22e43069a353cabf2e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 28bf753c82199c68f31a19234bc85b7fc03dc9cb Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15104)
* 989abd48b62ba85081f9b4cb4492dbff97fb7639 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15147)
* 6651b8a6ba48a81572de8d22e43069a353cabf2e UNKNOWN
<details>
<summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on a change in pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #15159:
URL: https://github.com/apache/flink/pull/15159#discussion_r598586104
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java
##########
@@ -90,13 +107,40 @@ public Logger getLogger() {
@Override
public void notifyNewResourcesAvailable() {
- if (context.hasEnoughResources(desiredResources)) {
+ checkDesiredOrSufficientResourcesAvailable();
+ }
+
+ private void checkDesiredOrSufficientResourcesAvailable() {
+ if (context.hasDesiredResources(desiredResources)) {
createExecutionGraphWithAvailableResources();
+ return;
}
+
+ if (context.hasSufficientResources()) {
Review comment:
Alright. Sounds good.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
}, {
"hash" : "30724c1ba85f417b86d00337257b2180f03d376a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792",
"triggerID" : "30724c1ba85f417b86d00337257b2180f03d376a",
"triggerType" : "PUSH"
}, {
"hash" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818",
"triggerID" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"triggerType" : "PUSH"
}, {
"hash" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15104",
"triggerID" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"triggerType" : "PUSH"
}, {
"hash" : "989abd48b62ba85081f9b4cb4492dbff97fb7639",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15147",
"triggerID" : "989abd48b62ba85081f9b4cb4492dbff97fb7639",
"triggerType" : "PUSH"
}, {
"hash" : "6651b8a6ba48a81572de8d22e43069a353cabf2e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15154",
"triggerID" : "6651b8a6ba48a81572de8d22e43069a353cabf2e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 989abd48b62ba85081f9b4cb4492dbff97fb7639 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15147)
* 6651b8a6ba48a81572de8d22e43069a353cabf2e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15154)
<details>
<summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
}, {
"hash" : "30724c1ba85f417b86d00337257b2180f03d376a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792",
"triggerID" : "30724c1ba85f417b86d00337257b2180f03d376a",
"triggerType" : "PUSH"
}, {
"hash" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818",
"triggerID" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"triggerType" : "PUSH"
}, {
"hash" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15104",
"triggerID" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"triggerType" : "PUSH"
}, {
"hash" : "989abd48b62ba85081f9b4cb4492dbff97fb7639",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "989abd48b62ba85081f9b4cb4492dbff97fb7639",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 28bf753c82199c68f31a19234bc85b7fc03dc9cb Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15104)
* 989abd48b62ba85081f9b4cb4492dbff97fb7639 UNKNOWN
<details>
<summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tillrohrmann commented on a change in pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #15159:
URL: https://github.com/apache/flink/pull/15159#discussion_r593157108
##########
File path: flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
##########
@@ -411,6 +411,25 @@
.text(
"Increasing this value will make the cluster more resilient against temporary resources shortages (e.g., there is more time for a failed TaskManager to be restarted), "
+ "while decreasing this value reduces downtime of a job (provided that enough slots are available to still run the job).")
+ .linebreak()
+ .text(
+ "Setting a negative duration will disable the resource timeout: The JobManager will wait indefinitely for resources to appear.")
+ .build());
+
+ @Documentation.Section({
+ Documentation.Sections.EXPERT_SCHEDULING,
+ Documentation.Sections.ALL_JOB_MANAGER
+ })
+ public static final ConfigOption<Duration> RESOURCE_STABILIZATION_TIMEOUT =
Review comment:
I think we need to document the different default values when using the reactive mode. Also for `RESOURCE_WAIT_TIMEOUT`.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/ReactiveModeUtils.java
##########
@@ -47,5 +51,20 @@ public static void configureJobGraphForReactiveMode(JobGraph jobGraph) {
}
}
+ public static void configureClusterForReactiveMode(Configuration configuration) {
+ LOG.info("Modifying Cluster configuration for reactive mode");
+
+ if (!configuration.contains(JobManagerOptions.RESOURCE_STABILIZATION_TIMEOUT)) {
+ // Configure adaptive scheduler to schedule job even if desired resources are not
+ // available (but sufficient resources)
+ configuration.set(
+ JobManagerOptions.RESOURCE_STABILIZATION_TIMEOUT, Duration.ofMillis(0));
+ }
+ if (!configuration.contains(JobManagerOptions.RESOURCE_WAIT_TIMEOUT)) {
+ // configure adaptive scheduler to wait forever for TaskManagers to register
+ configuration.set(JobManagerOptions.RESOURCE_WAIT_TIMEOUT, Duration.ofMillis(-1));
+ }
+ }
Review comment:
I am not saying that we have to change it but I wanted to make an observation about this kind of pattern (changing the `Configuration` so that the right configuration values are being picked up later):
The problem I see with this pattern is that it decouples the instantiation of the required components and their configuration. Consequently, we as the developers have to remember that at every entrypoint which might use the `AdaptiveScheduler` we have to call `ReactiveModeUtils.configureclusterForReactiveMode` (here we have the `ClusterEntrypoint` and the `MiniCluster`). If we forget it, then the reactive mode might not really work.
If these context sensitive configuration option overwrites would be moved closer to the instantiation of the components, then it is easier to reuse the "reactive mode" in another setting. E.g. when seeing in the `DefaultJobManagerRunnerFactory` that we are running a job in reactive mode, then one could read `RESOURCE_STABILIZATION_TIMEOUT` and `RESOURCE_WAIT_TIMEOUT` with different default values (`0` and `-1`) which are then passed to the factory of the `AdaptiveScheduler`.
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java
##########
@@ -110,41 +115,164 @@ public void testNotEnoughResources() throws Exception {
@Test
public void testNotifyNewResourcesAvailable() throws Exception {
try (MockContext ctx = new MockContext()) {
- ctx.setHasEnoughResources(() -> false); // initially, not enough resources
+ ctx.setHasDesiredResources(() -> false); // initially, not enough resources
WaitingForResources wfr =
- new WaitingForResources(ctx, log, RESOURCE_COUNTER, Duration.ZERO);
- ctx.setHasEnoughResources(() -> true); // make resources available
+ new WaitingForResources(
+ ctx, log, RESOURCE_COUNTER, Duration.ZERO, STABILIZATION_TIMEOUT);
+ ctx.setHasDesiredResources(() -> true); // make resources available
ctx.setExpectExecuting(assertNonNull());
wfr.notifyNewResourcesAvailable(); // .. and notify
}
}
@Test
- public void testResourceTimeout() throws Exception {
+ public void testSchedulingWithSufficientResourcesAndNoStabilizationTimeout() throws Exception {
+ try (MockContext ctx = new MockContext()) {
+
+ Duration noStabilizationTimeout = Duration.ofMillis(0);
+ WaitingForResources wfr =
+ new WaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ Duration.ofSeconds(1000),
+ noStabilizationTimeout);
+
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> true);
+ ctx.setExpectExecuting(assertNonNull());
+ wfr.notifyNewResourcesAvailable();
+ }
+ }
+
+ @Test
+ public void testNoSchedulingIfStabilizationTimeoutIsConfigured() throws Exception {
try (MockContext ctx = new MockContext()) {
- ctx.setHasEnoughResources(() -> false);
+
+ Duration stabilizationTimeout = Duration.ofMillis(50000);
+
WaitingForResources wfr =
- new WaitingForResources(ctx, log, RESOURCE_COUNTER, Duration.ZERO);
+ new WaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ Duration.ofSeconds(1000),
+ stabilizationTimeout);
+
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> true);
+ wfr.notifyNewResourcesAvailable();
+ // we are not triggering the scheduled tasks, to simulate a long stabilization timeout
+
+ assertThat(ctx.hasStateTransition(), is(false));
+ }
+ }
+
+ @Test
+ public void testSchedulingIfStabilizationTimeoutIsConfigured() throws Exception {
+ try (MockContext ctx = new MockContext()) {
+
+ Duration initialResourceTimeout = Duration.ofMillis(120948);
+ Duration stabilizationTimeout = Duration.ofMillis(50000);
+
+ TestingWaitingForResources wfr =
+ new TestingWaitingForResources(
+ ctx,
+ log,
+ RESOURCE_COUNTER,
+ initialResourceTimeout,
+ stabilizationTimeout);
+
+ ctx.setHasDesiredResources(() -> false);
+ ctx.setHasSufficientResources(() -> true);
+ wfr.setStabilizationTimeoutOver(false);
ctx.setExpectExecuting(assertNonNull());
+ wfr.notifyNewResourcesAvailable();
+
+ wfr.setStabilizationTimeoutOver(true);
- // immediately execute all scheduled runnables
- assertThat(ctx.getScheduledRunnables().size(), greaterThan(0));
+ // we are executing all scheduled runnables, except those with the
+ // initialResourceTimeout or a ZERO timeout to make sure we are testing the
+ // stabilization timeout
for (ScheduledRunnable scheduledRunnable : ctx.getScheduledRunnables()) {
- if (scheduledRunnable.getExpectedState() == wfr) {
+ if (scheduledRunnable.getExpectedState() == wfr
+ && !scheduledRunnable.getDelay().equals(initialResourceTimeout)
+ && !scheduledRunnable.getDelay().equals(Duration.ZERO)) {
scheduledRunnable.runAction();
Review comment:
This looks a bit hacky. Would it make sense to add something like advance time which runs all the runnables which are eligible for running? Right now we effectively test a scenario which is not possible because we leave the 0 delay runnables out.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java
##########
@@ -90,13 +102,35 @@ public Logger getLogger() {
@Override
public void notifyNewResourcesAvailable() {
- if (context.hasEnoughResources(desiredResources)) {
+ checkDesiredOrSufficientResourcesAvailable();
+ }
+
+ private void checkDesiredOrSufficientResourcesAvailable() {
+ if (context.hasDesiredResources(desiredResources)) {
createExecutionGraphWithAvailableResources();
+ return;
+ }
+
+ if (context.hasSufficientResources()) {
+ if (isResourceStabilizationDeadlineOverdue()) {
+ createExecutionGraphWithAvailableResources();
+ } else {
+ // schedule next resource check
+ context.runIfState(
+ this,
+ this::checkDesiredOrSufficientResourcesAvailable,
+ resourceStabilizationDeadline.timeLeft());
+ }
}
}
Review comment:
I think the implementation of the resource stabilization timeout differs from the description of the option. According to the description of the option I would have expected that the timeout only starts after we have acquired enough resources to run the job. With the current implementation we already start the stabilization timeout when we enter the state.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/ReactiveModeUtils.java
##########
@@ -47,5 +51,20 @@ public static void configureJobGraphForReactiveMode(JobGraph jobGraph) {
}
}
+ public static void configureClusterForReactiveMode(Configuration configuration) {
+ LOG.info("Modifying Cluster configuration for reactive mode");
+
+ if (!configuration.contains(JobManagerOptions.RESOURCE_STABILIZATION_TIMEOUT)) {
+ // Configure adaptive scheduler to schedule job even if desired resources are not
+ // available (but sufficient resources)
+ configuration.set(
+ JobManagerOptions.RESOURCE_STABILIZATION_TIMEOUT, Duration.ofMillis(0));
+ }
+ if (!configuration.contains(JobManagerOptions.RESOURCE_WAIT_TIMEOUT)) {
+ // configure adaptive scheduler to wait forever for TaskManagers to register
+ configuration.set(JobManagerOptions.RESOURCE_WAIT_TIMEOUT, Duration.ofMillis(-1));
+ }
+ }
Review comment:
I think the key observation is that if we keep related things closer together, then it is easier to reuse them and it is easier to maintain them because they are not spread all over the place.
##########
File path: flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
##########
@@ -411,6 +411,25 @@
.text(
"Increasing this value will make the cluster more resilient against temporary resources shortages (e.g., there is more time for a failed TaskManager to be restarted), "
+ "while decreasing this value reduces downtime of a job (provided that enough slots are available to still run the job).")
+ .linebreak()
+ .text(
+ "Setting a negative duration will disable the resource timeout: The JobManager will wait indefinitely for resources to appear.")
+ .build());
+
+ @Documentation.Section({
+ Documentation.Sections.EXPERT_SCHEDULING,
+ Documentation.Sections.ALL_JOB_MANAGER
+ })
+ public static final ConfigOption<Duration> RESOURCE_STABILIZATION_TIMEOUT =
+ key("jobmanager.adaptive-scheduler.resource-stabilization-timeout")
+ .durationType()
+ .defaultValue(RESOURCE_WAIT_TIMEOUT.defaultValue())
+ .withDescription(
+ Description.builder()
+ .text(
+ "The resource stabilization timeout defines the time the JobManager will wait "
+ + "if fewer than the required resources are available, but sufficient resources for execution are there."
Review comment:
```suggestion
+ "if fewer than the required resources are available, but sufficient resources for execution are there. "
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
}, {
"hash" : "30724c1ba85f417b86d00337257b2180f03d376a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792",
"triggerID" : "30724c1ba85f417b86d00337257b2180f03d376a",
"triggerType" : "PUSH"
}, {
"hash" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818",
"triggerID" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* aab6b14a76ca84b1fab62d698c2c5914a8f37807 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735)
* 30724c1ba85f417b86d00337257b2180f03d376a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792)
* 57d9b95fac95a649bd8295c294b77aeee5df12fc Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
}, {
"hash" : "30724c1ba85f417b86d00337257b2180f03d376a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792",
"triggerID" : "30724c1ba85f417b86d00337257b2180f03d376a",
"triggerType" : "PUSH"
}, {
"hash" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* aab6b14a76ca84b1fab62d698c2c5914a8f37807 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735)
* 30724c1ba85f417b86d00337257b2180f03d376a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792)
* 57d9b95fac95a649bd8295c294b77aeee5df12fc UNKNOWN
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
}, {
"hash" : "30724c1ba85f417b86d00337257b2180f03d376a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792",
"triggerID" : "30724c1ba85f417b86d00337257b2180f03d376a",
"triggerType" : "PUSH"
}, {
"hash" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818",
"triggerID" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"triggerType" : "PUSH"
}, {
"hash" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15104",
"triggerID" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"triggerType" : "PUSH"
}, {
"hash" : "989abd48b62ba85081f9b4cb4492dbff97fb7639",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15147",
"triggerID" : "989abd48b62ba85081f9b4cb4492dbff97fb7639",
"triggerType" : "PUSH"
}, {
"hash" : "6651b8a6ba48a81572de8d22e43069a353cabf2e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15154",
"triggerID" : "6651b8a6ba48a81572de8d22e43069a353cabf2e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 28bf753c82199c68f31a19234bc85b7fc03dc9cb Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15104)
* 989abd48b62ba85081f9b4cb4492dbff97fb7639 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15147)
* 6651b8a6ba48a81572de8d22e43069a353cabf2e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15154)
<details>
<summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
}, {
"hash" : "30724c1ba85f417b86d00337257b2180f03d376a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792",
"triggerID" : "30724c1ba85f417b86d00337257b2180f03d376a",
"triggerType" : "PUSH"
}, {
"hash" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818",
"triggerID" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 57d9b95fac95a649bd8295c294b77aeee5df12fc Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #15159:
URL: https://github.com/apache/flink/pull/15159#discussion_r597583173
##########
File path: docs/layouts/shortcodes/generated/all_jobmanager_section.html
##########
@@ -14,11 +14,17 @@
<td>Integer</td>
<td>Configure the minimum increase in parallelism for a job to scale up.</td>
</tr>
+ <tr>
+ <td><h5>jobmanager.adaptive-scheduler.resource-stabilization-timeout</h5></td>
+ <td style="word-wrap: break-word;">10 s</td>
+ <td>Duration</td>
+ <td>The resource stabilization timeout defines the time the JobManager will wait if fewer than the required resources are available, but sufficient resources for execution are there. Once this timeout has passed, the job will start executing with the available resources, or fail, if the resources are not sufficient. The timeout starts as soon as there are sufficient resources available for execution.<br />If <span markdown="span">`scheduler-mode`</span> is configured to <span markdown="span">`REACTIVE`</span>, this configuration value will default to 0, so that jobs are starting immediately with the available resources.</td>
Review comment:
I thought that we can not guarantee sufficient resources, even if SlotAllocator.determineParallelism says so, because we could loose slots between "sufficient resources" and "start scheduling".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15159:
URL: https://github.com/apache/flink/pull/15159#issuecomment-797005525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14499",
"triggerID" : "ddeb93a7a3b5de3bb3d3a19bea0936f9e2296c9f",
"triggerType" : "PUSH"
}, {
"hash" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14732",
"triggerID" : "001d7c8625cf8ac8574fab93ee122d73d2722cd7",
"triggerType" : "PUSH"
}, {
"hash" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14735",
"triggerID" : "aab6b14a76ca84b1fab62d698c2c5914a8f37807",
"triggerType" : "PUSH"
}, {
"hash" : "30724c1ba85f417b86d00337257b2180f03d376a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14792",
"triggerID" : "30724c1ba85f417b86d00337257b2180f03d376a",
"triggerType" : "PUSH"
}, {
"hash" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14818",
"triggerID" : "57d9b95fac95a649bd8295c294b77aeee5df12fc",
"triggerType" : "PUSH"
}, {
"hash" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15104",
"triggerID" : "28bf753c82199c68f31a19234bc85b7fc03dc9cb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 28bf753c82199c68f31a19234bc85b7fc03dc9cb Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15104)
<details>
<summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #15159:
URL: https://github.com/apache/flink/pull/15159#discussion_r597649121
##########
File path: docs/layouts/shortcodes/generated/all_jobmanager_section.html
##########
@@ -14,11 +14,17 @@
<td>Integer</td>
<td>Configure the minimum increase in parallelism for a job to scale up.</td>
</tr>
+ <tr>
+ <td><h5>jobmanager.adaptive-scheduler.resource-stabilization-timeout</h5></td>
+ <td style="word-wrap: break-word;">10 s</td>
+ <td>Duration</td>
+ <td>The resource stabilization timeout defines the time the JobManager will wait if fewer than the required resources are available, but sufficient resources for execution are there. Once this timeout has passed, the job will start executing with the available resources, or fail, if the resources are not sufficient. The timeout starts as soon as there are sufficient resources available for execution.<br />If <span markdown="span">`scheduler-mode`</span> is configured to <span markdown="span">`REACTIVE`</span>, this configuration value will default to 0, so that jobs are starting immediately with the available resources.</td>
Review comment:
Thanks for the clarification.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #15159: [FLINK-21136] Adjust timeouts for reactive mode
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #15159:
URL: https://github.com/apache/flink/pull/15159#discussion_r594806790
##########
File path: flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
##########
@@ -411,6 +411,35 @@
.text(
"Increasing this value will make the cluster more resilient against temporary resources shortages (e.g., there is more time for a failed TaskManager to be restarted), "
+ "while decreasing this value reduces downtime of a job (provided that enough slots are available to still run the job).")
+ .linebreak()
+ .text(
+ "Setting a negative duration will disable the resource timeout: The JobManager will wait indefinitely for resources to appear.")
+ .linebreak()
+ .text(
+ "If %s is configured to %s, this configuration value will default to a negative value to disable the resource timeout.",
+ code(SCHEDULER_MODE.key()),
+ code(SchedulerExecutionMode.REACTIVE.name()))
+ .build());
+
+ @Documentation.Section({
+ Documentation.Sections.EXPERT_SCHEDULING,
+ Documentation.Sections.ALL_JOB_MANAGER
+ })
+ public static final ConfigOption<Duration> RESOURCE_STABILIZATION_TIMEOUT =
+ key("jobmanager.adaptive-scheduler.resource-stabilization-timeout")
+ .durationType()
+ .defaultValue(RESOURCE_WAIT_TIMEOUT.defaultValue())
+ .withDescription(
+ Description.builder()
+ .text(
+ "The resource stabilization timeout defines the time the JobManager will wait "
Review comment:
I found it quite difficult to grasp what exactly this option does, and would suggest a re-phrasing:
`The time the JobManager will wait for to acquire further resources, once the minimum set of resources required for the execution of a job has been acquired. Once [...]`
Maybe also expand the documentation along the lines of (a lower value results in the job starting up faster, while higher value ...)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org