You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by xt...@apache.org on 2023/02/02 09:12:00 UTC
[flink] 05/05: [FLINK-30674] Enabling consume partial finished partition by default for speculative execution in hybrid shuffle mode
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7bc18ef7c3065b8c2210f66b2329dc6bdf49ff56
Author: Weijie Guo <re...@163.com>
AuthorDate: Fri Jan 13 17:05:44 2023 +0800
[FLINK-30674] Enabling consume partial finished partition by default for speculative execution in hybrid shuffle mode
This closes #21804
---
.../AdaptiveBatchSchedulerFactory.java | 22 +++++++++++++++++-----
1 file changed, 17 insertions(+), 5 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerFactory.java
index 7e69a0762b8..73712faeacc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerFactory.java
@@ -71,6 +71,7 @@ import org.apache.flink.util.Preconditions;
import org.apache.flink.util.concurrent.ScheduledExecutorServiceAdapter;
import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
@@ -78,7 +79,6 @@ import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.function.Consumer;
-import static org.apache.flink.configuration.JobManagerOptions.HybridPartitionDataConsumeConstraint.ALL_PRODUCERS_FINISHED;
import static org.apache.flink.configuration.JobManagerOptions.HybridPartitionDataConsumeConstraint.ONLY_FINISHED_PRODUCERS;
import static org.apache.flink.configuration.JobManagerOptions.HybridPartitionDataConsumeConstraint.UNFINISHED_PRODUCERS;
import static org.apache.flink.util.Preconditions.checkState;
@@ -86,6 +86,8 @@ import static org.apache.flink.util.Preconditions.checkState;
/** Factory for {@link AdaptiveBatchScheduler}. */
public class AdaptiveBatchSchedulerFactory implements SchedulerNGFactory {
+ private static final Logger LOG = LoggerFactory.getLogger(AdaptiveBatchSchedulerFactory.class);
+
@Override
public SchedulerNG createInstance(
Logger log,
@@ -263,10 +265,20 @@ public class AdaptiveBatchSchedulerFactory implements SchedulerNGFactory {
final HybridPartitionDataConsumeConstraint hybridPartitionDataConsumeConstraint =
configuration
.getOptional(JobManagerOptions.HYBRID_PARTITION_DATA_CONSUME_CONSTRAINT)
- .orElse(
- enableSpeculativeExecution
- ? ALL_PRODUCERS_FINISHED
- : UNFINISHED_PRODUCERS);
+ .orElseGet(
+ () -> {
+ HybridPartitionDataConsumeConstraint defaultConstraint =
+ enableSpeculativeExecution
+ ? ONLY_FINISHED_PRODUCERS
+ : UNFINISHED_PRODUCERS;
+ LOG.info(
+ "Set {} to {} as it is not configured",
+ JobManagerOptions
+ .HYBRID_PARTITION_DATA_CONSUME_CONSTRAINT
+ .key(),
+ defaultConstraint.name());
+ return defaultConstraint;
+ });
if (enableSpeculativeExecution) {
Preconditions.checkState(
hybridPartitionDataConsumeConstraint != UNFINISHED_PRODUCERS,