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/23 18:12:53 UTC
[6/7] flink git commit: [FLINK-8964][tests] Port
JobSubmissionFailsITCase to flip6
[FLINK-8964][tests] Port JobSubmissionFailsITCase to flip6
This closes #5727.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f0bd7b67
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f0bd7b67
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f0bd7b67
Branch: refs/heads/release-1.5
Commit: f0bd7b67aa348597cda9c4d3cf920ffd5a320896
Parents: 555e5e8
Author: zentol <ch...@apache.org>
Authored: Tue Mar 20 11:40:48 2018 +0100
Committer: zentol <ch...@apache.org>
Committed: Fri Mar 23 19:12:29 2018 +0100
----------------------------------------------------------------------
.../failing/JobSubmissionFailsITCase.java | 169 ++++++-------------
1 file changed, 55 insertions(+), 114 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/f0bd7b67/flink-tests/src/test/java/org/apache/flink/test/example/failing/JobSubmissionFailsITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/failing/JobSubmissionFailsITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/failing/JobSubmissionFailsITCase.java
index a647af9..ecd16a1 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/example/failing/JobSubmissionFailsITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/failing/JobSubmissionFailsITCase.java
@@ -19,30 +19,25 @@
package org.apache.flink.test.example.failing;
-import org.apache.flink.api.common.JobExecutionResult;
-import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.TaskManagerOptions;
-import org.apache.flink.runtime.client.JobExecutionException;
-import org.apache.flink.runtime.client.JobSubmissionException;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobVertex;
-import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
-import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.flink.runtime.testtasks.NoOpInvokable;
+import org.apache.flink.test.util.MiniClusterResource;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.TestLogger;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
+import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Optional;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/**
@@ -51,47 +46,32 @@ import static org.junit.Assert.fail;
@RunWith(Parameterized.class)
public class JobSubmissionFailsITCase extends TestLogger {
+ private static final int NUM_TM = 2;
private static final int NUM_SLOTS = 20;
- private static LocalFlinkMiniCluster cluster;
- private static JobGraph workingJobGraph;
-
- @BeforeClass
- public static void setup() {
- try {
- Configuration config = new Configuration();
- config.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, 4L);
- config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 2);
- config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, NUM_SLOTS / 2);
-
- cluster = new LocalFlinkMiniCluster(config);
-
- cluster.start();
-
- final JobVertex jobVertex = new JobVertex("Working job vertex.");
- jobVertex.setInvokableClass(NoOpInvokable.class);
- workingJobGraph = new JobGraph("Working testing job", jobVertex);
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
+ @ClassRule
+ public static final MiniClusterResource MINI_CLUSTER_RESOURCE = new MiniClusterResource(
+ new MiniClusterResource.MiniClusterResourceConfiguration(
+ getConfiguration(),
+ NUM_TM,
+ NUM_SLOTS / NUM_TM),
+ true);
+
+ private static Configuration getConfiguration() {
+ Configuration config = new Configuration();
+ config.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, 4L);
+ return config;
}
- @AfterClass
- public static void teardown() {
- try {
- cluster.stop();
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
+ private static JobGraph getWorkingJobGraph() {
+ final JobVertex jobVertex = new JobVertex("Working job vertex.");
+ jobVertex.setInvokableClass(NoOpInvokable.class);
+ return new JobGraph("Working testing job", jobVertex);
}
// --------------------------------------------------------------------------------------------
- private boolean detached;
+ private final boolean detached;
public JobSubmissionFailsITCase(boolean detached) {
this.detached = detached;
@@ -105,90 +85,51 @@ public class JobSubmissionFailsITCase extends TestLogger {
// --------------------------------------------------------------------------------------------
- private JobExecutionResult submitJob(JobGraph jobGraph) throws Exception {
- if (detached) {
- cluster.submitJobDetached(jobGraph);
- return null;
- }
- else {
- return cluster.submitJobAndWait(jobGraph, false, TestingUtils.TESTING_DURATION());
- }
- }
-
@Test
- public void testExceptionInInitializeOnMaster() {
- try {
- final JobVertex failingJobVertex = new FailingJobVertex("Failing job vertex");
- failingJobVertex.setInvokableClass(NoOpInvokable.class);
-
- final JobGraph failingJobGraph = new JobGraph("Failing testing job", failingJobVertex);
+ public void testExceptionInInitializeOnMaster() throws Exception {
+ final JobVertex failingJobVertex = new FailingJobVertex("Failing job vertex");
+ failingJobVertex.setInvokableClass(NoOpInvokable.class);
- try {
- submitJob(failingJobGraph);
- fail("Expected JobExecutionException.");
- }
- catch (JobExecutionException e) {
- assertEquals("Test exception.", e.getCause().getMessage());
- }
- catch (Throwable t) {
- t.printStackTrace();
- fail("Caught wrong exception of type " + t.getClass() + ".");
- }
+ final JobGraph failingJobGraph = new JobGraph("Failing testing job", failingJobVertex);
- cluster.submitJobAndWait(workingJobGraph, false);
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
- }
+ ClusterClient<?> client = MINI_CLUSTER_RESOURCE.getClusterClient();
+ client.setDetached(detached);
- @Test
- public void testSubmitEmptyJobGraph() {
try {
- final JobGraph jobGraph = new JobGraph("Testing job");
-
- try {
- submitJob(jobGraph);
- fail("Expected JobSubmissionException.");
- }
- catch (JobSubmissionException e) {
- assertTrue(e.getMessage() != null && e.getMessage().contains("empty"));
+ client.submitJob(failingJobGraph, JobSubmissionFailsITCase.class.getClassLoader());
+ fail("Job submission should have thrown an exception.");
+ } catch (Exception e) {
+ Optional<Throwable> expectedCause = ExceptionUtils.findThrowable(e,
+ candidate -> "Test exception.".equals(candidate.getMessage()));
+ if (!expectedCause.isPresent()) {
+ throw e;
}
- catch (Throwable t) {
- t.printStackTrace();
- fail("Caught wrong exception of type " + t.getClass() + ".");
- }
-
- cluster.submitJobAndWait(workingJobGraph, false);
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
}
+
+ client.setDetached(false);
+ client.submitJob(getWorkingJobGraph(), JobSubmissionFailsITCase.class.getClassLoader());
}
@Test
- public void testSubmitNullJobGraph() {
+ public void testSubmitEmptyJobGraph() throws Exception {
+ final JobGraph jobGraph = new JobGraph("Testing job");
+
+ ClusterClient<?> client = MINI_CLUSTER_RESOURCE.getClusterClient();
+ client.setDetached(detached);
+
try {
- try {
- submitJob(null);
- fail("Expected JobSubmissionException.");
- }
- catch (NullPointerException e) {
- // yo!
+ client.submitJob(jobGraph, JobSubmissionFailsITCase.class.getClassLoader());
+ fail("Job submission should have thrown an exception.");
+ } catch (Exception e) {
+ Optional<Throwable> expectedCause = ExceptionUtils.findThrowable(e,
+ throwable -> throwable.getMessage() != null && throwable.getMessage().contains("empty"));
+ if (!expectedCause.isPresent()) {
+ throw e;
}
- catch (Throwable t) {
- t.printStackTrace();
- fail("Caught wrong exception of type " + t.getClass() + ".");
- }
-
- cluster.submitJobAndWait(workingJobGraph, false);
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
}
+
+ client.setDetached(false);
+ client.submitJob(getWorkingJobGraph(), JobSubmissionFailsITCase.class.getClassLoader());
}
// --------------------------------------------------------------------------------------------