You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2018/03/09 08:40:25 UTC
[5/5] flink git commit: [hotfix][tests] Do not use singleActorSystem
in LocalFlinkMiniCluster
[hotfix][tests] Do not use singleActorSystem in LocalFlinkMiniCluster
Using a singleActorSystem rendered the returned client unusable.
This closes #5652.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4e0bc008
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4e0bc008
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4e0bc008
Branch: refs/heads/master
Commit: 4e0bc008e162bfbdec2151fbb10f815b81bc24db
Parents: 94e959f
Author: zentol <ch...@apache.org>
Authored: Mon Mar 5 13:45:33 2018 +0100
Committer: zentol <ch...@apache.org>
Committed: Fri Mar 9 09:39:38 2018 +0100
----------------------------------------------------------------------
.../flink/test/util/MiniClusterResource.java | 43 +++++++++++++++-----
1 file changed, 33 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4e0bc008/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/MiniClusterResource.java
----------------------------------------------------------------------
diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/MiniClusterResource.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/MiniClusterResource.java
index 954b06f..66cbb9f 100644
--- a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/MiniClusterResource.java
+++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/MiniClusterResource.java
@@ -62,6 +62,8 @@ public class MiniClusterResource extends ExternalResource {
private JobExecutorService jobExecutorService;
+ private final boolean enableClusterClient;
+
private ClusterClient<?> clusterClient;
private int numberSlots = -1;
@@ -69,16 +71,25 @@ public class MiniClusterResource extends ExternalResource {
private TestEnvironment executionEnvironment;
public MiniClusterResource(final MiniClusterResourceConfiguration miniClusterResourceConfiguration) {
+ this(miniClusterResourceConfiguration, false);
+ }
+
+ public MiniClusterResource(
+ final MiniClusterResourceConfiguration miniClusterResourceConfiguration,
+ final boolean enableClusterClient) {
this(
miniClusterResourceConfiguration,
- Objects.equals(FLIP6_CODEBASE, System.getProperty(CODEBASE_KEY)) ? MiniClusterType.FLIP6 : MiniClusterType.OLD);
+ Objects.equals(FLIP6_CODEBASE, System.getProperty(CODEBASE_KEY)) ? MiniClusterType.FLIP6 : MiniClusterType.OLD,
+ enableClusterClient);
}
- public MiniClusterResource(
+ private MiniClusterResource(
final MiniClusterResourceConfiguration miniClusterResourceConfiguration,
- final MiniClusterType miniClusterType) {
+ final MiniClusterType miniClusterType,
+ final boolean enableClusterClient) {
this.miniClusterResourceConfiguration = Preconditions.checkNotNull(miniClusterResourceConfiguration);
this.miniClusterType = Preconditions.checkNotNull(miniClusterType);
+ this.enableClusterClient = enableClusterClient;
}
public int getNumberSlots() {
@@ -86,6 +97,12 @@ public class MiniClusterResource extends ExternalResource {
}
public ClusterClient<?> getClusterClient() {
+ if (!enableClusterClient) {
+ // this check is technically only necessary for legacy clusters
+ // we still fail here for flip6 to keep the behaviors in sync
+ throw new IllegalStateException("To use the client you must enable it with the constructor.");
+ }
+
return clusterClient;
}
@@ -113,10 +130,12 @@ public class MiniClusterResource extends ExternalResource {
Exception exception = null;
- try {
- clusterClient.shutdown();
- } catch (Exception e) {
- exception = e;
+ if (clusterClient != null) {
+ try {
+ clusterClient.shutdown();
+ } catch (Exception e) {
+ exception = e;
+ }
}
clusterClient = null;
@@ -158,10 +177,12 @@ public class MiniClusterResource extends ExternalResource {
final LocalFlinkMiniCluster flinkMiniCluster = TestBaseUtils.startCluster(
configuration,
- true);
+ !enableClusterClient); // the cluster client only works if separate actor systems are used
jobExecutorService = flinkMiniCluster;
- clusterClient = new StandaloneClusterClient(configuration, flinkMiniCluster.highAvailabilityServices(), true);
+ if (enableClusterClient) {
+ clusterClient = new StandaloneClusterClient(configuration, flinkMiniCluster.highAvailabilityServices(), true);
+ }
}
private void startFlip6MiniCluster() throws Exception {
@@ -188,7 +209,9 @@ public class MiniClusterResource extends ExternalResource {
configuration.setInteger(RestOptions.REST_PORT, miniCluster.getRestAddress().getPort());
jobExecutorService = miniCluster;
- clusterClient = new MiniClusterClient(configuration, miniCluster);
+ if (enableClusterClient) {
+ clusterClient = new MiniClusterClient(configuration, miniCluster);
+ }
}
/**