You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/02/20 10:17:10 UTC
[1/3] flink git commit: [FLINK-8698] [flip6] Use
Flip6LocalStreamEnvironment instead of LocalStreamEnvironment
Repository: flink
Updated Branches:
refs/heads/master 9da315234 -> d87531294
[FLINK-8698] [flip6] Use Flip6LocalStreamEnvironment instead of LocalStreamEnvironment
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5c18fa5b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5c18fa5b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5c18fa5b
Branch: refs/heads/master
Commit: 5c18fa5b7214a0be73d208046533477927a36532
Parents: 9da3152
Author: Till Rohrmann <tr...@apache.org>
Authored: Mon Feb 19 12:00:08 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Tue Feb 20 11:16:57 2018 +0100
----------------------------------------------------------------------
.../environment/Flip6LocalStreamEnvironment.java | 15 ++-------------
.../api/environment/LocalStreamEnvironment.java | 2 +-
.../environment/StreamExecutionEnvironment.java | 19 +++++++++++--------
3 files changed, 14 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5c18fa5b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/Flip6LocalStreamEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/Flip6LocalStreamEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/Flip6LocalStreamEnvironment.java
index 9d4f2a7..c5952fb 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/Flip6LocalStreamEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/Flip6LocalStreamEnvironment.java
@@ -18,9 +18,7 @@
package org.apache.flink.streaming.api.environment;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.JobExecutionResult;
-import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -40,13 +38,10 @@ import org.slf4j.LoggerFactory;
* parallelism can be set via {@link #setParallelism(int)}.
*/
@Internal
-public class Flip6LocalStreamEnvironment extends StreamExecutionEnvironment {
+public class Flip6LocalStreamEnvironment extends LocalStreamEnvironment {
private static final Logger LOG = LoggerFactory.getLogger(Flip6LocalStreamEnvironment.class);
- /** The configuration to use for the mini cluster. */
- private final Configuration conf;
-
/**
* Creates a new mini cluster stream environment that uses the default configuration.
*/
@@ -60,13 +55,7 @@ public class Flip6LocalStreamEnvironment extends StreamExecutionEnvironment {
* @param config The configuration used to configure the local executor.
*/
public Flip6LocalStreamEnvironment(Configuration config) {
- if (!ExecutionEnvironment.areExplicitEnvironmentsAllowed()) {
- throw new InvalidProgramException(
- "The Flip6LocalStreamEnvironment cannot be used when submitting a program through a client, " +
- "or running in a TestEnvironment context.");
- }
-
- this.conf = config == null ? new Configuration() : config;
+ super(config);
setParallelism(1);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5c18fa5b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java
index 6b31ff8..a53e2a3 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java
@@ -49,7 +49,7 @@ public class LocalStreamEnvironment extends StreamExecutionEnvironment {
private static final Logger LOG = LoggerFactory.getLogger(LocalStreamEnvironment.class);
/** The configuration to use for the local cluster. */
- private final Configuration conf;
+ protected final Configuration conf;
/**
* Creates a new local stream environment that uses the default configuration.
http://git-wip-us.apache.org/repos/asf/flink/blob/5c18fa5b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
index 56a7e29..c39201c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -47,6 +47,7 @@ import org.apache.flink.client.program.OptimizerPlanEnvironment;
import org.apache.flink.client.program.PreviewPlanEnvironment;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
@@ -1633,9 +1634,7 @@ public abstract class StreamExecutionEnvironment {
* @return A local execution environment with the specified parallelism.
*/
public static LocalStreamEnvironment createLocalEnvironment(int parallelism) {
- LocalStreamEnvironment env = new LocalStreamEnvironment();
- env.setParallelism(parallelism);
- return env;
+ return createLocalEnvironment(parallelism, new Configuration());
}
/**
@@ -1651,7 +1650,14 @@ public abstract class StreamExecutionEnvironment {
* @return A local execution environment with the specified parallelism.
*/
public static LocalStreamEnvironment createLocalEnvironment(int parallelism, Configuration configuration) {
- LocalStreamEnvironment currentEnvironment = new LocalStreamEnvironment(configuration);
+ final LocalStreamEnvironment currentEnvironment;
+
+ if (CoreOptions.FLIP6_MODE.equals(configuration.getString(CoreOptions.MODE))) {
+ currentEnvironment = new Flip6LocalStreamEnvironment(configuration);
+ } else {
+ currentEnvironment = new LocalStreamEnvironment(configuration);
+ }
+
currentEnvironment.setParallelism(parallelism);
return currentEnvironment;
}
@@ -1673,10 +1679,7 @@ public abstract class StreamExecutionEnvironment {
conf.setBoolean(ConfigConstants.LOCAL_START_WEBSERVER, true);
- LocalStreamEnvironment localEnv = new LocalStreamEnvironment(conf);
- localEnv.setParallelism(defaultLocalParallelism);
-
- return localEnv;
+ return createLocalEnvironment(defaultLocalParallelism, conf);
}
/**
[2/3] flink git commit: [FLINK-8698] [flip6] Let LocalExecutor use
Flip-6 MiniCluster
Posted by tr...@apache.org.
[FLINK-8698] [flip6] Let LocalExecutor use Flip-6 MiniCluster
This closes #5524.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/87f14d8b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/87f14d8b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/87f14d8b
Branch: refs/heads/master
Commit: 87f14d8b87da6a4e7242e925f025ef793a7adaea
Parents: 5c18fa5
Author: Till Rohrmann <tr...@apache.org>
Authored: Mon Feb 19 12:41:41 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Tue Feb 20 11:16:58 2018 +0100
----------------------------------------------------------------------
.../org/apache/flink/client/LocalExecutor.java | 109 ++++++++++++-------
.../flink/api/common/io/FileOutputFormat.java | 2 +-
.../flink/api/java/ExecutionEnvironment.java | 26 +++--
.../network/AbstractServerTest.java | 3 +-
.../flink/runtime/minicluster/MiniCluster.java | 8 ++
5 files changed, 102 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/87f14d8b/flink-clients/src/main/java/org/apache/flink/client/LocalExecutor.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/LocalExecutor.java b/flink-clients/src/main/java/org/apache/flink/client/LocalExecutor.java
index dd7e3ef..7d71f48 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/LocalExecutor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/LocalExecutor.java
@@ -27,17 +27,18 @@ import org.apache.flink.api.common.Program;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.optimizer.DataStatistics;
import org.apache.flink.optimizer.Optimizer;
import org.apache.flink.optimizer.dag.DataSinkNode;
import org.apache.flink.optimizer.plan.OptimizedPlan;
import org.apache.flink.optimizer.plandump.PlanJSONDumpGenerator;
import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
-import org.apache.flink.runtime.akka.AkkaUtils;
-import org.apache.flink.runtime.instance.ActorGateway;
import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.messages.JobManagerMessages;
+import org.apache.flink.runtime.minicluster.JobExecutorService;
import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
import java.util.List;
@@ -59,11 +60,14 @@ public class LocalExecutor extends PlanExecutor {
/** we lock to ensure singleton execution. */
private final Object lock = new Object();
- /** The mini cluster on which to execute the local programs. */
- private LocalFlinkMiniCluster flink;
-
/** Custom user configuration for the execution. */
- private final Configuration configuration;
+ private final Configuration baseConfiguration;
+
+ /** Service for executing Flink jobs. */
+ private JobExecutorService jobExecutorService;
+
+ /** Current job executor service configuration. */
+ private Configuration jobExecutorServiceConfiguration;
/** Config value for how many slots to provide in the local cluster. */
private int taskManagerNumSlots = DEFAULT_TASK_MANAGER_NUM_SLOTS;
@@ -78,7 +82,7 @@ public class LocalExecutor extends PlanExecutor {
}
public LocalExecutor(Configuration conf) {
- this.configuration = conf != null ? conf : new Configuration();
+ this.baseConfiguration = conf != null ? conf : new Configuration();
}
// ------------------------------------------------------------------------
@@ -106,27 +110,59 @@ public class LocalExecutor extends PlanExecutor {
@Override
public void start() throws Exception {
synchronized (lock) {
- if (flink == null) {
+ if (jobExecutorService == null) {
// create the embedded runtime
- Configuration configuration = createConfiguration();
- if (this.configuration != null) {
- configuration.addAll(this.configuration);
- }
+ jobExecutorServiceConfiguration = createConfiguration();
+
// start it up
- flink = new LocalFlinkMiniCluster(configuration, true);
- this.flink.start();
+ jobExecutorService = createJobExecutorService(jobExecutorServiceConfiguration);
} else {
throw new IllegalStateException("The local executor was already started.");
}
}
}
+ private JobExecutorService createJobExecutorService(Configuration configuration) throws Exception {
+ final JobExecutorService newJobExecutorService;
+ if (CoreOptions.FLIP6_MODE.equals(configuration.getString(CoreOptions.MODE))) {
+ final MiniClusterConfiguration miniClusterConfiguration = new MiniClusterConfiguration.Builder()
+ .setConfiguration(configuration)
+ .setNumJobManagers(
+ configuration.getInteger(
+ ConfigConstants.LOCAL_NUMBER_JOB_MANAGER,
+ ConfigConstants.DEFAULT_LOCAL_NUMBER_JOB_MANAGER))
+ .setNumResourceManagers(
+ configuration.getInteger(ResourceManagerOptions.LOCAL_NUMBER_RESOURCE_MANAGER))
+ .setNumTaskManagers(
+ configuration.getInteger(
+ ConfigConstants.LOCAL_NUMBER_TASK_MANAGER,
+ ConfigConstants.DEFAULT_LOCAL_NUMBER_TASK_MANAGER))
+ .setRpcServiceSharing(MiniClusterConfiguration.RpcServiceSharing.SHARED)
+ .setNumSlotsPerTaskManager(
+ configuration.getInteger(
+ ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 1))
+ .build();
+
+ final MiniCluster miniCluster = new MiniCluster(miniClusterConfiguration);
+ miniCluster.start();
+
+ newJobExecutorService = miniCluster;
+ } else {
+ final LocalFlinkMiniCluster localFlinkMiniCluster = new LocalFlinkMiniCluster(configuration, true);
+ localFlinkMiniCluster.start();
+
+ newJobExecutorService = localFlinkMiniCluster;
+ }
+
+ return newJobExecutorService;
+ }
+
@Override
public void stop() throws Exception {
synchronized (lock) {
- if (flink != null) {
- flink.stop();
- flink = null;
+ if (jobExecutorService != null) {
+ jobExecutorService.terminate().get();
+ jobExecutorService = null;
}
}
}
@@ -134,7 +170,7 @@ public class LocalExecutor extends PlanExecutor {
@Override
public boolean isRunning() {
synchronized (lock) {
- return flink != null;
+ return jobExecutorService != null;
}
}
@@ -162,7 +198,7 @@ public class LocalExecutor extends PlanExecutor {
// check if we start a session dedicated for this execution
final boolean shutDownAtEnd;
- if (flink == null) {
+ if (jobExecutorService == null) {
shutDownAtEnd = true;
// configure the number of local slots equal to the parallelism of the local plan
@@ -182,16 +218,18 @@ public class LocalExecutor extends PlanExecutor {
}
try {
- Configuration configuration = this.flink.configuration();
+ // TODO: Set job's default parallelism to max number of slots
+ final int slotsPerTaskManager = jobExecutorServiceConfiguration.getInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, taskManagerNumSlots);
+ final int numTaskManagers = jobExecutorServiceConfiguration.getInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1);
+ plan.setDefaultParallelism(slotsPerTaskManager * numTaskManagers);
- Optimizer pc = new Optimizer(new DataStatistics(), configuration);
+ Optimizer pc = new Optimizer(new DataStatistics(), jobExecutorServiceConfiguration);
OptimizedPlan op = pc.compile(plan);
- JobGraphGenerator jgg = new JobGraphGenerator(configuration);
+ JobGraphGenerator jgg = new JobGraphGenerator(jobExecutorServiceConfiguration);
JobGraph jobGraph = jgg.compileJobGraph(op, plan.getJobId());
- boolean sysoutPrint = isPrintingStatusDuringExecution();
- return flink.submitJobAndWait(jobGraph, sysoutPrint);
+ return jobExecutorService.executeJobBlocking(jobGraph);
}
finally {
if (shutDownAtEnd) {
@@ -212,7 +250,7 @@ public class LocalExecutor extends PlanExecutor {
public String getOptimizerPlanAsJSON(Plan plan) throws Exception {
final int parallelism = plan.getDefaultParallelism() == ExecutionConfig.PARALLELISM_DEFAULT ? 1 : plan.getDefaultParallelism();
- Optimizer pc = new Optimizer(new DataStatistics(), this.configuration);
+ Optimizer pc = new Optimizer(new DataStatistics(), this.baseConfiguration);
pc.setDefaultParallelism(parallelism);
OptimizedPlan op = pc.compile(plan);
@@ -221,20 +259,17 @@ public class LocalExecutor extends PlanExecutor {
@Override
public void endSession(JobID jobID) throws Exception {
- synchronized (this.lock) {
- LocalFlinkMiniCluster flink = this.flink;
- if (flink != null) {
- ActorGateway leaderGateway = flink.getLeaderGateway(AkkaUtils.getDefaultTimeoutAsFiniteDuration());
- leaderGateway.tell(new JobManagerMessages.RemoveCachedJob(jobID));
- }
- }
+ // no op
}
private Configuration createConfiguration() {
- Configuration configuration = new Configuration();
- configuration.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, getTaskManagerNumSlots());
- configuration.setBoolean(CoreOptions.FILESYTEM_DEFAULT_OVERRIDE, isDefaultOverwriteFiles());
- return configuration;
+ Configuration newConfiguration = new Configuration();
+ newConfiguration.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, getTaskManagerNumSlots());
+ newConfiguration.setBoolean(CoreOptions.FILESYTEM_DEFAULT_OVERRIDE, isDefaultOverwriteFiles());
+
+ newConfiguration.addAll(baseConfiguration);
+
+ return newConfiguration;
}
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/87f14d8b/flink-core/src/main/java/org/apache/flink/api/common/io/FileOutputFormat.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/FileOutputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/FileOutputFormat.java
index 585d7cc..5076c43 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/FileOutputFormat.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/io/FileOutputFormat.java
@@ -72,7 +72,7 @@ public abstract class FileOutputFormat<IT> extends RichOutputFormat<IT> implemen
* cluster execution, see LocalFlinkMiniCluster.
* @param configuration The configuration to load defaults from
*/
- private static void initDefaultsFromConfiguration(Configuration configuration) {
+ public static void initDefaultsFromConfiguration(Configuration configuration) {
final boolean overwrite = configuration.getBoolean(CoreOptions.FILESYTEM_DEFAULT_OVERRIDE);
DEFAULT_WRITE_MODE = overwrite ? WriteMode.OVERWRITE : WriteMode.NO_OVERWRITE;
http://git-wip-us.apache.org/repos/asf/flink/blob/87f14d8b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
index 02acfc6..1ce2221 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
@@ -1093,9 +1093,7 @@ public abstract class ExecutionEnvironment {
* @return A local execution environment with the specified parallelism.
*/
public static LocalEnvironment createLocalEnvironment(int parallelism) {
- LocalEnvironment lee = new LocalEnvironment();
- lee.setParallelism(parallelism);
- return lee;
+ return createLocalEnvironment(new Configuration(), parallelism);
}
/**
@@ -1107,7 +1105,7 @@ public abstract class ExecutionEnvironment {
* @return A local execution environment with the specified parallelism.
*/
public static LocalEnvironment createLocalEnvironment(Configuration customConfiguration) {
- return new LocalEnvironment(customConfiguration);
+ return createLocalEnvironment(customConfiguration, -1);
}
/**
@@ -1127,10 +1125,24 @@ public abstract class ExecutionEnvironment {
conf.setBoolean(ConfigConstants.LOCAL_START_WEBSERVER, true);
- LocalEnvironment localEnv = new LocalEnvironment(conf);
- localEnv.setParallelism(defaultLocalDop);
+ return createLocalEnvironment(conf, -1);
+ }
+
+ /**
+ * Creates a {@link LocalEnvironment} which is used for executing Flink jobs.
+ *
+ * @param configuration to start the {@link LocalEnvironment} with
+ * @param defaultParallelism to initialize the {@link LocalEnvironment} with
+ * @return {@link LocalEnvironment}
+ */
+ private static LocalEnvironment createLocalEnvironment(Configuration configuration, int defaultParallelism) {
+ final LocalEnvironment localEnvironment = new LocalEnvironment(configuration);
+
+ if (defaultParallelism > 0) {
+ localEnvironment.setParallelism(defaultParallelism);
+ }
- return localEnv;
+ return localEnvironment;
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/87f14d8b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java
----------------------------------------------------------------------
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java
index 103c638..4ef6bf7 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java
@@ -27,6 +27,7 @@ import org.apache.flink.queryablestate.network.stats.DisabledKvStateRequestStats
import org.apache.flink.queryablestate.network.stats.KvStateRequestStats;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
@@ -47,7 +48,7 @@ import java.util.concurrent.CompletableFuture;
/**
* Tests general behavior of the {@link AbstractServerBase}.
*/
-public class AbstractServerTest {
+public class AbstractServerTest extends TestLogger {
@Rule
public ExpectedException expectedEx = ExpectedException.none();
http://git-wip-us.apache.org/repos/asf/flink/blob/87f14d8b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
index c54bf54..7692ddf 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.minicluster;
import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.io.FileOutputFormat;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
@@ -159,6 +160,8 @@ public class MiniCluster implements JobExecutorService {
final boolean useSingleRpcService = miniClusterConfiguration.getRpcServiceSharing() == MiniClusterConfiguration.RpcServiceSharing.SHARED;
try {
+ initializeIOFormatClasses(configuration);
+
LOG.info("Starting Metrics Registry");
metricRegistry = createMetricRegistry(configuration);
@@ -605,6 +608,11 @@ public class MiniCluster implements JobExecutorService {
// miscellaneous utilities
// ------------------------------------------------------------------------
+ private void initializeIOFormatClasses(Configuration configuration) {
+ // TODO: That we still have to call something like this is a crime against humanity
+ FileOutputFormat.initDefaultsFromConfiguration(configuration);
+ }
+
private static Throwable shutDownRpc(RpcService rpcService, Throwable priorException) {
if (rpcService != null) {
try {
[3/3] flink git commit: [FLINK-8709] [tests] Harden
SlotPoolRpcTest.testCancelSlotAllocationWithoutResourceManager
Posted by tr...@apache.org.
[FLINK-8709] [tests] Harden SlotPoolRpcTest.testCancelSlotAllocationWithoutResourceManager
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d8753129
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d8753129
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d8753129
Branch: refs/heads/master
Commit: d87531294d4bd2823cbf9cec40bb5563ce38a70a
Parents: 87f14d8
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue Feb 20 09:00:31 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Tue Feb 20 11:16:58 2018 +0100
----------------------------------------------------------------------
.../runtime/jobmaster/slotpool/SlotPool.java | 24 +++++++++++---------
.../jobmaster/slotpool/SlotPoolGateway.java | 8 +++----
2 files changed, 17 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d8753129/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
index 6d714e1..b5cd129 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
@@ -50,6 +50,7 @@ import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
import org.apache.flink.runtime.util.clock.Clock;
import org.apache.flink.runtime.util.clock.SystemClock;
import org.apache.flink.util.AbstractID;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
@@ -657,24 +658,25 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
slotRequestId,
resourceProfile);
+ if (resourceManagerGateway == null) {
+ stashRequestWaitingForResourceManager(pendingRequest);
+ } else {
+ requestSlotFromResourceManager(resourceManagerGateway, pendingRequest);
+ }
+
// register request timeout
- FutureUtils
+ return FutureUtils
.orTimeout(pendingRequest.getAllocatedSlotFuture(), allocationTimeout.toMilliseconds(), TimeUnit.MILLISECONDS)
- .whenCompleteAsync(
- (AllocatedSlot ignored, Throwable throwable) -> {
+ .handleAsync(
+ (AllocatedSlot allocatedSlot, Throwable throwable) -> {
if (throwable != null) {
removePendingRequest(slotRequestId);
+ throw new CompletionException(ExceptionUtils.stripCompletionException(throwable));
+ } else {
+ return allocatedSlot;
}
},
getMainThreadExecutor());
-
- if (resourceManagerGateway == null) {
- stashRequestWaitingForResourceManager(pendingRequest);
- } else {
- requestSlotFromResourceManager(resourceManagerGateway, pendingRequest);
- }
-
- return pendingRequest.getAllocatedSlotFuture();
}
private void requestSlotFromResourceManager(
http://git-wip-us.apache.org/repos/asf/flink/blob/d8753129/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolGateway.java
index 0df6262..7d11681 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolGateway.java
@@ -38,7 +38,7 @@ import java.util.Collection;
import java.util.concurrent.CompletableFuture;
/**
- * The gateway for calls on the {@link SlotPool}.
+ * The gateway for calls on the {@link SlotPool}.
*/
public interface SlotPoolGateway extends AllocatedSlotActions, RpcGateway {
@@ -55,7 +55,7 @@ public interface SlotPoolGateway extends AllocatedSlotActions, RpcGateway {
/**
* Connects the SlotPool to the given ResourceManager. After this method is called, the
* SlotPool will be able to request resources from the given ResourceManager.
- *
+ *
* @param resourceManagerGateway The RPC gateway for the resource manager.
*/
void connectToResourceManager(ResourceManagerGateway resourceManagerGateway);
@@ -64,7 +64,7 @@ public interface SlotPoolGateway extends AllocatedSlotActions, RpcGateway {
* Disconnects the slot pool from its current Resource Manager. After this call, the pool will not
* be able to request further slots from the Resource Manager, and all currently pending requests
* to the resource manager will be canceled.
- *
+ *
* <p>The slot pool will still be able to serve slots from its internal pool.
*/
void disconnectResourceManager();
@@ -147,7 +147,7 @@ public interface SlotPoolGateway extends AllocatedSlotActions, RpcGateway {
* @param locationPreferences which define where the allocated slot should be placed, this can also be empty
* @param allowQueuedScheduling true if the slot request can be queued (e.g. the returned future must not be completed)
* @param timeout for the operation
- * @return
+ * @return Future which is completed with the allocated {@link LogicalSlot}
*/
CompletableFuture<LogicalSlot> allocateSlot(
SlotRequestId slotRequestId,