flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [2/2] flink git commit: [FLINK-4672] [taskmanager] Do not decorate Actor Kill messages
Date Mon, 26 Sep 2016 12:11:42 GMT
[FLINK-4672] [taskmanager] Do not decorate Actor Kill messages


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6f237cfe
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6f237cfe
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6f237cfe

Branch: refs/heads/master
Commit: 6f237cfe6f70b5b72fedd3dea6fbeb6c929631e8
Parents: 28ff5a3
Author: Stephan Ewen <sewen@apache.org>
Authored: Fri Sep 23 18:42:47 2016 +0200
Committer: Stephan Ewen <sewen@apache.org>
Committed: Mon Sep 26 14:11:05 2016 +0200

----------------------------------------------------------------------
 .../flink/runtime/taskmanager/TaskManager.scala |  2 +-
 .../runtime/taskmanager/TaskManagerTest.java    | 33 ++++++++++++++++++--
 2 files changed, 31 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/6f237cfe/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
index 9e2feb5..04f3168 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -1379,7 +1379,7 @@ class TaskManager(
       "\n" +
       "A fatal error occurred, forcing the TaskManager to shut down: " + message, cause)
 
-    self ! decorateMessage(Kill)
+    self ! Kill
   }
 
   override def notifyLeaderAddress(leaderAddress: String, leaderSessionID: UUID): Unit =
{

http://git-wip-us.apache.org/repos/asf/flink/blob/6f237cfe/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
index 0e53673..0774fd5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.taskmanager;
 
 import akka.actor.ActorRef;
 import akka.actor.ActorSystem;
+import akka.actor.Kill;
 import akka.actor.Props;
 import akka.japi.Creator;
 import akka.testkit.JavaTestKit;
@@ -55,6 +56,7 @@ import org.apache.flink.runtime.messages.StackTraceSampleMessages.ResponseStackT
 import org.apache.flink.runtime.messages.StackTraceSampleMessages.ResponseStackTraceSampleSuccess;
 import org.apache.flink.runtime.messages.StackTraceSampleMessages.TriggerStackTraceSample;
 import org.apache.flink.runtime.messages.TaskManagerMessages;
+import org.apache.flink.runtime.messages.TaskManagerMessages.FatalError;
 import org.apache.flink.runtime.messages.TaskMessages;
 import org.apache.flink.runtime.messages.TaskMessages.CancelTask;
 import org.apache.flink.runtime.messages.TaskMessages.PartitionState;
@@ -1367,6 +1369,28 @@ public class TaskManagerTest extends TestLogger {
 		}};
 	}
 
+	@Test
+	public void testTerminationOnFatalError() {
+		new JavaTestKit(system){{
+
+			final ActorGateway taskManager = TestingUtils.createTaskManager(
+					system,
+					system.deadLetters(), // no jobmanager
+					new Configuration(),
+					true,
+					false);
+
+			try {
+				watch(taskManager.actor());
+				taskManager.tell(new FatalError("test fatal error", new Exception("something super bad")));
+				expectTerminated(d, taskManager.actor());
+			}
+			finally {
+				taskManager.tell(Kill.getInstance());
+			}
+		}};
+	}
+	
 	// --------------------------------------------------------------------------------------------
 
 	public static class SimpleJobManager extends FlinkUntypedActor {
@@ -1547,11 +1571,14 @@ public class TaskManagerTest extends TestLogger {
 
 		@Override
 		public void invoke() throws Exception {
-			Object o = new Object();
+			final Object o = new Object();
+			//noinspection SynchronizationOnLocalVariableOrMethodParameter
 			synchronized (o) {
-				o.wait();
+				//noinspection InfiniteLoopStatement
+				while (true) {
+					o.wait();
+				}
 			}
 		}
 	}
-
 }


Mime
View raw message