[FLINK-28271][runtime] Add config options for speculative scheduler

This closes apache#20087.
tongcheng-elong · Jul 1, 2022 · 2aae497 · 2aae497
1 parent 67aeb81
commit 2aae497
Show file tree

Hide file tree

Showing 4 changed files with 88 additions and 0 deletions.
diff --git a/docs/layouts/shortcodes/generated/all_jobmanager_section.html b/docs/layouts/shortcodes/generated/all_jobmanager_section.html
@@ -32,6 +32,24 @@
  <td>Integer</td>
  <td>The lower bound of allowed parallelism to set adaptively if <code class="highlighter-rouge">jobmanager.scheduler</code> has been set to <code class="highlighter-rouge">AdaptiveBatch</code>. Currently, this option should be configured as a power of 2, otherwise it will also be rounded up to a power of 2 automatically.</td>
  </tr>
+ <tr>
+ <td><h5>jobmanager.adaptive-batch-scheduler.speculative.block-slow-node-duration</h5></td>
+ <td style="word-wrap: break-word;">1 min</td>
+ <td>Duration</td>
+ <td>Controls how long an detected slow node should be blocked for.</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.adaptive-batch-scheduler.speculative.enabled</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Controls whether to enable speculative execution.</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.adaptive-batch-scheduler.speculative.max-concurrent-executions</h5></td>
+ <td style="word-wrap: break-word;">2</td>
+ <td>Integer</td>
+ <td>Controls the maximum number of execution attempts of each operator that can execute concurrently, including the original one and speculative ones.</td>
+ </tr>
  <tr>
  <td><h5>jobmanager.adaptive-scheduler.min-parallelism-increase</h5></td>
  <td style="word-wrap: break-word;">1</td>

diff --git a/docs/layouts/shortcodes/generated/expert_scheduling_section.html b/docs/layouts/shortcodes/generated/expert_scheduling_section.html
@@ -50,6 +50,24 @@
  <td>Integer</td>
  <td>The lower bound of allowed parallelism to set adaptively if <code class="highlighter-rouge">jobmanager.scheduler</code> has been set to <code class="highlighter-rouge">AdaptiveBatch</code>. Currently, this option should be configured as a power of 2, otherwise it will also be rounded up to a power of 2 automatically.</td>
  </tr>
+ <tr>
+ <td><h5>jobmanager.adaptive-batch-scheduler.speculative.block-slow-node-duration</h5></td>
+ <td style="word-wrap: break-word;">1 min</td>
+ <td>Duration</td>
+ <td>Controls how long an detected slow node should be blocked for.</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.adaptive-batch-scheduler.speculative.enabled</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Controls whether to enable speculative execution.</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.adaptive-batch-scheduler.speculative.max-concurrent-executions</h5></td>
+ <td style="word-wrap: break-word;">2</td>
+ <td>Integer</td>
+ <td>Controls the maximum number of execution attempts of each operator that can execute concurrently, including the original one and speculative ones.</td>
+ </tr>
  <tr>
  <td><h5>jobmanager.adaptive-scheduler.min-parallelism-increase</h5></td>
  <td style="word-wrap: break-word;">1</td>

diff --git a/docs/layouts/shortcodes/generated/job_manager_configuration.html b/docs/layouts/shortcodes/generated/job_manager_configuration.html
@@ -32,6 +32,24 @@
  <td>Integer</td>
  <td>The lower bound of allowed parallelism to set adaptively if <code class="highlighter-rouge">jobmanager.scheduler</code> has been set to <code class="highlighter-rouge">AdaptiveBatch</code>. Currently, this option should be configured as a power of 2, otherwise it will also be rounded up to a power of 2 automatically.</td>
  </tr>
+ <tr>
+ <td><h5>jobmanager.adaptive-batch-scheduler.speculative.block-slow-node-duration</h5></td>
+ <td style="word-wrap: break-word;">1 min</td>
+ <td>Duration</td>
+ <td>Controls how long an detected slow node should be blocked for.</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.adaptive-batch-scheduler.speculative.enabled</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Controls whether to enable speculative execution.</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.adaptive-batch-scheduler.speculative.max-concurrent-executions</h5></td>
+ <td style="word-wrap: break-word;">2</td>
+ <td>Integer</td>
+ <td>Controls the maximum number of execution attempts of each operator that can execute concurrently, including the original one and speculative ones.</td>
+ </tr>
  <tr>
  <td><h5>jobmanager.adaptive-scheduler.min-parallelism-increase</h5></td>
  <td style="word-wrap: break-word;">1</td>

diff --git a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
@@ -593,6 +593,40 @@ public enum SchedulerType {
  code(SchedulerType.AdaptiveBatch.name()))
  .build());
 
+ @Documentation.Section({
+ Documentation.Sections.EXPERT_SCHEDULING,
+ Documentation.Sections.ALL_JOB_MANAGER
+ })
+ public static final ConfigOption<Boolean> SPECULATIVE_ENABLED =
+ key("jobmanager.adaptive-batch-scheduler.speculative.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Controls whether to enable speculative execution.");
+
+ @Documentation.Section({
+ Documentation.Sections.EXPERT_SCHEDULING,
+ Documentation.Sections.ALL_JOB_MANAGER
+ })
+ public static final ConfigOption<Integer> SPECULATIVE_MAX_CONCURRENT_EXECUTIONS =
+ key("jobmanager.adaptive-batch-scheduler.speculative.max-concurrent-executions")
+ .intType()
+ .defaultValue(2)
+ .withDescription(
+ "Controls the maximum number of execution attempts of each operator "
+ + "that can execute concurrently, including the original one "
+ + "and speculative ones.");
+
+ @Documentation.Section({
+ Documentation.Sections.EXPERT_SCHEDULING,
+ Documentation.Sections.ALL_JOB_MANAGER
+ })
+ public static final ConfigOption<Duration> BLOCK_SLOW_NODE_DURATION =
+ key("jobmanager.adaptive-batch-scheduler.speculative.block-slow-node-duration")
+ .durationType()
+ .defaultValue(Duration.ofMinutes(1))
+ .withDescription(
+ "Controls how long an detected slow node should be blocked for.");
+
  /**
  * The JobManager's ResourceID. If not configured, the ResourceID will be generated randomly.
  */