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/06/22 15:45:37 UTC
[4/9] flink git commit: [hotfix] Only run MiniClusterResource if
specified CodebaseType equals actual CodebaseType
[hotfix] Only run MiniClusterResource if specified CodebaseType equals actual CodebaseType
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e64d4dbd
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e64d4dbd
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e64d4dbd
Branch: refs/heads/master
Commit: e64d4dbd3e8b02b7070118b007cbfd0433da7b54
Parents: 8bb086d
Author: Till Rohrmann <tr...@apache.org>
Authored: Thu Jun 21 15:08:18 2018 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Fri Jun 22 17:45:04 2018 +0200
----------------------------------------------------------------------
.../runtime/webmonitor/WebFrontendITCase.java | 7 ++-
.../flink/test/util/MiniClusterResource.java | 46 +++++++++++++-------
.../flink/test/misc/AutoParallelismITCase.java | 2 +-
3 files changed, 35 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e64d4dbd/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/WebFrontendITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/WebFrontendITCase.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/WebFrontendITCase.java
index 776a267..c20a44e 100644
--- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/WebFrontendITCase.java
+++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/WebFrontendITCase.java
@@ -56,7 +56,6 @@ import java.nio.file.Files;
import java.util.Collection;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -155,7 +154,7 @@ public class WebFrontendITCase extends TestLogger {
if (notFoundJobConnection.getResponseCode() >= 400) {
// we don't set the content-encoding header
Assert.assertNull(notFoundJobConnection.getContentEncoding());
- if (Objects.equals(CLUSTER.getMiniClusterType(), TestBaseUtils.CodebaseType.NEW)) {
+ if (CLUSTER.getCodebaseType() == TestBaseUtils.CodebaseType.NEW) {
Assert.assertEquals("application/json; charset=UTF-8", notFoundJobConnection.getContentType());
} else {
Assert.assertEquals("text/plain; charset=UTF-8", notFoundJobConnection.getContentType());
@@ -283,7 +282,7 @@ public class WebFrontendITCase extends TestLogger {
final Deadline deadline = testTimeout.fromNow();
try (HttpTestClient client = new HttpTestClient("localhost", CLUSTER.getWebUIPort())) {
- if (Objects.equals(CLUSTER.getMiniClusterType(), TestBaseUtils.CodebaseType.NEW)) {
+ if (CLUSTER.getCodebaseType() == TestBaseUtils.CodebaseType.NEW) {
// stop the job
client.sendPatchRequest("/jobs/" + jid + "/?mode=stop", deadline.timeLeft());
HttpTestClient.SimpleHttpResponse response = client.getNextResponse(deadline.timeLeft());
@@ -358,7 +357,7 @@ public class WebFrontendITCase extends TestLogger {
HttpTestClient.SimpleHttpResponse response = client
.getNextResponse(deadline.timeLeft());
- if (Objects.equals(CLUSTER.getMiniClusterType(), TestBaseUtils.CodebaseType.NEW)) {
+ if (CLUSTER.getCodebaseType() == TestBaseUtils.CodebaseType.NEW) {
assertEquals(HttpResponseStatus.ACCEPTED, response.getStatus());
} else {
assertEquals(HttpResponseStatus.OK, response.getStatus());
http://git-wip-us.apache.org/repos/asf/flink/blob/e64d4dbd/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 bbb945f..150d956 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
@@ -21,7 +21,13 @@ package org.apache.flink.test.util;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.MiniClusterClient;
import org.apache.flink.client.program.StandaloneClusterClient;
-import org.apache.flink.configuration.*;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.configuration.UnmodifiableConfiguration;
import org.apache.flink.runtime.minicluster.JobExecutorService;
import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
import org.apache.flink.runtime.minicluster.MiniCluster;
@@ -30,6 +36,8 @@ import org.apache.flink.streaming.util.TestStreamEnvironment;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.Preconditions;
+
+import org.junit.Assume;
import org.junit.rules.ExternalResource;
import org.junit.rules.TemporaryFolder;
import org.slf4j.Logger;
@@ -38,6 +46,9 @@ import org.slf4j.LoggerFactory;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.is;
+
/**
* Starts a Flink mini cluster as a resource and registers the respective
* ExecutionEnvironment and StreamExecutionEnvironment.
@@ -50,7 +61,7 @@ public class MiniClusterResource extends ExternalResource {
private final MiniClusterResourceConfiguration miniClusterResourceConfiguration;
- private final TestBaseUtils.CodebaseType miniClusterType;
+ private final TestBaseUtils.CodebaseType codebaseType;
private JobExecutorService jobExecutorService;
@@ -74,12 +85,12 @@ public class MiniClusterResource extends ExternalResource {
final MiniClusterResourceConfiguration miniClusterResourceConfiguration,
final boolean enableClusterClient) {
this.miniClusterResourceConfiguration = Preconditions.checkNotNull(miniClusterResourceConfiguration);
- this.miniClusterType = miniClusterResourceConfiguration.getCodebaseType();
+ this.codebaseType = miniClusterResourceConfiguration.getCodebaseType();
this.enableClusterClient = enableClusterClient;
}
- public TestBaseUtils.CodebaseType getMiniClusterType() {
- return miniClusterType;
+ public TestBaseUtils.CodebaseType getCodebaseType() {
+ return codebaseType;
}
public int getNumberSlots() {
@@ -110,9 +121,12 @@ public class MiniClusterResource extends ExternalResource {
@Override
public void before() throws Exception {
+ // verify that we are running in the correct test profile
+ Assume.assumeThat(TestBaseUtils.getCodebaseType(), is(equalTo(codebaseType)));
+
temporaryFolder.create();
- startJobExecutorService(miniClusterType);
+ startJobExecutorService(codebaseType);
numberSlots = miniClusterResourceConfiguration.getNumberSlotsPerTaskManager() * miniClusterResourceConfiguration.getNumberTaskManagers();
@@ -140,17 +154,19 @@ public class MiniClusterResource extends ExternalResource {
clusterClient = null;
- final CompletableFuture<?> terminationFuture = jobExecutorService.closeAsync();
+ if (jobExecutorService != null) {
+ final CompletableFuture<?> terminationFuture = jobExecutorService.closeAsync();
- try {
- terminationFuture.get(
- miniClusterResourceConfiguration.getShutdownTimeout().toMilliseconds(),
- TimeUnit.MILLISECONDS);
- } catch (Exception e) {
- exception = ExceptionUtils.firstOrSuppressed(e, exception);
- }
+ try {
+ terminationFuture.get(
+ miniClusterResourceConfiguration.getShutdownTimeout().toMilliseconds(),
+ TimeUnit.MILLISECONDS);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
- jobExecutorService = null;
+ jobExecutorService = null;
+ }
if (exception != null) {
LOG.warn("Could not properly shut down the MiniClusterResource.", exception);
http://git-wip-us.apache.org/repos/asf/flink/blob/e64d4dbd/flink-tests/src/test/java/org/apache/flink/test/misc/AutoParallelismITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/misc/AutoParallelismITCase.java b/flink-tests/src/test/java/org/apache/flink/test/misc/AutoParallelismITCase.java
index 189962c..4826a46 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/misc/AutoParallelismITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/misc/AutoParallelismITCase.java
@@ -80,7 +80,7 @@ public class AutoParallelismITCase extends TestLogger {
assertEquals(PARALLELISM, resultCollection.size());
}
catch (Exception ex) {
- if (MINI_CLUSTER_RESOURCE.getMiniClusterType().equals(TestBaseUtils.CodebaseType.LEGACY)) {
+ if (MINI_CLUSTER_RESOURCE.getCodebaseType() == TestBaseUtils.CodebaseType.LEGACY) {
throw ex;
}
assertTrue(