You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/08/31 14:13:57 UTC
[4/5] flink git commit: [FLINK-2543] [core] Make exception
communication and result/failure notifications consistent with respect to
serialization of exceptions
http://git-wip-us.apache.org/repos/asf/flink/blob/13a95961/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/util/ClusterUtil.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/util/ClusterUtil.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/util/ClusterUtil.java
index 763885c..c2ab424 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/util/ClusterUtil.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/util/ClusterUtil.java
@@ -20,9 +20,9 @@ package org.apache.flink.streaming.util;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.client.SerializedJobExecutionResult;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,8 +67,7 @@ public class ClusterUtil {
exec.submitJobDetached(jobGraph);
return null;
} else {
- SerializedJobExecutionResult result = exec.submitJobAndWait(jobGraph, printDuringExecution);
- return result.toJobExecutionResult(ClusterUtil.class.getClassLoader());
+ return exec.submitJobAndWait(jobGraph, printDuringExecution);
}
} finally {
if (exec != null && !detached) {
http://git-wip-us.apache.org/repos/asf/flink/blob/13a95961/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java
index ae31d95..ccaa486 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java
@@ -20,10 +20,10 @@ package org.apache.flink.test.accumulators;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
-import akka.actor.Status;
import akka.pattern.Patterns;
import akka.testkit.JavaTestKit;
import akka.util.Timeout;
+
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.Plan;
@@ -55,11 +55,14 @@ import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
+
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
@@ -190,7 +193,7 @@ public class AccumulatorLiveITCase {
// submit job
jobManagerGateway.tell(new JobManagerMessages.SubmitJob(jobGraph, false), selfGateway);
- expectMsgClass(TIMEOUT, Status.Success.class);
+ expectMsgClass(TIMEOUT, JobManagerMessages.JobSubmitSuccess.class);
TestingJobManagerMessages.UpdatedAccumulators msg = (TestingJobManagerMessages.UpdatedAccumulators) receiveOne(TIMEOUT);
http://git-wip-us.apache.org/repos/asf/flink/blob/13a95961/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java b/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java
index e43a9e4..5c2f2dc 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java
@@ -26,11 +26,13 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.test.testdata.KMeansData;
import org.apache.flink.test.util.ForkableFlinkMiniCluster;
-import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
public class ClassLoaderITCase {
private static final String INPUT_SPLITS_PROG_JAR_FILE = "target/customsplit-test-jar.jar";
@@ -47,7 +49,6 @@ public class ClassLoaderITCase {
@Test
public void testJobsWithCustomClassLoader() {
try {
-
Configuration config = new Configuration();
config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 2);
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 2);
@@ -62,37 +63,45 @@ public class ClassLoaderITCase {
try {
int port = testCluster.getJobManagerRPCPort();
- PackagedProgram inputSplitTestProg = new PackagedProgram(new File(INPUT_SPLITS_PROG_JAR_FILE),
+ PackagedProgram inputSplitTestProg = new PackagedProgram(
+ new File(INPUT_SPLITS_PROG_JAR_FILE),
new String[] { INPUT_SPLITS_PROG_JAR_FILE,
"localhost",
String.valueOf(port),
"4" // parallelism
- } );
+ });
inputSplitTestProg.invokeInteractiveModeForExecution();
// regular streaming job
- PackagedProgram streamingProg = new PackagedProgram(new File(STREAMING_PROG_JAR_FILE),
- new String[] { STREAMING_PROG_JAR_FILE, "localhost", String.valueOf(port) } );
+ PackagedProgram streamingProg = new PackagedProgram(
+ new File(STREAMING_PROG_JAR_FILE),
+ new String[] {
+ STREAMING_PROG_JAR_FILE,
+ "localhost",
+ String.valueOf(port)
+ });
streamingProg.invokeInteractiveModeForExecution();
// checkpointed streaming job with custom classes for the checkpoint (FLINK-2543)
// the test also ensures that user specific exceptions are serializable between JobManager <--> JobClient.
try {
- PackagedProgram streamingCheckpointedProg = new PackagedProgram(new File(STREAMING_CHECKPOINTED_PROG_JAR_FILE),
- new String[]{STREAMING_CHECKPOINTED_PROG_JAR_FILE, "localhost", String.valueOf(port)});
+ PackagedProgram streamingCheckpointedProg = new PackagedProgram(
+ new File(STREAMING_CHECKPOINTED_PROG_JAR_FILE),
+ new String[] {
+ STREAMING_CHECKPOINTED_PROG_JAR_FILE,
+ "localhost",
+ String.valueOf(port)});
streamingCheckpointedProg.invokeInteractiveModeForExecution();
- } catch(Exception e) {
+ }
+ catch (Exception e) {
// we can not access the SuccessException here when executing the tests with maven, because its not available in the jar.
- try {
- if (!(e.getCause().getCause().getClass().getCanonicalName().equals("org.apache.flink.test.classloading.jar.CheckpointedStreamingProgram.SuccessException"))) {
- throw e;
- }
- } catch(Throwable ignore) {
- throw e;
- }
+ assertEquals("Program should terminate with a 'SuccessException'",
+ "org.apache.flink.test.classloading.jar.CheckpointedStreamingProgram.SuccessException",
+ e.getCause().getCause().getClass().getCanonicalName());
}
- PackagedProgram kMeansProg = new PackagedProgram(new File(KMEANS_JAR_PATH),
+ PackagedProgram kMeansProg = new PackagedProgram(
+ new File(KMEANS_JAR_PATH),
new String[] { KMEANS_JAR_PATH,
"localhost",
String.valueOf(port),
@@ -100,7 +109,7 @@ public class ClassLoaderITCase {
KMeansData.DATAPOINTS,
KMeansData.INITIAL_CENTERS,
"25"
- } );
+ });
kMeansProg.invokeInteractiveModeForExecution();
}
finally {
@@ -109,8 +118,7 @@ public class ClassLoaderITCase {
}
catch (Exception e) {
e.printStackTrace();
- Assert.fail(e.getMessage());
+ fail(e.getMessage());
}
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/13a95961/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerFailsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerFailsITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerFailsITCase.scala
index feb90b2..577b5c6 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerFailsITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerFailsITCase.scala
@@ -21,7 +21,9 @@ package org.apache.flink.api.scala.runtime.jobmanager
import akka.actor.Status.Success
import akka.actor.{ActorSystem, PoisonPill}
import akka.testkit.{ImplicitSender, TestKit}
+
import org.junit.runner.RunWith
+
import org.scalatest.junit.JUnitRunner
import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
@@ -57,7 +59,7 @@ class JobManagerFailsITCase(_system: ActorSystem)
val cluster = startDeathwatchCluster(num_slots, 1)
val tm = cluster.getTaskManagers(0)
- val jmGateway = cluster.getJobManagerGateway
+ val jmGateway = cluster.getJobManagerGateway()
// disable disconnect message to test death watch
tm ! DisableDisconnect
@@ -76,7 +78,7 @@ class JobManagerFailsITCase(_system: ActorSystem)
cluster.waitForTaskManagersToBeRegistered()
- cluster.getJobManagerGateway.tell(RequestNumberRegisteredTaskManager, self)
+ cluster.getJobManagerGateway().tell(RequestNumberRegisteredTaskManager, self)
expectMsg(1)
} finally {
@@ -99,13 +101,13 @@ class JobManagerFailsITCase(_system: ActorSystem)
val cluster = startDeathwatchCluster(num_slots / 2, 2)
- var jmGateway = cluster.getJobManagerGateway
+ var jmGateway = cluster.getJobManagerGateway()
val tm = cluster.getTaskManagers(0)
try {
within(TestingUtils.TESTING_DURATION) {
jmGateway.tell(SubmitJob(jobGraph, false), self)
- expectMsg(Success(jobGraph.getJobID))
+ expectMsg(JobSubmitSuccess(jobGraph.getJobID))
tm.tell(NotifyWhenJobManagerTerminated(jmGateway.actor()), self)
@@ -115,13 +117,13 @@ class JobManagerFailsITCase(_system: ActorSystem)
cluster.restartJobManager()
- jmGateway = cluster.getJobManagerGateway
+ jmGateway = cluster.getJobManagerGateway()
cluster.waitForTaskManagersToBeRegistered()
jmGateway.tell(SubmitJob(jobGraph2, false), self)
- val failure = expectMsgType[Success]
+ expectMsg(JobSubmitSuccess(jobGraph2.getJobID()))
val result = expectMsgType[JobResultSuccess]
http://git-wip-us.apache.org/repos/asf/flink/blob/13a95961/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala
index 1952760..a37fae3 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala
@@ -28,7 +28,7 @@ import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.client.JobExecutionException
import org.apache.flink.runtime.jobgraph.{JobVertex, DistributionPattern, JobGraph}
import org.apache.flink.runtime.jobmanager.Tasks.{NoOpInvokable, BlockingNoOpInvokable, BlockingReceiver, Sender}
-import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultSuccess, RequestNumberRegisteredTaskManager, SubmitJob}
+import org.apache.flink.runtime.messages.JobManagerMessages._
import org.apache.flink.runtime.messages.TaskManagerMessages.{RegisteredAtJobManager, NotifyWhenRegisteredAtJobManager}
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages._
import org.apache.flink.runtime.testingUtils.TestingMessages.DisableDisconnect
@@ -63,7 +63,7 @@ class TaskManagerFailsITCase(_system: ActorSystem)
val cluster = startDeathwatchCluster(num_slots, 2)
val taskManagers = cluster.getTaskManagers
- val jmGateway = cluster.getJobManagerGateway
+ val jmGateway = cluster.getJobManagerGateway()
jmGateway.tell(DisableDisconnect)
@@ -103,12 +103,12 @@ class TaskManagerFailsITCase(_system: ActorSystem)
val cluster = ForkableFlinkMiniCluster.startCluster(num_tasks, 2)
- val jmGateway = cluster.getJobManagerGateway
+ val jmGateway = cluster.getJobManagerGateway()
try {
within(TestingUtils.TESTING_DURATION) {
jmGateway.tell(SubmitJob(jobGraph, false), self)
- expectMsg(Success(jobGraph.getJobID))
+ expectMsg(JobSubmitSuccess(jobGraph.getJobID()))
jmGateway.tell(WaitForAllVerticesToBeRunningOrFinished(jobID), self)
@@ -126,8 +126,8 @@ class TaskManagerFailsITCase(_system: ActorSystem)
case None => fail("Could not retrieve a working task manager.")
}
- val failure = expectMsgType[Failure]
- val exception = SerializedThrowable.get(failure.cause, this.getClass.getClassLoader)
+ val failure = expectMsgType[JobResultFailure]
+ val exception = failure.cause.deserializeError(getClass.getClassLoader())
exception match {
case e: JobExecutionException =>
jobGraph.getJobID should equal(e.getJobID)
@@ -156,12 +156,12 @@ class TaskManagerFailsITCase(_system: ActorSystem)
val cluster = ForkableFlinkMiniCluster.startCluster(num_tasks, 2)
val taskManagers = cluster.getTaskManagers
- val jmGateway = cluster.getJobManagerGateway
+ val jmGateway = cluster.getJobManagerGateway()
try {
within(TestingUtils.TESTING_DURATION) {
jmGateway.tell(SubmitJob(jobGraph, false), self)
- expectMsg(Success(jobGraph.getJobID))
+ expectMsg(JobSubmitSuccess(jobGraph.getJobID()))
jmGateway.tell(WaitForAllVerticesToBeRunningOrFinished(jobID), self)
expectMsg(AllVerticesRunning(jobID))
@@ -169,8 +169,8 @@ class TaskManagerFailsITCase(_system: ActorSystem)
// kill one task manager
taskManagers(0) ! Kill
- val failure = expectMsgType[Failure]
- val exception = SerializedThrowable.get(failure.cause, this.getClass.getClassLoader)
+ val failure = expectMsgType[JobResultFailure]
+ val exception = failure.cause.deserializeError(getClass.getClassLoader())
exception match {
case e: JobExecutionException =>
jobGraph.getJobID should equal(e.getJobID)
@@ -199,17 +199,17 @@ class TaskManagerFailsITCase(_system: ActorSystem)
val cluster = startDeathwatchCluster(num_slots/2, 2)
var tm = cluster.getTaskManagers(0)
- val jmGateway = cluster.getJobManagerGateway
+ val jmGateway = cluster.getJobManagerGateway()
try{
within(TestingUtils.TESTING_DURATION){
jmGateway.tell(SubmitJob(jobGraph, false), self)
- expectMsg(Success(jobGraph.getJobID))
+ expectMsg(JobSubmitSuccess(jobGraph.getJobID))
tm ! PoisonPill
- val failure = expectMsgType[Failure]
- val exception = SerializedThrowable.get(failure.cause, this.getClass.getClassLoader)
+ val failure = expectMsgType[JobResultFailure]
+ val exception = failure.cause.deserializeError(getClass.getClassLoader())
exception match {
case e: JobExecutionException =>
jobGraph.getJobID should equal(e.getJobID)
@@ -227,7 +227,7 @@ class TaskManagerFailsITCase(_system: ActorSystem)
jmGateway.tell(SubmitJob(jobGraph2, false), self)
- expectMsgType[Success]
+ expectMsg(JobSubmitSuccess(jobGraph2.getJobID()))
val result = expectMsgType[JobResultSuccess]
result.result.getJobId() should equal(jobGraph2.getJobID)