Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/750#discussion_r37990019
--- Diff:
flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
---
@@ -363,7 +363,135 @@ protected void run() {
}
}};
}
-
+
+ @Test
+ public void testJobSubmissionAndStop() {
+
+ LOG.info(
"--------------------------------------------------------------------\n" +
+ " Starting
testJobSubmissionAndStop() \n" +
+
"--------------------------------------------------------------------");
+
+ new JavaTestKit(system){{
+
+ ActorRef jobManager = null;
+ ActorRef taskManager = null;
+ try {
+ jobManager =
system.actorOf(Props.create(SimpleJobManager.class, leaderSessionID.get()));
+ taskManager = createTaskManager(jobManager,
true);
+
+ final JobID jid1 = new JobID();
+ final JobID jid2 = new JobID();
+
+ JobVertexID vid1 = new JobVertexID();
+ JobVertexID vid2 = new JobVertexID();
+
+ final ExecutionAttemptID eid1 = new
ExecutionAttemptID();
+ final ExecutionAttemptID eid2 = new
ExecutionAttemptID();
+
+ final TaskDeploymentDescriptor tdd1 = new
TaskDeploymentDescriptor(jid1, vid1, eid1, "TestTask1", 1, 5,
+ new Configuration(), new
Configuration(), StopableInvokable.class.getName(),
+
Collections.<ResultPartitionDeploymentDescriptor>emptyList(),
+
Collections.<InputGateDeploymentDescriptor>emptyList(),
+ new ArrayList<BlobKey>(), 0);
+
+ final TaskDeploymentDescriptor tdd2 = new
TaskDeploymentDescriptor(jid2, vid2, eid2, "TestTask2", 2, 7,
+ new Configuration(), new
Configuration(), TestInvokableBlockingCancelable.class.getName(),
+
Collections.<ResultPartitionDeploymentDescriptor>emptyList(),
+
Collections.<InputGateDeploymentDescriptor>emptyList(),
+ new ArrayList<BlobKey>(), 0);
+
+ final ActorRef tm = taskManager;
+
+ new Within(d) {
+
+ @Override
+ protected void run() {
+ try {
+ Future<Object>
t1Running = Patterns.ask(
+ tm,
+ new
TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid1),
+
timeout);
+ Future<Object>
t2Running = Patterns.ask(
+ tm,
+ new
TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid2),
+
timeout);
+
+
tm.tell(decorator.decorate(new SubmitTask(tdd1)), getRef());
+
tm.tell(decorator.decorate(new SubmitTask(tdd2)), getRef());
+
+
expectMsgEquals(Messages.getAcknowledge());
+
expectMsgEquals(Messages.getAcknowledge());
+
+ Await.ready(t1Running,
d);
+ Await.ready(t2Running,
d);
+
+
tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
+
+ Map<ExecutionAttemptID,
Task> runningTasks = expectMsgClass(TestingTaskManagerMessages
+
.ResponseRunningTasks.class).asJava();
+
+ assertEquals(2,
runningTasks.size());
+ Task t1 =
runningTasks.get(eid1);
+ Task t2 =
runningTasks.get(eid2);
+ assertNotNull(t1);
+ assertNotNull(t2);
+
+
assertEquals(ExecutionState.RUNNING, t1.getExecutionState());
+
assertEquals(ExecutionState.RUNNING, t2.getExecutionState());
+
+
tm.tell(decorator.decorate(new StopTask(eid1)), getRef());
+
+ expectMsgEquals(new
TaskOperationResult(eid1, true));
+
+ Future<Object> response
= Patterns.ask(tm, new TestingTaskManagerMessages.NotifyWhenTaskRemoved(eid1),
+
timeout);
+ Await.ready(response,
d);
+
+
assertEquals(ExecutionState.FINISHED, t1.getExecutionState());
+
+
tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
+ runningTasks =
expectMsgClass(TestingTaskManagerMessages
+
.ResponseRunningTasks.class).asJava();
+
+ assertEquals(1,
runningTasks.size());
+
+
tm.tell(decorator.decorate(new StopTask(eid1)), getRef());
+ expectMsgEquals(new
TaskOperationResult(eid1, false, "No task with that execution ID was " +
+
"found."));
+
+
tm.tell(decorator.decorate(new StopTask(eid2)), getRef());
+ expectMsgEquals(new
TaskOperationResult(eid2, false, "Stopping not supported by this task."));
+
+
assertEquals(ExecutionState.RUNNING, t2.getExecutionState());
+
+
tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
+ runningTasks =
expectMsgClass(TestingTaskManagerMessages
+
.ResponseRunningTasks.class).asJava();
+
+ assertEquals(1,
runningTasks.size());
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+ };
+ }
+ catch(Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
--- End diff --
stack trace
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
