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/02/19 20:53:25 UTC
[1/9] flink git commit: [FLINK-1585] [tests] Fix mini clusters to
respect memory and buffer configurations and improve relative memory
computation.
Repository: flink
Updated Branches:
refs/heads/master 8111177cf -> a51c02f6e
[FLINK-1585] [tests] Fix mini clusters to respect memory and buffer configurations and improve relative memory computation.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a1104491
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a1104491
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a1104491
Branch: refs/heads/master
Commit: a1104491b88ff3293142992900e69253206736ac
Parents: 8111177
Author: Stephan Ewen <se...@apache.org>
Authored: Thu Feb 19 14:34:14 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Feb 19 18:53:42 2015 +0100
----------------------------------------------------------------------
.../minicluster/LocalFlinkMiniCluster.scala | 77 +++++++++++++-------
.../test/util/ForkableFlinkMiniCluster.scala | 9 +++
2 files changed, 60 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a1104491/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
index 88006ac..66ff4f3 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
@@ -154,40 +154,65 @@ class LocalFlinkMiniCluster(userConfiguration: Configuration, singleActorSystem:
}
def setMemory(config: Configuration): Unit = {
- var memorySize: Long = EnvironmentInformation.getSizeOfFreeHeapMemoryWithDefrag
- val bufferMem: Long = ConfigConstants.DEFAULT_TASK_MANAGER_NETWORK_NUM_BUFFERS *
- ConfigConstants.DEFAULT_TASK_MANAGER_NETWORK_BUFFER_SIZE
- val numTaskManager = config.getInteger(ConfigConstants
- .LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 1)
- val taskManagerNumSlots: Int = ConfigConstants.DEFAULT_TASK_MANAGER_NUM_TASK_SLOTS
- memorySize = memorySize - (bufferMem * numTaskManager)
- memorySize = (memorySize * ConfigConstants.DEFAULT_MEMORY_MANAGER_MEMORY_FRACTION).toLong
- memorySize >>>= 20
- memorySize /= numTaskManager
- config.setLong(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, memorySize)
+ // set this only if no memory was preconfigured
+ if (config.getInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, -1) == -1) {
+
+ val bufferMem: Long =
+ config.getLong(ConfigConstants.TASK_MANAGER_NETWORK_NUM_BUFFERS_KEY,
+ ConfigConstants.DEFAULT_TASK_MANAGER_NETWORK_NUM_BUFFERS) *
+ config.getLong(ConfigConstants.TASK_MANAGER_NETWORK_BUFFER_SIZE_KEY,
+ ConfigConstants.DEFAULT_TASK_MANAGER_NETWORK_BUFFER_SIZE)
+
+ val numTaskManager = config.getInteger(
+ ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 1)
+
+ val memoryFraction = config.getFloat(ConfigConstants.TASK_MANAGER_MEMORY_FRACTION_KEY,
+ ConfigConstants.DEFAULT_MEMORY_MANAGER_MEMORY_FRACTION)
+
+ // full memory size
+ var memorySize: Long = EnvironmentInformation.getSizeOfFreeHeapMemoryWithDefrag
+
+ // compute the memory size per task manager. we assume equally much memory for
+ // each TaskManagers and each JobManager
+ memorySize /= numTaskManager + 1 // the +1 is the job manager
+
+ // for each TaskManager, subtract the memory needed for memory buffers
+ memorySize -= bufferMem;
+ memorySize = (memorySize * memoryFraction).toLong
+ memorySize >>>= 20 // bytes to megabytes
+ config.setLong(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, memorySize)
+ }
}
def getDefaultConfig: Configuration = {
- val config: Configuration = new Configuration
+ val config: Configuration = new Configuration()
+
config.setString(ConfigConstants.JOB_MANAGER_IPC_ADDRESS_KEY, HOSTNAME)
- config.setInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY, ConfigConstants
- .DEFAULT_JOB_MANAGER_IPC_PORT)
- config.setInteger(ConfigConstants.TASK_MANAGER_IPC_PORT_KEY, ConfigConstants
- .DEFAULT_TASK_MANAGER_IPC_PORT)
- config.setInteger(ConfigConstants.TASK_MANAGER_DATA_PORT_KEY, ConfigConstants
- .DEFAULT_TASK_MANAGER_DATA_PORT)
- config.setBoolean(ConfigConstants.TASK_MANAGER_MEMORY_LAZY_ALLOCATION_KEY, ConfigConstants
- .DEFAULT_TASK_MANAGER_MEMORY_LAZY_ALLOCATION)
- config.setInteger(ConfigConstants.JOBCLIENT_POLLING_INTERVAL_KEY, ConfigConstants
- .DEFAULT_JOBCLIENT_POLLING_INTERVAL)
- config.setBoolean(ConfigConstants.FILESYSTEM_DEFAULT_OVERWRITE_KEY, ConfigConstants
- .DEFAULT_FILESYSTEM_OVERWRITE)
+
+ config.setInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY,
+ ConfigConstants.DEFAULT_JOB_MANAGER_IPC_PORT)
+
+ config.setInteger(ConfigConstants.TASK_MANAGER_IPC_PORT_KEY,
+ ConfigConstants.DEFAULT_TASK_MANAGER_IPC_PORT)
+
+ config.setInteger(ConfigConstants.TASK_MANAGER_DATA_PORT_KEY,
+ ConfigConstants.DEFAULT_TASK_MANAGER_DATA_PORT)
+
+ config.setBoolean(ConfigConstants.TASK_MANAGER_MEMORY_LAZY_ALLOCATION_KEY,
+ ConfigConstants.DEFAULT_TASK_MANAGER_MEMORY_LAZY_ALLOCATION)
+
+ config.setBoolean(ConfigConstants.FILESYSTEM_DEFAULT_OVERWRITE_KEY,
+ ConfigConstants.DEFAULT_FILESYSTEM_OVERWRITE)
+
config.setBoolean(ConfigConstants.FILESYSTEM_OUTPUT_ALWAYS_CREATE_DIRECTORY_KEY,
ConfigConstants.DEFAULT_FILESYSTEM_ALWAYS_CREATE_DIRECTORY)
+
config.setLong(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, -1)
+
config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 1)
- config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, ConfigConstants
- .DEFAULT_TASK_MANAGER_NUM_TASK_SLOTS)
+
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS,
+ ConfigConstants.DEFAULT_TASK_MANAGER_NUM_TASK_SLOTS)
// Reduce number of threads for local execution
config.setInteger(NettyConfig.NUM_THREADS_CLIENT, 1)
http://git-wip-us.apache.org/repos/asf/flink/blob/a1104491/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
index 23975e2..5b22895 100644
--- a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
+++ b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
@@ -24,6 +24,15 @@ import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster
import org.apache.flink.runtime.taskmanager.TaskManager
import org.apache.flink.runtime.testingUtils.TestingTaskManager
+/**
+ * A forkable mini cluster is a special case of the mini cluster, used for parallel test execution
+ * on build servers. If multiple tests run in parallel, the cluster picks up the fork number and
+ * uses it to avoid port conflicts.
+ *
+ * @param userConfiguration Configuration object with the user provided configuration values
+ * @param singleActorSystem true, if all actors (JobManager and TaskManager) shall be run in the
+ * same [[ActorSystem]], otherwise false.
+ */
class ForkableFlinkMiniCluster(userConfiguration: Configuration, singleActorSystem: Boolean)
extends LocalFlinkMiniCluster(userConfiguration, singleActorSystem) {
[9/9] flink git commit: [tests] Cleanup style and timeouts in
recovery restart tests
Posted by se...@apache.org.
[tests] Cleanup style and timeouts in recovery restart tests
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a51c02f6
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a51c02f6
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a51c02f6
Branch: refs/heads/master
Commit: a51c02f6e8be948d71a00c492808115d622379a7
Parents: 665e601
Author: Stephan Ewen <se...@apache.org>
Authored: Thu Feb 19 20:31:56 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Feb 19 20:31:56 2015 +0100
----------------------------------------------------------------------
.../runtime/jobmanager/RecoveryITCase.scala | 23 +++++++++++----
.../runtime/testingUtils/TestingUtils.scala | 3 +-
flink-tests/src/test/resources/log4j.properties | 27 -----------------
.../jobmanager/JobManagerFailsITCase.scala | 24 ++++++++-------
.../taskmanager/TaskManagerFailsITCase.scala | 31 +++++++++-----------
5 files changed, 46 insertions(+), 62 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a51c02f6/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
index 2c1f82f..e7d1d83 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
@@ -21,13 +21,13 @@ package org.apache.flink.runtime.jobmanager
import akka.actor.Status.Success
import akka.actor.{ActorRef, PoisonPill, ActorSystem}
import akka.testkit.{ImplicitSender, TestKit}
-import org.apache.flink.runtime.jobgraph.{JobStatus, JobGraph, DistributionPattern,
-AbstractJobVertex}
+import org.apache.flink.configuration.{ConfigConstants, Configuration}
+import org.apache.flink.runtime.jobgraph.{JobStatus, JobGraph, DistributionPattern,AbstractJobVertex}
import org.apache.flink.runtime.jobmanager.Tasks.{BlockingOnceReceiver, FailingOnceReceiver}
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup
import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultSuccess, SubmitJob}
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages._
-import org.apache.flink.runtime.testingUtils.TestingUtils
+import org.apache.flink.runtime.testingUtils.{TestingCluster, TestingUtils}
import org.junit.runner.RunWith
import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
import org.scalatest.junit.JUnitRunner
@@ -35,12 +35,23 @@ import org.scalatest.junit.JUnitRunner
@RunWith(classOf[JUnitRunner])
class RecoveryITCase(_system: ActorSystem) extends TestKit(_system) with ImplicitSender with
WordSpecLike with Matchers with BeforeAndAfterAll {
+
def this() = this(ActorSystem("TestingActorSystem", TestingUtils.testConfig))
override def afterAll(): Unit = {
TestKit.shutdownActorSystem(system)
}
+ def startTestClusterWithHeartbeatTimeout(numSlots: Int,
+ numTaskManagers: Int,
+ heartbeatTimeout: String): TestingCluster = {
+ val config = new Configuration()
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlots)
+ config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, numTaskManagers)
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, heartbeatTimeout)
+ new TestingCluster(config)
+ }
+
val NUM_TASKS = 31
"The recovery" must {
@@ -61,7 +72,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
val jobGraph = new JobGraph("Pointwise job", sender, receiver)
jobGraph.setNumberOfExecutionRetries(1)
- val cluster = TestingUtils.startTestingCluster(2 * NUM_TASKS)
+ val cluster = startTestClusterWithHeartbeatTimeout(2 * NUM_TASKS, 1, "2 s")
val jm = cluster.getJobManager
try {
@@ -104,7 +115,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
val jobGraph = new JobGraph("Pointwise job", sender, receiver)
jobGraph.setNumberOfExecutionRetries(1)
- val cluster = TestingUtils.startTestingCluster(NUM_TASKS)
+ val cluster = startTestClusterWithHeartbeatTimeout(NUM_TASKS, 1, "2 s")
val jm = cluster.getJobManager
try {
@@ -147,7 +158,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
val jobGraph = new JobGraph("Pointwise job", sender, receiver)
jobGraph.setNumberOfExecutionRetries(1)
- val cluster = TestingUtils.startTestingCluster(NUM_TASKS, 2)
+ val cluster = startTestClusterWithHeartbeatTimeout(NUM_TASKS, 2, "2 s")
val jm = cluster.getJobManager
http://git-wip-us.apache.org/repos/asf/flink/blob/a51c02f6/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
index 72ed9e7..147cc8a 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
@@ -96,8 +96,7 @@ object TestingUtils {
}
def startTestingCluster(numSlots: Int, numTMs: Int = 1,
- timeout: String = DEFAULT_AKKA_ASK_TIMEOUT):
- TestingCluster = {
+ timeout: String = DEFAULT_AKKA_ASK_TIMEOUT): TestingCluster = {
val config = new Configuration()
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlots)
config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, numTMs)
http://git-wip-us.apache.org/repos/asf/flink/blob/a51c02f6/flink-tests/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/resources/log4j.properties b/flink-tests/src/test/resources/log4j.properties
deleted file mode 100644
index 6bf344a..0000000
--- a/flink-tests/src/test/resources/log4j.properties
+++ /dev/null
@@ -1,27 +0,0 @@
-################################################################################
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-################################################################################
-
-# This file ensures that tests executed from the IDE show log output
-
-log4j.rootLogger=INFO, console
-
-# Log all infos in the given file
-log4j.appender.console=org.apache.log4j.ConsoleAppender
-log4j.appender.console.target = System.err
-log4j.appender.console.layout=org.apache.log4j.PatternLayout
-log4j.appender.console.layout.ConversionPattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/a51c02f6/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 ac1864c..416470f 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
@@ -48,16 +48,9 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
"A TaskManager" should {
"detect a lost connection to the JobManager and try to reconnect to it" in {
- val num_slots = 11
-
- val config = new Configuration()
- config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, num_slots)
- config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 1)
- config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "1000 ms")
- config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "4000 ms")
- config.setDouble(ConfigConstants.AKKA_WATCH_THRESHOLD, 5)
- val cluster = new ForkableFlinkMiniCluster(config, singleActorSystem = false)
+ val num_slots = 11
+ val cluster = startDeathwatchCluster(num_slots, 1)
val tm = cluster.getTaskManagers(0)
val jm = cluster.getJobManager
@@ -100,7 +93,7 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
noOp.setInvokableClass(classOf[NoOpInvokable])
val jobGraph2 = new JobGraph("NoOp Testjob", noOp)
- val cluster = ForkableFlinkMiniCluster.startClusterDeathWatch(num_slots / 2, 2)
+ val cluster = startDeathwatchCluster(num_slots / 2, 2)
var jm = cluster.getJobManager
val tm = cluster.getTaskManagers(0)
@@ -135,4 +128,15 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
}
}
}
+
+ def startDeathwatchCluster(numSlots: Int, numTaskmanagers: Int): ForkableFlinkMiniCluster = {
+ val config = new Configuration()
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlots)
+ config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, numTaskmanagers)
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "1000 ms")
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "4000 ms")
+ config.setDouble(ConfigConstants.AKKA_WATCH_THRESHOLD, 5)
+
+ new ForkableFlinkMiniCluster(config, singleActorSystem = false)
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a51c02f6/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 c81ec88..245bcd9 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
@@ -52,16 +52,9 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
"The JobManager" should {
"detect a failing task manager" in {
- val num_slots = 11
-
- val config = new Configuration()
- config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, num_slots)
- config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 2)
- config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "1000 ms")
- config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "4000 ms")
- config.setDouble(ConfigConstants.AKKA_WATCH_THRESHOLD, 5)
- val cluster = new ForkableFlinkMiniCluster(config, singleActorSystem = false)
+ val num_slots = 11
+ val cluster = startDeathwatchCluster(num_slots, 2)
val taskManagers = cluster.getTaskManagers
val jm = cluster.getJobManager
@@ -89,6 +82,7 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
}
"handle gracefully failing task manager" in {
+
val num_tasks = 31
val sender = new AbstractJobVertex("Sender")
val receiver = new AbstractJobVertex("Receiver")
@@ -190,14 +184,7 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
noOp.setInvokableClass(classOf[NoOpInvokable])
val jobGraph2 = new JobGraph("NoOp Testjob", noOp)
- val config = new Configuration()
- config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, num_slots/2)
- config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 2)
- config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "1000 ms")
- config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "4000 ms")
- config.setDouble(ConfigConstants.AKKA_WATCH_THRESHOLD, 5)
-
- val cluster = new ForkableFlinkMiniCluster(config, singleActorSystem = false)
+ val cluster = startDeathwatchCluster(num_slots/2, 2)
var tm = cluster.getTaskManagers(0)
val jm = cluster.getJobManager
@@ -239,4 +226,14 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
}
}
+ def startDeathwatchCluster(numSlots: Int, numTaskmanagers: Int): ForkableFlinkMiniCluster = {
+ val config = new Configuration()
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlots)
+ config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, numTaskmanagers)
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "1000 ms")
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "4000 ms")
+ config.setDouble(ConfigConstants.AKKA_WATCH_THRESHOLD, 5)
+
+ new ForkableFlinkMiniCluster(config, singleActorSystem = false)
+ }
}
[3/9] flink git commit: [web frontend] Small UI css class fix
Posted by se...@apache.org.
[web frontend] Small UI css class fix
Height of information bullets was not equal due to text break.
This closes #425
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5424be4c
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5424be4c
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5424be4c
Branch: refs/heads/master
Commit: 5424be4c69dadf91baa012906b2c01eef42ea397
Parents: 7d80711
Author: Carsten Brandt <ma...@cebe.cc>
Authored: Thu Feb 19 17:57:21 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Feb 19 18:54:14 2015 +0100
----------------------------------------------------------------------
.../resources/web-docs-infoserver/index.html | 20 ++++++++++----------
1 file changed, 10 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5424be4c/flink-runtime/src/main/resources/web-docs-infoserver/index.html
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/resources/web-docs-infoserver/index.html b/flink-runtime/src/main/resources/web-docs-infoserver/index.html
index 43f4289..022e6a8 100644
--- a/flink-runtime/src/main/resources/web-docs-infoserver/index.html
+++ b/flink-runtime/src/main/resources/web-docs-infoserver/index.html
@@ -126,10 +126,10 @@ under the License.
<div class="panel panel-info">
<div class="panel-heading">
<div class="row">
- <div class="col-xs-6">
+ <div class="col-xs-4">
<i class="fa fa-info fa-5x"></i>
</div>
- <div class="col-xs-6 text-right">
+ <div class="col-xs-8 text-right">
<p class="announcement-heading"><span id="stat-taskmanagers" class="stats"></span></p>
<p class="announcement-text">Task Manager</p>
</div>
@@ -141,10 +141,10 @@ under the License.
<div class="panel panel-primary">
<div class="panel-heading">
<div class="row">
- <div class="col-xs-6">
+ <div class="col-xs-4">
<i class="fa fa-list-ol fa-5x"></i>
</div>
- <div class="col-xs-6 text-right">
+ <div class="col-xs-8 text-right">
<p class="announcement-heading"><span id="stat-slots" class="stats"></span></p>
<p class="announcement-text">Processing Slots</p>
</div>
@@ -156,10 +156,10 @@ under the License.
<div class="panel panel-success">
<div class="panel-heading">
<div class="row">
- <div class="col-xs-6">
+ <div class="col-xs-4">
<i class="fa fa-check fa-5x"></i>
</div>
- <div class="col-xs-6 text-right">
+ <div class="col-xs-8 text-right">
<p class="announcement-heading"><span id="jobs-finished" class="stats"></span></p>
<p class="announcement-text">Jobs Finished</p>
</div>
@@ -171,10 +171,10 @@ under the License.
<div class="panel panel-warning">
<div class="panel-heading">
<div class="row">
- <div class="col-xs-6">
+ <div class="col-xs-4">
<i class="fa fa-eraser fa-5x"></i>
</div>
- <div class="col-xs-6 text-right">
+ <div class="col-xs-8 text-right">
<p class="announcement-heading"><span id="jobs-canceled" class="stats"></span></p>
<p class="announcement-text">Jobs Canceled</p>
</div>
@@ -186,10 +186,10 @@ under the License.
<div class="panel panel-danger">
<div class="panel-heading">
<div class="row">
- <div class="col-xs-6">
+ <div class="col-xs-4">
<i class="fa fa-flash fa-5x"></i>
</div>
- <div class="col-xs-6 text-right">
+ <div class="col-xs-8 text-right">
<p class="announcement-heading"><span id="jobs-failed" class="stats"></span></p>
<p class="announcement-text">Jobs Failed</p>
</div>
[2/9] flink git commit: [FLINK-1584] [runtime][tests] Fixes
TaskManagerFailsITCase by replacing the TestingCluster with a
ForkableFlinkMiniCluster
Posted by se...@apache.org.
[FLINK-1584] [runtime][tests] Fixes TaskManagerFailsITCase by replacing the TestingCluster with a ForkableFlinkMiniCluster
This closes #419
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/52b40deb
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/52b40deb
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/52b40deb
Branch: refs/heads/master
Commit: 52b40debad6293cc0591eabb847eaac322d174aa
Parents: 5424be4
Author: Till Rohrmann <tr...@apache.org>
Authored: Thu Feb 19 11:04:56 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Feb 19 18:54:14 2015 +0100
----------------------------------------------------------------------
.../jobmanager/JobManagerFailsITCase.scala | 72 ---------
.../jobmanager/TaskManagerFailsITCase.scala | 147 ------------------
.../runtime/testingUtils/TestingCluster.scala | 11 +-
.../runtime/testingUtils/TestingUtils.scala | 14 --
.../test/util/ForkableFlinkMiniCluster.scala | 56 ++++++-
.../jobmanager/JobManagerFailsITCase.scala | 72 +++++++++
.../taskmanager/TaskManagerFailsITCase.scala | 148 +++++++++++++++++++
7 files changed, 277 insertions(+), 243 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/52b40deb/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerFailsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerFailsITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerFailsITCase.scala
deleted file mode 100644
index 88c604b..0000000
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerFailsITCase.scala
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.jobmanager
-
-import akka.actor.{PoisonPill, ActorSystem}
-import akka.testkit.{ImplicitSender, TestKit}
-import org.apache.flink.runtime.messages.JobManagerMessages.RequestNumberRegisteredTaskManager
-import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages.{JobManagerTerminated,
-NotifyWhenJobManagerTerminated}
-import org.apache.flink.runtime.testingUtils.TestingUtils
-import org.junit.runner.RunWith
-import org.scalatest.{WordSpecLike, Matchers, BeforeAndAfterAll}
-import org.scalatest.junit.JUnitRunner
-
-@RunWith(classOf[JUnitRunner])
-class JobManagerFailsITCase(_system: ActorSystem) extends TestKit(_system) with ImplicitSender
-with WordSpecLike with Matchers with BeforeAndAfterAll {
-
- def this() = this(ActorSystem("TestingActorSystem", TestingUtils.testConfig))
-
- override def afterAll(): Unit = {
- TestKit.shutdownActorSystem(system)
- }
-
- "The TaskManager" should {
- "detect a lost connection to the JobManager and try to reconnect to it" in {
- val num_slots = 11
-
- val cluster = TestingUtils.startTestingClusterDeathWatch(num_slots, 1)
-
- val tm = cluster.getTaskManagers(0)
- val jm = cluster.getJobManager
-
- try{
- jm ! RequestNumberRegisteredTaskManager
- expectMsg(1)
-
- tm ! NotifyWhenJobManagerTerminated(jm)
-
- jm ! PoisonPill
-
- expectMsgClass(classOf[JobManagerTerminated])
-
- cluster.restartJobManager()
-
- cluster.waitForTaskManagersToBeRegistered()
-
- cluster.getJobManager ! RequestNumberRegisteredTaskManager
-
- expectMsg(1)
- }finally{
- cluster.stop()
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/52b40deb/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/TaskManagerFailsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/TaskManagerFailsITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/TaskManagerFailsITCase.scala
deleted file mode 100644
index 8d4a9ba..0000000
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/TaskManagerFailsITCase.scala
+++ /dev/null
@@ -1,147 +0,0 @@
-/*
-* Licensed to the Apache Software Foundation (ASF) under one
-* or more contributor license agreements. See the NOTICE file
-* distributed with this work for additional information
-* regarding copyright ownership. The ASF licenses this file
-* to you under the Apache License, Version 2.0 (the
-* "License"); you may not use this file except in compliance
-* with the License. You may obtain a copy of the License at
-*
-* http://www.apache.org/licenses/LICENSE-2.0
-*
-* Unless required by applicable law or agreed to in writing, software
-* distributed under the License is distributed on an "AS IS" BASIS,
-* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-* See the License for the specific language governing permissions and
-* limitations under the License.
-*/
-
-package org.apache.flink.runtime.jobmanager
-
-import akka.actor.{Kill, ActorSystem, PoisonPill}
-import akka.testkit.{ImplicitSender, TestKit}
-import org.apache.flink.runtime.jobgraph.{AbstractJobVertex, DistributionPattern, JobGraph}
-import org.apache.flink.runtime.jobmanager.Tasks.{BlockingReceiver, Sender}
-import org.apache.flink.runtime.messages.JobManagerMessages.{RequestNumberRegisteredTaskManager,
-JobResultFailed, SubmissionSuccess, SubmitJob}
-import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages._
-import org.apache.flink.runtime.testingUtils.TestingUtils
-import org.junit.runner.RunWith
-import org.scalatest.junit.JUnitRunner
-import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
-
-@RunWith(classOf[JUnitRunner])
-class TaskManagerFailsITCase(_system: ActorSystem) extends TestKit(_system) with ImplicitSender
-with WordSpecLike with Matchers with BeforeAndAfterAll {
-
- def this() = this(ActorSystem("TestingActorSystem", TestingUtils.testConfig))
-
- override def afterAll(): Unit = {
- TestKit.shutdownActorSystem(system)
- }
-
- "The JobManager" should {
-
- "detect a failing task manager" in {
- val num_slots = 11
-
- val cluster = TestingUtils.startTestingClusterDeathWatch(num_slots, 2)
-
- val taskManagers = cluster.getTaskManagers
- val jm = cluster.getJobManager
-
- try{
- within(TestingUtils.TESTING_DURATION){
- jm ! RequestNumberRegisteredTaskManager
- expectMsg(2)
-
- jm ! NotifyWhenTaskManagerTerminated(taskManagers(0))
-
- taskManagers(0) ! PoisonPill
-
- val TaskManagerTerminated(tm) = expectMsgClass(classOf[TaskManagerTerminated])
-
- jm ! RequestNumberRegisteredTaskManager
- expectMsg(1)
- }
- }finally{
- cluster.stop()
- }
-
- }
-
- "handle gracefully failing task manager" in {
- val num_tasks = 31
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
- sender.setInvokableClass(classOf[Sender])
- receiver.setInvokableClass(classOf[BlockingReceiver])
- sender.setParallelism(num_tasks)
- receiver.setParallelism(num_tasks)
- receiver.connectNewDataSetAsInput(sender, DistributionPattern.POINTWISE)
-
- val jobGraph = new JobGraph("Pointwise Job", sender, receiver)
- val jobID = jobGraph.getJobID
-
- val cluster = TestingUtils.startTestingCluster(num_tasks, 2)
-
- val jm = cluster.getJobManager
-
- try {
- within(TestingUtils.TESTING_DURATION) {
- jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
-
- jm ! WaitForAllVerticesToBeRunningOrFinished(jobID)
-
- expectMsg(AllVerticesRunning(jobID))
-
- jm ! RequestWorkingTaskManager(jobID)
-
- val tm = expectMsgType[WorkingTaskManager].taskManager
- // kill one task manager
- tm ! PoisonPill
- expectMsgType[JobResultFailed]
- }
- }finally{
- cluster.stop()
- }
- }
-
- "handle hard failing task manager" in {
- val num_tasks = 31
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
- sender.setInvokableClass(classOf[Sender])
- receiver.setInvokableClass(classOf[BlockingReceiver])
- sender.setParallelism(num_tasks)
- receiver.setParallelism(num_tasks)
- receiver.connectNewDataSetAsInput(sender, DistributionPattern.POINTWISE)
-
- val jobGraph = new JobGraph("Pointwise Job", sender, receiver)
- val jobID = jobGraph.getJobID
-
- val cluster = TestingUtils.startTestingCluster(num_tasks, 2)
-
- val taskManagers = cluster.getTaskManagers
- val jm = cluster.getJobManager
-
- try {
- within(TestingUtils.TESTING_DURATION) {
- jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
-
- jm ! WaitForAllVerticesToBeRunningOrFinished(jobID)
- expectMsg(AllVerticesRunning(jobID))
-
- // kill one task manager
- taskManagers(0) ! Kill
- expectMsgType[JobResultFailed]
- }
- }finally{
- cluster.stop()
- }
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/52b40deb/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
index 9d132b6..87fca99 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
@@ -48,9 +48,9 @@ FlinkMiniCluster(userConfiguration, singleActorSystem) {
override def startJobManager(actorSystem: ActorSystem): ActorRef = {
- val (instanceManager, scheduler, libraryCacheManager, _, accumulatorManager, _ ,
- executionRetries, delayBetweenRetries,
- timeout, archiveCount) = JobManager.createJobManagerComponents(configuration)
+ val (instanceManager, scheduler, libraryCacheManager, _, accumulatorManager, _,
+ executionRetries, delayBetweenRetries,
+ timeout, archiveCount) = JobManager.createJobManagerComponents(configuration)
val testArchiveProps = Props(new MemoryArchivist(archiveCount) with TestingMemoryArchivist)
val archive = actorSystem.actorOf(testArchiveProps, JobManager.ARCHIVE_NAME)
@@ -71,9 +71,4 @@ FlinkMiniCluster(userConfiguration, singleActorSystem) {
networkConnectionConfig) with TestingTaskManager), TaskManager.TASK_MANAGER_NAME + "_" +
(index + 1))
}
-
- def restartJobManager(): Unit = {
- jobManagerActorSystem.stop(jobManagerActor)
- jobManagerActor = startJobManager(jobManagerActorSystem)
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/52b40deb/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
index c26b255..72ed9e7 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
@@ -105,20 +105,6 @@ object TestingUtils {
new TestingCluster(config)
}
- def startTestingClusterDeathWatch(numSlots: Int, numTMs: Int,
- timeout: String = DEFAULT_AKKA_ASK_TIMEOUT):
- TestingCluster = {
- val config = new Configuration()
- config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlots)
- config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, numTMs)
- config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, timeout)
- config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "200 ms")
- config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "50 ms")
- config.setDouble(ConfigConstants.AKKA_WATCH_THRESHOLD, 1)
-
- new TestingCluster(config, singleActorSystem = false)
- }
-
def setGlobalExecutionContext(): Unit = {
AkkaUtils.globalExecutionContext = ExecutionContext.global
}
http://git-wip-us.apache.org/repos/asf/flink/blob/52b40deb/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
index 5b22895..38095b9 100644
--- a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
+++ b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
@@ -18,11 +18,13 @@
package org.apache.flink.test.util
-import akka.actor.{ActorRef, ActorSystem, Props}
+import akka.actor.{Props, ActorRef, ActorSystem}
import org.apache.flink.configuration.{ConfigConstants, Configuration}
+import org.apache.flink.runtime.jobmanager.{MemoryArchivist, JobManager}
import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster
import org.apache.flink.runtime.taskmanager.TaskManager
-import org.apache.flink.runtime.testingUtils.TestingTaskManager
+import org.apache.flink.runtime.testingUtils.{TestingJobManager, TestingMemoryArchivist,
+TestingTaskManager}
/**
* A forkable mini cluster is a special case of the mini cluster, used for parallel test execution
@@ -62,6 +64,22 @@ class ForkableFlinkMiniCluster(userConfiguration: Configuration, singleActorSyst
super.generateConfiguration(config)
}
+ override def startJobManager(actorSystem: ActorSystem): ActorRef = {
+
+ val (instanceManager, scheduler, libraryCacheManager, _, accumulatorManager, _,
+ executionRetries, delayBetweenRetries,
+ timeout, archiveCount) = JobManager.createJobManagerComponents(configuration)
+
+ val testArchiveProps = Props(new MemoryArchivist(archiveCount) with TestingMemoryArchivist)
+ val archive = actorSystem.actorOf(testArchiveProps, JobManager.ARCHIVE_NAME)
+
+ val jobManagerProps = Props(new JobManager(configuration, instanceManager, scheduler,
+ libraryCacheManager, archive, accumulatorManager, None, executionRetries,
+ delayBetweenRetries, timeout) with TestingJobManager)
+
+ actorSystem.actorOf(jobManagerProps, JobManager.JOB_MANAGER_NAME)
+ }
+
override def startTaskManager(index: Int)(implicit system: ActorSystem): ActorRef = {
val config = configuration.clone()
@@ -86,4 +104,38 @@ class ForkableFlinkMiniCluster(userConfiguration: Configuration, singleActorSyst
system.actorOf(Props(new TaskManager(connectionInfo, jobManagerAkkaURL, taskManagerConfig,
networkConnectionConfig) with TestingTaskManager), TaskManager.TASK_MANAGER_NAME + index)
}
+
+ def restartJobManager(): Unit = {
+ jobManagerActorSystem.stop(jobManagerActor)
+ jobManagerActor = startJobManager(jobManagerActorSystem)
+ }
+}
+
+object ForkableFlinkMiniCluster {
+
+ import org.apache.flink.runtime.testingUtils.TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT
+
+ def startClusterDeathWatch(numSlots: Int, numTaskManagers: Int,
+ timeout: String = DEFAULT_AKKA_ASK_TIMEOUT):
+ ForkableFlinkMiniCluster = {
+ val config = new Configuration()
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlots)
+ config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, numTaskManagers)
+ config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, timeout)
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "1000 ms")
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "4000 ms")
+ config.setDouble(ConfigConstants.AKKA_WATCH_THRESHOLD, 5)
+
+ new ForkableFlinkMiniCluster(config, singleActorSystem = false)
+ }
+
+ def startCluster(numSlots: Int, numTaskManagers: Int, timeout: String = DEFAULT_AKKA_ASK_TIMEOUT):
+ ForkableFlinkMiniCluster = {
+ val config = new Configuration()
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlots)
+ config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, numTaskManagers)
+ config.setInteger(ConfigConstants.JOB_MANAGER_DEAD_TASKMANAGER_TIMEOUT_KEY, 1000)
+ config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, timeout)
+ new ForkableFlinkMiniCluster(config)
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/52b40deb/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
new file mode 100644
index 0000000..1e97f7b
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerFailsITCase.scala
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.scala.runtime.jobmanager
+
+import akka.actor.{ActorSystem, PoisonPill}
+import akka.testkit.{ImplicitSender, TestKit}
+import org.apache.flink.runtime.akka.AkkaUtils
+import org.apache.flink.runtime.messages.JobManagerMessages.RequestNumberRegisteredTaskManager
+import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages.{JobManagerTerminated, NotifyWhenJobManagerTerminated}
+import org.apache.flink.test.util.ForkableFlinkMiniCluster
+import org.junit.runner.RunWith
+import org.scalatest.junit.JUnitRunner
+import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
+
+@RunWith(classOf[JUnitRunner])
+class JobManagerFailsITCase(_system: ActorSystem) extends TestKit(_system) with ImplicitSender
+with WordSpecLike with Matchers with BeforeAndAfterAll {
+
+ def this() = this(ActorSystem("TestingActorSystem", AkkaUtils.getDefaultAkkaConfig))
+
+ override def afterAll(): Unit = {
+ TestKit.shutdownActorSystem(system)
+ }
+
+ "The TaskManager" should {
+ "detect a lost connection to the JobManager and try to reconnect to it" in {
+ val num_slots = 11
+
+ val cluster = ForkableFlinkMiniCluster.startClusterDeathWatch(num_slots, 1)
+
+ val tm = cluster.getTaskManagers(0)
+ val jm = cluster.getJobManager
+
+ try{
+ jm ! RequestNumberRegisteredTaskManager
+ expectMsg(1)
+
+ tm ! NotifyWhenJobManagerTerminated(jm)
+
+ jm ! PoisonPill
+
+ expectMsgClass(classOf[JobManagerTerminated])
+
+ cluster.restartJobManager()
+
+ cluster.waitForTaskManagersToBeRegistered()
+
+ cluster.getJobManager ! RequestNumberRegisteredTaskManager
+
+ expectMsg(1)
+ }finally{
+ cluster.stop()
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/52b40deb/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
new file mode 100644
index 0000000..5ae464f
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala
@@ -0,0 +1,148 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.flink.api.scala.runtime.taskmanager
+
+import akka.actor.{ActorSystem, Kill, PoisonPill}
+import akka.testkit.{ImplicitSender, TestKit}
+import org.apache.flink.runtime.akka.AkkaUtils
+import org.apache.flink.runtime.jobgraph.{AbstractJobVertex, DistributionPattern, JobGraph}
+import org.apache.flink.runtime.jobmanager.Tasks.{BlockingReceiver, Sender}
+import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultFailed, RequestNumberRegisteredTaskManager, SubmissionSuccess, SubmitJob}
+import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages._
+import org.apache.flink.runtime.testingUtils.TestingUtils
+import org.apache.flink.test.util.ForkableFlinkMiniCluster
+import org.junit.runner.RunWith
+import org.scalatest.junit.JUnitRunner
+import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
+
+@RunWith(classOf[JUnitRunner])
+class TaskManagerFailsITCase(_system: ActorSystem) extends TestKit(_system) with ImplicitSender
+with WordSpecLike with Matchers with BeforeAndAfterAll {
+
+ def this() = this(ActorSystem("TestingActorSystem", AkkaUtils.getDefaultAkkaConfig))
+
+ override def afterAll(): Unit = {
+ TestKit.shutdownActorSystem(system)
+ }
+
+ "The JobManager" should {
+
+ "detect a failing task manager" in {
+ val num_slots = 11
+
+ val cluster = ForkableFlinkMiniCluster.startClusterDeathWatch(num_slots, 2)
+
+ val taskManagers = cluster.getTaskManagers
+ val jm = cluster.getJobManager
+
+ try{
+ within(TestingUtils.TESTING_DURATION){
+ jm ! RequestNumberRegisteredTaskManager
+ expectMsg(2)
+
+ jm ! NotifyWhenTaskManagerTerminated(taskManagers(0))
+
+ taskManagers(0) ! PoisonPill
+
+ val TaskManagerTerminated(tm) = expectMsgClass(classOf[TaskManagerTerminated])
+
+ jm ! RequestNumberRegisteredTaskManager
+ expectMsg(1)
+ }
+ }finally{
+ cluster.stop()
+ }
+
+ }
+
+ "handle gracefully failing task manager" in {
+ val num_tasks = 31
+ val sender = new AbstractJobVertex("Sender")
+ val receiver = new AbstractJobVertex("Receiver")
+ sender.setInvokableClass(classOf[Sender])
+ receiver.setInvokableClass(classOf[BlockingReceiver])
+ sender.setParallelism(num_tasks)
+ receiver.setParallelism(num_tasks)
+ receiver.connectNewDataSetAsInput(sender, DistributionPattern.POINTWISE)
+
+ val jobGraph = new JobGraph("Pointwise Job", sender, receiver)
+ val jobID = jobGraph.getJobID
+
+ val cluster = ForkableFlinkMiniCluster.startCluster(num_tasks, 2)
+
+ val jm = cluster.getJobManager
+
+ try {
+ within(TestingUtils.TESTING_DURATION) {
+ jm ! SubmitJob(jobGraph)
+ expectMsg(SubmissionSuccess(jobGraph.getJobID))
+
+ jm ! WaitForAllVerticesToBeRunningOrFinished(jobID)
+
+ expectMsg(AllVerticesRunning(jobID))
+
+ jm ! RequestWorkingTaskManager(jobID)
+
+ val tm = expectMsgType[WorkingTaskManager].taskManager
+ // kill one task manager
+ tm ! PoisonPill
+ expectMsgType[JobResultFailed]
+ }
+ }finally{
+ cluster.stop()
+ }
+ }
+
+ "handle hard failing task manager" in {
+ val num_tasks = 31
+ val sender = new AbstractJobVertex("Sender")
+ val receiver = new AbstractJobVertex("Receiver")
+ sender.setInvokableClass(classOf[Sender])
+ receiver.setInvokableClass(classOf[BlockingReceiver])
+ sender.setParallelism(num_tasks)
+ receiver.setParallelism(num_tasks)
+ receiver.connectNewDataSetAsInput(sender, DistributionPattern.POINTWISE)
+
+ val jobGraph = new JobGraph("Pointwise Job", sender, receiver)
+ val jobID = jobGraph.getJobID
+
+ val cluster = ForkableFlinkMiniCluster.startCluster(num_tasks, 2)
+
+ val taskManagers = cluster.getTaskManagers
+ val jm = cluster.getJobManager
+
+ try {
+ within(TestingUtils.TESTING_DURATION) {
+ jm ! SubmitJob(jobGraph)
+ expectMsg(SubmissionSuccess(jobGraph.getJobID))
+
+ jm ! WaitForAllVerticesToBeRunningOrFinished(jobID)
+ expectMsg(AllVerticesRunning(jobID))
+
+ // kill one task manager
+ taskManagers(0) ! Kill
+ expectMsgType[JobResultFailed]
+ }
+ }finally{
+ cluster.stop()
+ }
+ }
+ }
+
+}
[7/9] flink git commit: [Flink-1484] [runtime] JobManager sends an
explicit disconnect message to the registered TaskManagers in case of a
graceful shutdown.
Posted by se...@apache.org.
[Flink-1484] [runtime] JobManager sends an explicit disconnect message to the registered TaskManagers in case of a graceful shutdown.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5ac1911a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5ac1911a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5ac1911a
Branch: refs/heads/master
Commit: 5ac1911a831c310e9f6634099e9767a2f0bce306
Parents: 4ff91cd
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue Feb 17 15:54:42 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Feb 19 19:30:27 2015 +0100
----------------------------------------------------------------------
.../runtime/execution/RuntimeEnvironment.java | 10 +-
.../flink/runtime/jobmanager/JobManager.scala | 12 +-
.../runtime/messages/JobManagerMessages.scala | 330 +++++++++++++++++++
.../runtime/messages/JobmanagerMessages.scala | 330 -------------------
.../flink/runtime/messages/Messages.scala | 10 +
.../flink/runtime/taskmanager/TaskManager.scala | 228 +++++++------
.../runtime/testingUtils/TestingCluster.scala | 10 +
.../testingUtils/TestingJobManager.scala | 54 +--
.../runtime/testingUtils/TestingMessages.scala | 24 ++
.../testingUtils/TestingTaskManager.scala | 33 +-
.../jobmanager/JobManagerFailsITCase.scala | 62 +++-
11 files changed, 628 insertions(+), 475 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5ac1911a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/RuntimeEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/RuntimeEnvironment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/RuntimeEnvironment.java
index f78ea92..85c0e8e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/RuntimeEnvironment.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/RuntimeEnvironment.java
@@ -227,7 +227,15 @@ public class RuntimeEnvironment implements Environment, Runnable {
}
}
catch (Throwable t) {
- LOG.error("Error during running invokable: " + t.getMessage(), t);
+ String msg;
+
+ if(t.getMessage() != null){
+ msg = "Error during running invokable: " + t.getMessage();
+ } else {
+ msg = "Error during running invokable.";
+ }
+
+ LOG.error(msg, t);
if (!owner.isCanceledOrFailed()) {
// Perform clean up when the task failed and has been not canceled by the user
http://git-wip-us.apache.org/repos/asf/flink/blob/5ac1911a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
index 1399bce..32a4e37 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
@@ -31,7 +31,7 @@ import org.apache.flink.runtime.executiongraph.{ExecutionJobVertex, ExecutionGra
import org.apache.flink.runtime.jobmanager.web.WebInfoServer
import org.apache.flink.runtime.messages.ArchiveMessages.ArchiveExecutionGraph
import org.apache.flink.runtime.messages.ExecutionGraphMessages.JobStatusChanged
-import org.apache.flink.runtime.messages.Messages.Acknowledge
+import org.apache.flink.runtime.messages.Messages.{Disconnect, Acknowledge}
import org.apache.flink.runtime.security.SecurityUtils
import org.apache.flink.runtime.security.SecurityUtils.FlinkSecuredRunner
import org.apache.flink.runtime.taskmanager.TaskManager
@@ -112,6 +112,11 @@ class JobManager(val configuration: Configuration,
override def postStop(): Unit = {
log.info(s"Stopping job manager ${self.path}.")
+ // disconnect the registered task managers
+ instanceManager.getAllRegisteredInstances.asScala.foreach {
+ _.getTaskManager ! Disconnect("JobManager is shutting down")
+ }
+
archive ! PoisonPill
profiler.foreach( ref => ref ! PoisonPill )
@@ -159,7 +164,6 @@ class JobManager(val configuration: Configuration,
profiler)
}
-
case RequestNumberRegisteredTaskManager =>
sender ! instanceManager.getNumberOfRegisteredTaskManagers
@@ -194,8 +198,10 @@ class JobManager(val configuration: Configuration,
case Some((executionGraph, _)) =>
val originalSender = sender
Future {
- originalSender ! executionGraph.updateState(taskExecutionState)
+ val result = executionGraph.updateState(taskExecutionState)
+ originalSender ! result
}
+ sender ! true
case None => log.error("Cannot find execution graph for ID {} to change state to {}.",
taskExecutionState.getJobID, taskExecutionState.getExecutionState)
sender ! false
http://git-wip-us.apache.org/repos/asf/flink/blob/5ac1911a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobManagerMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobManagerMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobManagerMessages.scala
new file mode 100644
index 0000000..8aef552
--- /dev/null
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobManagerMessages.scala
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.messages
+
+import org.apache.flink.runtime.accumulators.AccumulatorEvent
+import org.apache.flink.runtime.executiongraph.{ExecutionAttemptID, ExecutionGraph}
+import org.apache.flink.runtime.instance.{InstanceID, Instance}
+import org.apache.flink.runtime.jobgraph.{JobGraph, JobID, JobStatus, JobVertexID}
+import org.apache.flink.runtime.taskmanager.TaskExecutionState
+
+/**
+ * The job manager specific messages
+ */
+object JobManagerMessages {
+ /**
+ * Submits a job to the job manager. If [[registerForEvents]] is true,
+ * then the sender will be registered as listener for the state change messages. If [[detached]]
+ * is set to true, then the sender will detach from the job execution. Consequently,
+ * he will not receive the job execution result [[JobResult]]. The submission result will be sent
+ * back to the
+ * sender as a [[SubmissionResponse]] message.
+ *
+ * @param jobGraph
+ * @param registerForEvents if true, then register for state change events
+ * @param detached if true, then detach from the job execution
+ */
+ case class SubmitJob(jobGraph: JobGraph, registerForEvents: Boolean = false,
+ detached: Boolean = false)
+
+ /**
+ * Cancels a job with the given [[jobID]] at the JobManager. The result of the cancellation is
+ * sent back to the sender as a [[CancellationResponse]] message.
+ *
+ * @param jobID
+ */
+ case class CancelJob(jobID: JobID)
+
+ /**
+ * Denotes a state change of a task at the JobManager. The update success is acknowledged by a
+ * boolean value which is sent back to the sender.
+ *
+ * @param taskExecutionState
+ */
+ case class UpdateTaskExecutionState(taskExecutionState: TaskExecutionState)
+
+ /**
+ * Requesting next input split for the
+ * [[org.apache.flink.runtime.executiongraph.ExecutionJobVertex]]
+ * of the job specified by [[jobID]]. The next input split is sent back to the sender as a
+ * [[org.apache.flink.runtime.messages.TaskManagerMessages.NextInputSplit]] message.
+ *
+ * @param jobID
+ * @param vertexID
+ */
+ case class RequestNextInputSplit(jobID: JobID, vertexID: JobVertexID, executionAttempt:
+ ExecutionAttemptID)
+
+ /**
+ * Notifies the [[org.apache.flink.runtime.jobmanager.JobManager]] about available data for a
+ * produced partition.
+ * <p>
+ * There is a call to this method for each
+ * [[org.apache.flink.runtime.executiongraph.ExecutionVertex]] instance once per produced
+ * [[org.apache.flink.runtime.io.network.partition.IntermediateResultPartition]] instance,
+ * either when first producing data (for pipelined executions) or when all data has been produced
+ * (for staged executions).
+ * <p>
+ * The [[org.apache.flink.runtime.jobmanager.JobManager]] then can decide when to schedule the
+ * partition consumers of the given session.
+ *
+ * @see [[org.apache.flink.runtime.io.network.partition.IntermediateResultPartition]]
+ */
+ case class ScheduleOrUpdateConsumers(jobId: JobID,
+ executionId: ExecutionAttemptID,
+ partitionIndex: Int)
+
+ case class ConsumerNotificationResult(success: Boolean, error: Option[Throwable] = None)
+
+ /**
+ * Reports the accumulator results of the individual tasks to the job manager.
+ *
+ * @param accumulatorEvent
+ */
+ case class ReportAccumulatorResult(accumulatorEvent: AccumulatorEvent)
+
+ /**
+ * Requests the accumulator results of the job identified by [[jobID]] from the job manager.
+ * The result is sent back to the sender as a [[AccumulatorResultsResponse]] message.
+ *
+ * @param jobID
+ */
+ case class RequestAccumulatorResults(jobID: JobID)
+
+ sealed trait AccumulatorResultsResponse{
+ val jobID: JobID
+ }
+
+ /**
+ * Contains the retrieved accumulator results from the job manager. This response is triggered
+ * by [[RequestAccumulatorResults]].
+ *
+ * @param jobID
+ * @param results
+ */
+ case class AccumulatorResultsFound(jobID: JobID, results: Map[String,
+ Object]) extends AccumulatorResultsResponse{
+ def asJavaMap: java.util.Map[String, Object] = {
+ import scala.collection.JavaConverters._
+ results.asJava
+ }
+ }
+
+ /**
+ * Denotes that no accumulator results for [[jobID]] could be found at the job manager.
+ * @param jobID
+ */
+ case class AccumulatorResultsNotFound(jobID: JobID) extends AccumulatorResultsResponse
+
+ /**
+ * Requests the current [[JobStatus]] of the job identified by [[jobID]]. This message triggers
+ * as response a [[JobStatusResponse]] message.
+ *
+ * @param jobID
+ */
+ case class RequestJobStatus(jobID: JobID)
+
+ sealed trait JobStatusResponse {
+ def jobID: JobID
+ }
+
+ /**
+ * Denotes the current [[JobStatus]] of the job with [[jobID]].
+ *
+ * @param jobID
+ * @param status
+ */
+ case class CurrentJobStatus(jobID: JobID, status: JobStatus) extends JobStatusResponse
+
+ /**
+ * Requests the number of currently registered task manager at the job manager. The result is
+ * sent back to the sender as an [[Int]].
+ */
+ case object RequestNumberRegisteredTaskManager
+
+ /**
+ * Requests the maximum number of slots available to the job manager. The result is sent back
+ * to the sender as an [[Int]].
+ */
+ case object RequestTotalNumberOfSlots
+
+ /**
+ * Requests the port of the blob manager from the job manager. The result is sent back to the
+ * sender as an [[Int]].
+ */
+ case object RequestBlobManagerPort
+
+ /**
+ * Requests the final job status of the job with [[jobID]]. If the job has not been terminated
+ * then the result is sent back upon termination of the job. The result is a
+ * [[JobStatusResponse]] message.
+ *
+ * @param jobID
+ */
+ case class RequestFinalJobStatus(jobID: JobID)
+
+ /**
+ * Denotes a successful job execution.
+ *
+ * @param jobID
+ * @param runtime
+ * @param accumulatorResults
+ */
+ case class JobResultSuccess(jobID: JobID, runtime: Long, accumulatorResults: java.util.Map[String,
+ AnyRef]) {}
+
+ sealed trait CancellationResponse{
+ def jobID: JobID
+ }
+
+ /**
+ * Denotes a successful job cancellation
+ * @param jobID
+ */
+ case class CancellationSuccess(jobID: JobID) extends CancellationResponse
+
+ /**
+ * Denotes a failed job cancellation
+ * @param jobID
+ * @param cause
+ */
+ case class CancellationFailure(jobID: JobID, cause: Throwable) extends CancellationResponse
+
+ /**
+ * Requests all currently running jobs from the job manager. This message triggers a
+ * [[RunningJobs]] response.
+ */
+ case object RequestRunningJobs
+
+ /**
+ * This message is the response to the [[RequestRunningJobs]] message. It contains all
+ * execution graphs of the currently running jobs.
+ *
+ * @param runningJobs
+ */
+ case class RunningJobs(runningJobs: Iterable[ExecutionGraph]) {
+ def this() = this(Seq())
+ def asJavaIterable: java.lang.Iterable[ExecutionGraph] = {
+ import scala.collection.JavaConverters._
+ runningJobs.asJava
+ }
+ }
+
+ /**
+ * Requests the execution graph of a specific job identified by [[jobID]]. The result is sent
+ * back to the sender as a [[JobResponse]].
+ * @param jobID
+ */
+ case class RequestJob(jobID: JobID)
+
+ sealed trait JobResponse{
+ def jobID: JobID
+ }
+
+ /**
+ * Contains the [[executionGraph]] of a job with [[jobID]]. This is the response to
+ * [[RequestJob]] if the job runs or is archived.
+ *
+ * @param jobID
+ * @param executionGraph
+ */
+ case class JobFound(jobID: JobID, executionGraph: ExecutionGraph) extends JobResponse
+
+ /**
+ * Denotes that there is no job with [[jobID]] retrievable. This message can be the response of
+ * [[RequestJob]] or [[RequestJobStatus]].
+ *
+ * @param jobID
+ */
+ case class JobNotFound(jobID: JobID) extends JobResponse with JobStatusResponse
+
+ /**
+ * Requests the instances of all registered task managers.
+ */
+ case object RequestRegisteredTaskManagers
+
+ /**
+ * Contains the [[Instance]] objects of all registered task managers. It is the response to the
+ * message [[RequestRegisteredTaskManagers]].
+ *
+ * @param taskManagers
+ */
+ case class RegisteredTaskManagers(taskManagers: Iterable[Instance]){
+ def asJavaIterable: java.lang.Iterable[Instance] = {
+ import scala.collection.JavaConverters._
+ taskManagers.asJava
+ }
+
+ def asJavaCollection: java.util.Collection[Instance] = {
+ import scala.collection.JavaConverters._
+ taskManagers.asJavaCollection
+ }
+ }
+
+ /**
+ * Requests stack trace messages of the task manager
+ *
+ * @param instanceID Instance ID of the task manager
+ */
+ case class RequestStackTrace(instanceID: InstanceID)
+
+ /**
+ * Requests the current state of the job manager
+ */
+ case object RequestJobManagerStatus
+
+ /**
+ * Response to RequestJobManagerStatus
+ */
+ sealed trait JobManagerStatus
+
+ case object JobManagerStatusAlive extends JobManagerStatus
+
+ // --------------------------------------------------------------------------
+ // Utility methods to allow simpler case object access from Java
+ // --------------------------------------------------------------------------
+
+ def getRequestNumberRegisteredTaskManager : AnyRef = {
+ RequestNumberRegisteredTaskManager
+ }
+
+ def getRequestTotalNumberOfSlots : AnyRef = {
+ RequestTotalNumberOfSlots
+ }
+
+ def getRequestBlobManagerPort : AnyRef = {
+ RequestBlobManagerPort
+ }
+
+ def getRequestRunningJobs : AnyRef = {
+ RequestRunningJobs
+ }
+
+ def getRequestRegisteredTaskManagers : AnyRef = {
+ RequestRegisteredTaskManagers
+ }
+
+ def getRequestJobManagerStatus : AnyRef = {
+ RequestJobManagerStatus
+ }
+
+ def getJobManagerStatusAlive : AnyRef = {
+ JobManagerStatusAlive
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/5ac1911a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobmanagerMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobmanagerMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobmanagerMessages.scala
deleted file mode 100644
index 8aef552..0000000
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobmanagerMessages.scala
+++ /dev/null
@@ -1,330 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.messages
-
-import org.apache.flink.runtime.accumulators.AccumulatorEvent
-import org.apache.flink.runtime.executiongraph.{ExecutionAttemptID, ExecutionGraph}
-import org.apache.flink.runtime.instance.{InstanceID, Instance}
-import org.apache.flink.runtime.jobgraph.{JobGraph, JobID, JobStatus, JobVertexID}
-import org.apache.flink.runtime.taskmanager.TaskExecutionState
-
-/**
- * The job manager specific messages
- */
-object JobManagerMessages {
- /**
- * Submits a job to the job manager. If [[registerForEvents]] is true,
- * then the sender will be registered as listener for the state change messages. If [[detached]]
- * is set to true, then the sender will detach from the job execution. Consequently,
- * he will not receive the job execution result [[JobResult]]. The submission result will be sent
- * back to the
- * sender as a [[SubmissionResponse]] message.
- *
- * @param jobGraph
- * @param registerForEvents if true, then register for state change events
- * @param detached if true, then detach from the job execution
- */
- case class SubmitJob(jobGraph: JobGraph, registerForEvents: Boolean = false,
- detached: Boolean = false)
-
- /**
- * Cancels a job with the given [[jobID]] at the JobManager. The result of the cancellation is
- * sent back to the sender as a [[CancellationResponse]] message.
- *
- * @param jobID
- */
- case class CancelJob(jobID: JobID)
-
- /**
- * Denotes a state change of a task at the JobManager. The update success is acknowledged by a
- * boolean value which is sent back to the sender.
- *
- * @param taskExecutionState
- */
- case class UpdateTaskExecutionState(taskExecutionState: TaskExecutionState)
-
- /**
- * Requesting next input split for the
- * [[org.apache.flink.runtime.executiongraph.ExecutionJobVertex]]
- * of the job specified by [[jobID]]. The next input split is sent back to the sender as a
- * [[org.apache.flink.runtime.messages.TaskManagerMessages.NextInputSplit]] message.
- *
- * @param jobID
- * @param vertexID
- */
- case class RequestNextInputSplit(jobID: JobID, vertexID: JobVertexID, executionAttempt:
- ExecutionAttemptID)
-
- /**
- * Notifies the [[org.apache.flink.runtime.jobmanager.JobManager]] about available data for a
- * produced partition.
- * <p>
- * There is a call to this method for each
- * [[org.apache.flink.runtime.executiongraph.ExecutionVertex]] instance once per produced
- * [[org.apache.flink.runtime.io.network.partition.IntermediateResultPartition]] instance,
- * either when first producing data (for pipelined executions) or when all data has been produced
- * (for staged executions).
- * <p>
- * The [[org.apache.flink.runtime.jobmanager.JobManager]] then can decide when to schedule the
- * partition consumers of the given session.
- *
- * @see [[org.apache.flink.runtime.io.network.partition.IntermediateResultPartition]]
- */
- case class ScheduleOrUpdateConsumers(jobId: JobID,
- executionId: ExecutionAttemptID,
- partitionIndex: Int)
-
- case class ConsumerNotificationResult(success: Boolean, error: Option[Throwable] = None)
-
- /**
- * Reports the accumulator results of the individual tasks to the job manager.
- *
- * @param accumulatorEvent
- */
- case class ReportAccumulatorResult(accumulatorEvent: AccumulatorEvent)
-
- /**
- * Requests the accumulator results of the job identified by [[jobID]] from the job manager.
- * The result is sent back to the sender as a [[AccumulatorResultsResponse]] message.
- *
- * @param jobID
- */
- case class RequestAccumulatorResults(jobID: JobID)
-
- sealed trait AccumulatorResultsResponse{
- val jobID: JobID
- }
-
- /**
- * Contains the retrieved accumulator results from the job manager. This response is triggered
- * by [[RequestAccumulatorResults]].
- *
- * @param jobID
- * @param results
- */
- case class AccumulatorResultsFound(jobID: JobID, results: Map[String,
- Object]) extends AccumulatorResultsResponse{
- def asJavaMap: java.util.Map[String, Object] = {
- import scala.collection.JavaConverters._
- results.asJava
- }
- }
-
- /**
- * Denotes that no accumulator results for [[jobID]] could be found at the job manager.
- * @param jobID
- */
- case class AccumulatorResultsNotFound(jobID: JobID) extends AccumulatorResultsResponse
-
- /**
- * Requests the current [[JobStatus]] of the job identified by [[jobID]]. This message triggers
- * as response a [[JobStatusResponse]] message.
- *
- * @param jobID
- */
- case class RequestJobStatus(jobID: JobID)
-
- sealed trait JobStatusResponse {
- def jobID: JobID
- }
-
- /**
- * Denotes the current [[JobStatus]] of the job with [[jobID]].
- *
- * @param jobID
- * @param status
- */
- case class CurrentJobStatus(jobID: JobID, status: JobStatus) extends JobStatusResponse
-
- /**
- * Requests the number of currently registered task manager at the job manager. The result is
- * sent back to the sender as an [[Int]].
- */
- case object RequestNumberRegisteredTaskManager
-
- /**
- * Requests the maximum number of slots available to the job manager. The result is sent back
- * to the sender as an [[Int]].
- */
- case object RequestTotalNumberOfSlots
-
- /**
- * Requests the port of the blob manager from the job manager. The result is sent back to the
- * sender as an [[Int]].
- */
- case object RequestBlobManagerPort
-
- /**
- * Requests the final job status of the job with [[jobID]]. If the job has not been terminated
- * then the result is sent back upon termination of the job. The result is a
- * [[JobStatusResponse]] message.
- *
- * @param jobID
- */
- case class RequestFinalJobStatus(jobID: JobID)
-
- /**
- * Denotes a successful job execution.
- *
- * @param jobID
- * @param runtime
- * @param accumulatorResults
- */
- case class JobResultSuccess(jobID: JobID, runtime: Long, accumulatorResults: java.util.Map[String,
- AnyRef]) {}
-
- sealed trait CancellationResponse{
- def jobID: JobID
- }
-
- /**
- * Denotes a successful job cancellation
- * @param jobID
- */
- case class CancellationSuccess(jobID: JobID) extends CancellationResponse
-
- /**
- * Denotes a failed job cancellation
- * @param jobID
- * @param cause
- */
- case class CancellationFailure(jobID: JobID, cause: Throwable) extends CancellationResponse
-
- /**
- * Requests all currently running jobs from the job manager. This message triggers a
- * [[RunningJobs]] response.
- */
- case object RequestRunningJobs
-
- /**
- * This message is the response to the [[RequestRunningJobs]] message. It contains all
- * execution graphs of the currently running jobs.
- *
- * @param runningJobs
- */
- case class RunningJobs(runningJobs: Iterable[ExecutionGraph]) {
- def this() = this(Seq())
- def asJavaIterable: java.lang.Iterable[ExecutionGraph] = {
- import scala.collection.JavaConverters._
- runningJobs.asJava
- }
- }
-
- /**
- * Requests the execution graph of a specific job identified by [[jobID]]. The result is sent
- * back to the sender as a [[JobResponse]].
- * @param jobID
- */
- case class RequestJob(jobID: JobID)
-
- sealed trait JobResponse{
- def jobID: JobID
- }
-
- /**
- * Contains the [[executionGraph]] of a job with [[jobID]]. This is the response to
- * [[RequestJob]] if the job runs or is archived.
- *
- * @param jobID
- * @param executionGraph
- */
- case class JobFound(jobID: JobID, executionGraph: ExecutionGraph) extends JobResponse
-
- /**
- * Denotes that there is no job with [[jobID]] retrievable. This message can be the response of
- * [[RequestJob]] or [[RequestJobStatus]].
- *
- * @param jobID
- */
- case class JobNotFound(jobID: JobID) extends JobResponse with JobStatusResponse
-
- /**
- * Requests the instances of all registered task managers.
- */
- case object RequestRegisteredTaskManagers
-
- /**
- * Contains the [[Instance]] objects of all registered task managers. It is the response to the
- * message [[RequestRegisteredTaskManagers]].
- *
- * @param taskManagers
- */
- case class RegisteredTaskManagers(taskManagers: Iterable[Instance]){
- def asJavaIterable: java.lang.Iterable[Instance] = {
- import scala.collection.JavaConverters._
- taskManagers.asJava
- }
-
- def asJavaCollection: java.util.Collection[Instance] = {
- import scala.collection.JavaConverters._
- taskManagers.asJavaCollection
- }
- }
-
- /**
- * Requests stack trace messages of the task manager
- *
- * @param instanceID Instance ID of the task manager
- */
- case class RequestStackTrace(instanceID: InstanceID)
-
- /**
- * Requests the current state of the job manager
- */
- case object RequestJobManagerStatus
-
- /**
- * Response to RequestJobManagerStatus
- */
- sealed trait JobManagerStatus
-
- case object JobManagerStatusAlive extends JobManagerStatus
-
- // --------------------------------------------------------------------------
- // Utility methods to allow simpler case object access from Java
- // --------------------------------------------------------------------------
-
- def getRequestNumberRegisteredTaskManager : AnyRef = {
- RequestNumberRegisteredTaskManager
- }
-
- def getRequestTotalNumberOfSlots : AnyRef = {
- RequestTotalNumberOfSlots
- }
-
- def getRequestBlobManagerPort : AnyRef = {
- RequestBlobManagerPort
- }
-
- def getRequestRunningJobs : AnyRef = {
- RequestRunningJobs
- }
-
- def getRequestRegisteredTaskManagers : AnyRef = {
- RequestRegisteredTaskManagers
- }
-
- def getRequestJobManagerStatus : AnyRef = {
- RequestJobManagerStatus
- }
-
- def getJobManagerStatusAlive : AnyRef = {
- JobManagerStatusAlive
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/5ac1911a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/Messages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/Messages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/Messages.scala
index 4f27761..91c3b2d 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/Messages.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/Messages.scala
@@ -20,6 +20,16 @@ package org.apache.flink.runtime.messages
object Messages {
+ /**
+ * Message to signal the successful reception of another message
+ */
case object Acknowledge
+ /**
+ * Signals that the JobManager/TaskManager shall disconnect from the sender
+ * (TaskManager/JobManager)
+ * @param reason
+ */
+ case class Disconnect(reason: String)
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5ac1911a/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 c14ebca..ae192ce 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
@@ -46,7 +46,7 @@ import org.apache.flink.runtime.jobgraph.IntermediateDataSetID
import org.apache.flink.runtime.jobmanager.JobManager
import org.apache.flink.runtime.memorymanager.DefaultMemoryManager
import org.apache.flink.runtime.messages.JobManagerMessages.UpdateTaskExecutionState
-import org.apache.flink.runtime.messages.Messages.Acknowledge
+import org.apache.flink.runtime.messages.Messages.{Disconnect, Acknowledge}
import org.apache.flink.runtime.messages.RegistrationMessages.{AlreadyRegistered,
RefuseRegistration, AcknowledgeRegistration, RegisterTaskManager}
import org.apache.flink.runtime.messages.TaskManagerMessages._
@@ -106,6 +106,7 @@ import scala.collection.JavaConverters._
var registrationDelay = 50 milliseconds
var registrationDuration = 0 seconds
+ var registrationAttempts: Int = 0
TaskManager.checkTempDirs(tmpDirPaths)
val ioManager = new IOManagerAsync(tmpDirPaths)
@@ -131,11 +132,9 @@ import scala.collection.JavaConverters._
log.info(TaskManager.getMemoryUsageStatsAsString(ManagementFactory.getMemoryMXBean))
}
- var libraryCacheManager: LibraryCacheManager = null
+ var libraryCacheManager: Option[LibraryCacheManager] = None
var networkEnvironment: Option[NetworkEnvironment] = None
- var registrationAttempts: Int = 0
- var registered: Boolean = false
- var currentJobManager = ActorRef.noSender
+ var currentJobManager: Option[ActorRef] = None
var profilerListener: Option[ActorRef] = None
var instanceID: InstanceID = null
var heartbeatScheduler: Option[Cancellable] = None
@@ -155,18 +154,7 @@ import scala.collection.JavaConverters._
cancelAndClearEverything(new Exception("Task Manager is shutting down."))
- heartbeatScheduler foreach {
- _.cancel()
- }
-
- networkEnvironment foreach {
- ne =>
- try {
- ne.shutdown()
- } catch {
- case t: Throwable => log.error(t, "ChannelManager did not shutdown properly.")
- }
- }
+ cleanupTaskManager()
ioManager.shutdown()
memoryManager.shutdown()
@@ -177,38 +165,25 @@ import scala.collection.JavaConverters._
case t: Throwable => log.error(t, "FileCache did not shutdown properly.")
}
- if (libraryCacheManager != null) {
- try {
- libraryCacheManager.shutdown()
- }
- catch {
- case t: Throwable => log.error(t, "LibraryCacheManager did not shutdown properly.")
- }
- }
-
if(log.isDebugEnabled){
log.debug("Task manager {} is completely stopped.", self.path)
}
}
private def tryJobManagerRegistration(): Unit = {
- registrationDuration = 0 seconds
-
- registered = false
- currentJobManager = ActorRef.noSender
-
context.system.scheduler.scheduleOnce(registrationDelay, self, RegisterAtJobManager)
}
override def receiveWithLogMessages: Receive = {
case RegisterAtJobManager =>
- if(!registered) {
+ if(currentJobManager.isEmpty) {
registrationDuration += registrationDelay
// double delay for exponential backoff
registrationDelay *= 2
+ registrationAttempts += 1
if (registrationDuration > maxRegistrationDuration) {
- log.warning("TaskManager could not register at JobManager {} after {}.",
+ log.error("TaskManager could not register at JobManager {} after {}.",
jobManagerAkkaURL,
maxRegistrationDuration)
@@ -225,15 +200,15 @@ import scala.collection.JavaConverters._
}
case AcknowledgeRegistration(id, blobPort, profilerListener) =>
- if(!registered) {
+ if(currentJobManager.isEmpty) {
finishRegistration(sender, id, blobPort, profilerListener)
} else {
log.info("The TaskManager {} is already registered at the JobManager {}, but received " +
- "another AcknowledgeRegistration message.", self.path, currentJobManager.path)
+ "another AcknowledgeRegistration message.", self.path, sender.path)
}
case AlreadyRegistered(id, blobPort, profilerListener) =>
- if(!registered) {
+ if(currentJobManager.isEmpty) {
log.warning("The TaskManager {} seems to be already registered at the JobManager {} even" +
"though it has not yet finished the registration process.", self.path, sender.path)
@@ -245,7 +220,7 @@ import scala.collection.JavaConverters._
}
case RefuseRegistration(reason) =>
- if(!registered) {
+ if(currentJobManager.isEmpty) {
log.error("The registration of task manager {} was refused by the job manager {} " +
"because {}.", self.path, jobManagerAkkaURL, reason)
@@ -288,31 +263,35 @@ import scala.collection.JavaConverters._
unregisterTask(executionID)
case updateMsg:UpdateTaskExecutionState =>
- val futureResponse = (currentJobManager ? updateMsg)(timeout)
-
- val jobID = updateMsg.taskExecutionState.getJobID
- val executionID = updateMsg.taskExecutionState.getID
- val executionState = updateMsg.taskExecutionState.getExecutionState
-
- futureResponse.mapTo[Boolean].onComplete{
- case Success(result) =>
- if(!result){
- self ! FailTask(executionID,
- new IllegalStateException("Task has been disposed on JobManager."))
+ currentJobManager foreach {
+ jobManager => {
+ val futureResponse = (jobManager ? updateMsg)(timeout)
+
+ val jobID = updateMsg.taskExecutionState.getJobID
+ val executionID = updateMsg.taskExecutionState.getID
+ val executionState = updateMsg.taskExecutionState.getExecutionState
+
+ futureResponse.mapTo[Boolean].onComplete {
+ case Success(result) =>
+ if (!result) {
+ self ! FailTask(executionID,
+ new IllegalStateException("Task has been disposed on JobManager."))
+ }
+
+ if (!result || executionState == ExecutionState.FINISHED || executionState ==
+ ExecutionState.CANCELED || executionState == ExecutionState.FAILED) {
+ self ! UnregisterTask(executionID)
+ }
+ case Failure(t) =>
+ log.error(t, "Execution state change notification failed for task {}" +
+ s"of job {}.", executionID, jobID)
+ self ! UnregisterTask(executionID)
}
-
- if (!result || executionState == ExecutionState.FINISHED || executionState ==
- ExecutionState.CANCELED || executionState == ExecutionState.FAILED) {
- self ! UnregisterTask(executionID)
- }
- case Failure(t) =>
- log.warning(s"Execution state change notification failed for task $executionID " +
- s"of job $jobID. Cause ${t.getMessage}.")
- self ! UnregisterTask(executionID)
+ }
}
case SendHeartbeat =>
- currentJobManager ! Heartbeat(instanceID)
+ currentJobManager foreach { _ ! Heartbeat(instanceID) }
case LogMemoryUsage =>
logMemoryStats()
@@ -327,7 +306,7 @@ import scala.collection.JavaConverters._
sender ! StackTrace(instanceID, stackTraceStr)
case NotifyWhenRegisteredAtJobManager =>
- if (registered) {
+ if (currentJobManager.isDefined) {
sender ! RegisteredAtJobManager
} else {
waitForRegistration += sender
@@ -347,7 +326,7 @@ import scala.collection.JavaConverters._
case Terminated(jobManager) =>
log.info("Job manager {} is no longer reachable. Cancelling all tasks and trying to " +
- "reregister.", jobManager.path)
+ "reregister.", jobManagerAkkaURL)
cancelAndClearEverything(new Throwable("Lost connection to JobManager"))
@@ -355,6 +334,16 @@ import scala.collection.JavaConverters._
tryJobManagerRegistration()
+ case Disconnect(msg) =>
+ log.info("Job manager {} wants {} to disconnect. Reason {}.", jobManagerAkkaURL,
+ self.path, msg)
+
+ cancelAndClearEverything(new Throwable("Job manager wants me to disconnect."))
+
+ cleanupTaskManager()
+
+ tryJobManagerRegistration()
+
case FailIntermediateResultPartitions(executionID) =>
log.info("Fail intermediate result partitions associated with execution {}.", executionID)
networkEnvironment foreach {
@@ -386,19 +375,24 @@ import scala.collection.JavaConverters._
var task: Task = null
try {
- if (log.isDebugEnabled) {
- startRegisteringTask = System.currentTimeMillis()
- }
- libraryCacheManager.registerTask(jobID, executionID, tdd.getRequiredJarFiles)
- // triggers the download of all missing jar files from the job manager
- libraryCacheManager.registerTask(jobID, executionID, tdd.getRequiredJarFiles)
+ val userCodeClassLoader = libraryCacheManager match {
+ case Some(manager) =>
+ if (log.isDebugEnabled) {
+ startRegisteringTask = System.currentTimeMillis()
+ }
- if (log.isDebugEnabled) {
- log.debug("Register task {} took {}s", executionID,
- (System.currentTimeMillis() - startRegisteringTask) / 1000.0)
- }
+ manager.registerTask(jobID, executionID, tdd.getRequiredJarFiles)
+ // triggers the download of all missing jar files from the job manager
+ manager.registerTask(jobID, executionID, tdd.getRequiredJarFiles)
+
+ if (log.isDebugEnabled) {
+ log.debug("Register task {} took {}s", executionID,
+ (System.currentTimeMillis() - startRegisteringTask) / 1000.0)
+ }
- val userCodeClassLoader = libraryCacheManager.getClassLoader(jobID)
+ manager.getClassLoader(jobID)
+ case None => throw new IllegalStateException("There is no valid library cache manager.")
+ }
if (userCodeClassLoader == null) {
throw new RuntimeException("No user code Classloader available.")
@@ -413,11 +407,17 @@ import scala.collection.JavaConverters._
case None =>
}
- val splitProvider = new TaskInputSplitProvider(currentJobManager, jobID, vertexID,
- executionID, userCodeClassLoader, timeout)
+ val env = currentJobManager match {
+ case Some(jobManager) =>
+ val splitProvider = new TaskInputSplitProvider(jobManager, jobID, vertexID,
+ executionID, userCodeClassLoader, timeout)
- val env = new RuntimeEnvironment(currentJobManager, task, tdd, userCodeClassLoader,
- memoryManager, ioManager, splitProvider, bcVarManager, networkEnvironment.get)
+ new RuntimeEnvironment(jobManager, task, tdd, userCodeClassLoader,
+ memoryManager, ioManager, splitProvider, bcVarManager, networkEnvironment.get)
+
+ case None => throw new IllegalStateException("TaskManager has not yet registered at " +
+ "the JobManager.")
+ }
task.setEnvironment(env)
@@ -465,7 +465,7 @@ import scala.collection.JavaConverters._
removeAllTaskResources(task)
}
- libraryCacheManager.unregisterTask(jobID, executionID)
+ libraryCacheManager foreach { _.unregisterTask(jobID, executionID) }
} catch {
case t: Throwable => log.error(t, "Error during cleanup of task deployment.")
}
@@ -475,23 +475,31 @@ import scala.collection.JavaConverters._
}
private def cleanupTaskManager(): Unit = {
- context.unwatch(currentJobManager)
+ currentJobManager foreach {
+ context.unwatch(_)
+ }
networkEnvironment foreach {
- _.shutdown()
+ ne =>
+ try {
+ ne.shutdown()
+ } catch {
+ case t: Throwable => log.error(t, "ChannelManager did not shutdown properly.")
+ }
}
networkEnvironment = None
- if(libraryCacheManager != null){
- try {
- libraryCacheManager.shutdown()
- } catch {
- case t: Throwable => log.error(t, "Could not shut down the library cache manager.")
- }
+ libraryCacheManager foreach {
+ manager =>
+ try {
+ manager.shutdown()
+ } catch {
+ case t: Throwable => log.error(t, "Could not shut down the library cache manager.")
+ }
}
- libraryCacheManager = null
+ libraryCacheManager = None
heartbeatScheduler foreach {
_.cancel()
@@ -507,9 +515,10 @@ import scala.collection.JavaConverters._
}
profilerListener = None
- currentJobManager = ActorRef.noSender
+ currentJobManager = None
instanceID = null
- registered = false
+ registrationAttempts = 0
+ registrationDuration = 0 seconds
}
private def updateTask(executionId: ExecutionAttemptID,
@@ -567,15 +576,15 @@ import scala.collection.JavaConverters._
private def setupTaskManager(jobManager: ActorRef, id: InstanceID, blobPort: Int,
profilerListener: Option[ActorRef]): Unit = {
- registered = true
- currentJobManager = jobManager
+
+ currentJobManager = Some(jobManager)
this.profilerListener = profilerListener
instanceID = id
// watch job manager to detect when it dies
- context.watch(currentJobManager)
+ context.watch(jobManager)
- setupNetworkEnvironment()
+ setupNetworkEnvironment(jobManager)
setupLibraryCacheManager(blobPort)
// schedule regular heartbeat message for oneself
@@ -590,7 +599,7 @@ import scala.collection.JavaConverters._
}
}
- private def setupNetworkEnvironment(): Unit = {
+ private def setupNetworkEnvironment(jobManager: ActorRef): Unit = {
//shutdown existing network environment
networkEnvironment foreach {
ne =>
@@ -602,7 +611,7 @@ import scala.collection.JavaConverters._
}
try {
- networkEnvironment = Some(new NetworkEnvironment(self, currentJobManager, timeout,
+ networkEnvironment = Some(new NetworkEnvironment(self, jobManager, timeout,
networkConfig))
} catch {
case ioe: IOException =>
@@ -613,27 +622,29 @@ import scala.collection.JavaConverters._
private def setupLibraryCacheManager(blobPort: Int): Unit = {
// shutdown existing library cache manager first
- if (libraryCacheManager != null) {
- try {
- libraryCacheManager.shutdown()
- }
- catch {
- case t: Throwable => log.error(t, "Could not properly shut down LibraryCacheManager.")
+ libraryCacheManager foreach {
+ manager => {
+ try{
+ manager.shutdown()
+ } catch {
+ case t: Throwable => log.error(t, "Could not properly shut down LibraryCacheManager.")
+ }
}
- libraryCacheManager = null
}
// Check if a blob server is specified
if (blobPort > 0) {
- val address = new InetSocketAddress(currentJobManager.path.address.host.getOrElse
- ("localhost"), blobPort)
+
+ val address = new InetSocketAddress(
+ currentJobManager.flatMap(_.path.address.host).getOrElse("localhost"),
+ blobPort)
log.info("Determined BLOB server address to be {}.", address)
- libraryCacheManager = new BlobLibraryCacheManager(
- new BlobCache(address, configuration), cleanupInterval)
+ libraryCacheManager = Some(new BlobLibraryCacheManager(
+ new BlobCache(address, configuration), cleanupInterval))
} else {
- libraryCacheManager = new FallbackLibraryCacheManager
+ libraryCacheManager = Some(new FallbackLibraryCacheManager)
}
}
@@ -646,18 +657,17 @@ import scala.collection.JavaConverters._
for (t <- runningTasks.values) {
t.failExternally(cause)
- runningTasks.remove(t.getExecutionId)
+ unregisterTask(t.getExecutionId)
}
}
}
private def unregisterTask(executionID: ExecutionAttemptID): Unit = {
- log.info("Unregister task with execution ID {}.", executionID)
-
runningTasks.remove(executionID) match {
case Some(task) =>
+ log.info("Unregister task with execution ID {}.", executionID)
removeAllTaskResources(task)
- libraryCacheManager.unregisterTask(task.getJobID, executionID)
+ libraryCacheManager foreach { _.unregisterTask(task.getJobID, executionID) }
case None =>
if (log.isDebugEnabled) {
log.debug("Cannot find task with ID {} to unregister.", executionID)
http://git-wip-us.apache.org/repos/asf/flink/blob/5ac1911a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
index 87fca99..895da46 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
@@ -19,12 +19,15 @@
package org.apache.flink.runtime.testingUtils
import akka.actor.{ActorRef, Props, ActorSystem}
+import akka.pattern.Patterns.gracefulStop
import org.apache.flink.configuration.{ConfigConstants, Configuration}
import org.apache.flink.runtime.jobmanager.{MemoryArchivist, JobManager}
import org.apache.flink.runtime.minicluster.FlinkMiniCluster
import org.apache.flink.runtime.net.NetUtils
import org.apache.flink.runtime.taskmanager.TaskManager
+import scala.concurrent.Await
+
/**
* Testing cluster which starts the [[JobManager]] and [[TaskManager]] actors with testing support
* in the same [[ActorSystem]].
@@ -71,4 +74,11 @@ FlinkMiniCluster(userConfiguration, singleActorSystem) {
networkConnectionConfig) with TestingTaskManager), TaskManager.TASK_MANAGER_NAME + "_" +
(index + 1))
}
+ val stopped = gracefulStop(jobManagerActor, TestingUtils.TESTING_DURATION)
+ Await.result(stopped, TestingUtils.TESTING_DURATION)
+
+ jobManagerActorSystem.shutdown()
+ jobManagerActorSystem.awaitTermination()
+
+ jobManagerActorSystem = startJobManagerActorSystem()
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5ac1911a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
index 566f661..f41e114 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
@@ -111,6 +111,7 @@ trait TestingJobManager extends ActorLogMessages with WrapAsScala {
import context.dispatcher
Future.fold(responses)(true)(_ & _) pipeTo sender
+
case NotifyWhenTaskManagerTerminated(taskManager) =>
val waiting = waitForTaskManagerToBeTerminated.getOrElse(taskManager.path.name, Set())
waitForTaskManagerToBeTerminated += taskManager.path.name -> (waiting + sender)
@@ -118,7 +119,7 @@ trait TestingJobManager extends ActorLogMessages with WrapAsScala {
case msg@Terminated(taskManager) =>
super.receiveWithLogMessages(msg)
- waitForTaskManagerToBeTerminated.get(taskManager.path.name) foreach {
+ waitForTaskManagerToBeTerminated.remove(taskManager.path.name) foreach {
_ foreach {
listener =>
listener ! TaskManagerTerminated(taskManager)
@@ -151,16 +152,23 @@ trait TestingJobManager extends ActorLogMessages with WrapAsScala {
case msg@JobStatusChanged(jobID, newJobStatus, _, _) =>
super.receiveWithLogMessages(msg)
- waitForJobStatus.get(jobID) match {
+
+ val cleanup = waitForJobStatus.get(jobID) match {
case Some(stateListener) =>
- stateListener.get(newJobStatus) match {
+ stateListener.remove(newJobStatus) match {
case Some(listeners) =>
listeners foreach {
_ ! JobStatusIs(jobID, newJobStatus)
}
case _ =>
}
- case _ =>
+ stateListener.isEmpty
+
+ case _ => false
+ }
+
+ if (cleanup) {
+ waitForJobStatus.remove(jobID)
}
}
@@ -185,30 +193,24 @@ trait TestingJobManager extends ActorLogMessages with WrapAsScala {
}
def notifyListeners(jobID: JobID): Unit = {
- val cleanupRunning = waitForAllVerticesToBeRunning.get(jobID) match {
- case Some(listeners) if checkIfAllVerticesRunning(jobID) =>
- for(listener <- listeners){
- listener ! AllVerticesRunning(jobID)
- }
- true
- case _ => false
- }
-
- if(cleanupRunning){
- waitForAllVerticesToBeRunning.remove(jobID)
- }
-
- val cleanupRunningOrFinished = waitForAllVerticesToBeRunningOrFinished.get(jobID) match {
- case Some(listeners) if checkIfAllVerticesRunningOrFinished(jobID) =>
- for(listener <- listeners){
- listener ! AllVerticesRunning(jobID)
- }
- true
- case _ => false
+ if(checkIfAllVerticesRunning((jobID))) {
+ waitForAllVerticesToBeRunning.remove(jobID) match {
+ case Some(listeners) =>
+ for (listener <- listeners) {
+ listener ! AllVerticesRunning(jobID)
+ }
+ case _ =>
+ }
}
- if (cleanupRunningOrFinished) {
- waitForAllVerticesToBeRunningOrFinished.remove(jobID)
+ if(checkIfAllVerticesRunningOrFinished(jobID)) {
+ waitForAllVerticesToBeRunningOrFinished.remove(jobID) match {
+ case Some(listeners) =>
+ for (listener <- listeners) {
+ listener ! AllVerticesRunning(jobID)
+ }
+ case _ =>
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5ac1911a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingMessages.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingMessages.scala
new file mode 100644
index 0000000..52cc1f3
--- /dev/null
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingMessages.scala
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.testingUtils
+
+object TestingMessages {
+
+ case object DisableDisconnect
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/5ac1911a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
index abdb173..0ecbd04 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
@@ -22,10 +22,12 @@ import akka.actor.{Terminated, ActorRef}
import org.apache.flink.runtime.ActorLogMessages
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID
import org.apache.flink.runtime.jobgraph.JobID
+import org.apache.flink.runtime.messages.Messages.Disconnect
import org.apache.flink.runtime.messages.TaskManagerMessages.UnregisterTask
import org.apache.flink.runtime.taskmanager.TaskManager
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages.NotifyWhenJobRemoved
import org.apache.flink.runtime.ActorLogMessages
+import org.apache.flink.runtime.testingUtils.TestingMessages.DisableDisconnect
import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages._
import scala.concurrent.duration._
@@ -41,6 +43,8 @@ trait TestingTaskManager extends ActorLogMessages {
val waitForJobRemoval = scala.collection.mutable.HashMap[JobID, Set[ActorRef]]()
val waitForJobManagerToBeTerminated = scala.collection.mutable.HashMap[String, Set[ActorRef]]()
+ var disconnectDisabled = false
+
abstract override def receiveWithLogMessages = {
receiveTestMessages orElse super.receiveWithLogMessages
}
@@ -60,7 +64,7 @@ trait TestingTaskManager extends ActorLogMessages {
case UnregisterTask(executionID) =>
super.receiveWithLogMessages(UnregisterTask(executionID))
- waitForRemoval.get(executionID) match {
+ waitForRemoval.remove(executionID) match {
case Some(actors) => for(actor <- actors) actor ! true
case None =>
}
@@ -77,7 +81,7 @@ trait TestingTaskManager extends ActorLogMessages {
case None => sender ! ResponseNumActiveConnections(0)
}
- case NotifyWhenJobRemoved(jobID) =>
+ case NotifyWhenJobRemoved(jobID) =>
if(runningTasks.values.exists(_.getJobID == jobID)){
val set = waitForJobRemoval.getOrElse(jobID, Set())
waitForJobRemoval += (jobID -> (set + sender))
@@ -92,7 +96,7 @@ trait TestingTaskManager extends ActorLogMessages {
case CheckIfJobRemoved(jobID) =>
if(runningTasks.values.forall(_.getJobID != jobID)){
- waitForJobRemoval.get(jobID) match {
+ waitForJobRemoval.remove(jobID) match {
case Some(listeners) => listeners foreach (_ ! true)
case None =>
}
@@ -108,10 +112,31 @@ trait TestingTaskManager extends ActorLogMessages {
case msg@Terminated(jobManager) =>
super.receiveWithLogMessages(msg)
- waitForJobManagerToBeTerminated.get(jobManager.path.name) foreach {
+ waitForJobManagerToBeTerminated.remove(jobManager.path.name) foreach {
_ foreach {
_ ! JobManagerTerminated(jobManager)
}
}
+
+ case msg:Disconnect =>
+ super.receiveWithLogMessages(msg)
+
+ val jobManager = sender
+
+ waitForJobManagerToBeTerminated.remove(jobManager.path.name) foreach {
+ _ foreach {
+ _ ! JobManagerTerminated(jobManager)
+ }
+ }
+
+ case msg:Disconnect =>
+ if (!disconnectDisabled) {
+ super.receiveWithLogMessages(msg)
+ }
+
+ case DisableDisconnect =>
+ disconnectDisabled = true
}
+
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5ac1911a/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 66b2438..3813beb 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
@@ -22,7 +22,7 @@ import akka.actor.{ActorSystem, PoisonPill}
import akka.testkit.{ImplicitSender, TestKit}
import org.apache.flink.configuration.{ConfigConstants, Configuration}
import org.apache.flink.runtime.akka.AkkaUtils
-import org.apache.flink.runtime.messages.JobManagerMessages.RequestNumberRegisteredTaskManager
+import org.apache.flink.runtime.jobmanager.Tasks.{NoOpInvokable, BlockingNoOpInvokable}
import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages.{JobManagerTerminated, NotifyWhenJobManagerTerminated}
import org.apache.flink.test.util.ForkableFlinkMiniCluster
import org.junit.runner.RunWith
@@ -55,6 +55,9 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
val tm = cluster.getTaskManagers(0)
val jm = cluster.getJobManager
+ // disable disconnect message to test death watch
+ tm ! DisableDisconnect
+
try{
jm ! RequestNumberRegisteredTaskManager
expectMsg(1)
@@ -72,7 +75,62 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
cluster.getJobManager ! RequestNumberRegisteredTaskManager
expectMsg(1)
- }finally{
+ } finally {
+ cluster.stop()
+ }
+ }
+ }
+
+ "The system" should {
+ "go into a clean state in case of a JobManager failure" in {
+ val num_slots = 20
+
+ val sender = new AbstractJobVertex("BlockingSender")
+ sender.setParallelism(num_slots)
+ sender.setInvokableClass(classOf[BlockingNoOpInvokable])
+ val jobGraph = new JobGraph("Blocking Testjob", sender)
+
+ val noOp = new AbstractJobVertex("NoOpInvokable")
+ noOp.setParallelism(num_slots)
+ noOp.setInvokableClass(classOf[NoOpInvokable])
+ val jobGraph2 = new JobGraph("NoOp Testjob", noOp)
+
+ val cluster = TestingUtils.startTestingClusterDeathWatch(num_slots/2, 2)
+
+ var jm = cluster.getJobManager
+ val tm = cluster.getTaskManagers(0)
+
+ try{
+ within(TestingUtils.TESTING_DURATION) {
+ jm ! SubmitJob(jobGraph)
+ expectMsg(SubmissionSuccess(jobGraph.getJobID))
+
+ tm ! NotifyWhenJobManagerTerminated(jm)
+
+ jm ! PoisonPill
+
+ expectMsgClass(classOf[JobManagerTerminated])
+
+ cluster.restartJobManager()
+
+ jm = cluster.getJobManager
+
+ cluster.waitForTaskManagersToBeRegistered()
+
+ jm ! SubmitJob(jobGraph2)
+ val response = expectMsgType[SubmissionResponse]
+
+ response match {
+ case SubmissionSuccess(jobID) => jobID should equal(jobGraph2.getJobID)
+ case SubmissionFailure(jobID, t) =>
+ fail("Submission of the second job failed.", t)
+ }
+
+ val result = expectMsgType[JobResultSuccess]
+
+ result.jobID should equal(jobGraph2.getJobID)
+ }
+ } finally {
cluster.stop()
}
}
[6/9] flink git commit: [FLINK-1556] [runtime] Fails jobs properly in
case of a job submission exception
Posted by se...@apache.org.
[FLINK-1556] [runtime] Fails jobs properly in case of a job submission exception
This closes #422
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4ff91cd7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4ff91cd7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4ff91cd7
Branch: refs/heads/master
Commit: 4ff91cd7cd3e05637d70a56cfc6f5a4ba2f2501a
Parents: dbf589a
Author: Till Rohrmann <tr...@apache.org>
Authored: Thu Feb 19 12:44:32 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Feb 19 19:07:02 2015 +0100
----------------------------------------------------------------------
.../org/apache/flink/client/program/Client.java | 20 +-
.../apache/flink/client/program/ClientTest.java | 8 +-
.../client/JobCancellationException.java | 10 +-
.../runtime/client/JobExecutionException.java | 24 +-
.../runtime/client/JobSubmissionException.java | 37 +++
.../runtime/client/JobTimeoutException.java | 10 +-
.../runtime/executiongraph/ExecutionGraph.java | 2 +-
.../apache/flink/runtime/client/JobClient.scala | 40 ++--
.../flink/runtime/jobmanager/JobManager.scala | 68 +++---
.../runtime/messages/JobmanagerMessages.scala | 38 +---
.../runtime/taskmanager/TaskManagerTest.java | 2 -
.../jobmanager/CoLocationConstraintITCase.scala | 6 +-
.../runtime/jobmanager/JobManagerITCase.scala | 92 ++++++--
.../runtime/jobmanager/RecoveryITCase.scala | 10 +-
.../runtime/jobmanager/SlotSharingITCase.scala | 7 +-
.../TaskManagerFailsWithSlotSharingITCase.scala | 25 +-
.../testingUtils/TestingJobManager.scala | 1 +
.../JobSubmissionFailsITCase.java | 227 +++++++++++++++++++
.../taskmanager/TaskManagerFailsITCase.scala | 32 ++-
19 files changed, 487 insertions(+), 172 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-clients/src/main/java/org/apache/flink/client/program/Client.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/Client.java b/flink-clients/src/main/java/org/apache/flink/client/program/Client.java
index 1fc7696..bd364ac 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/Client.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/Client.java
@@ -45,13 +45,9 @@ import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.runtime.client.JobCancellationException;
import org.apache.flink.runtime.client.JobClient;
import org.apache.flink.runtime.client.JobExecutionException;
-import org.apache.flink.runtime.client.JobTimeoutException;
import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.messages.JobManagerMessages.SubmissionFailure;
-import org.apache.flink.runtime.messages.JobManagerMessages.SubmissionResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -342,24 +338,12 @@ public class Client {
return JobClient.submitJobAndWait(jobGraph, printStatusDuringExecution, client, timeout);
}
else {
- SubmissionResponse response = JobClient.submitJobDetached(jobGraph, client, timeout);
- if (response instanceof SubmissionFailure) {
- SubmissionFailure failure = (SubmissionFailure) response;
- throw new ProgramInvocationException(
- "Failed to submit the job to the JobManager.", failure.cause());
- }
+ JobClient.submitJobDetached(jobGraph, client, timeout);
}
} catch (JobExecutionException e) {
throw new ProgramInvocationException("The program execution failed.", e);
- } catch (JobTimeoutException e) {
- throw new ProgramInvocationException("Lost connection to the JobManager.", e);
- } catch (JobCancellationException e) {
- throw new ProgramInvocationException("The program has been canceled.", e);
- } catch (ProgramInvocationException e) {
- // forward exception resulting from submission failure
- throw e;
} catch (Exception e) {
- throw new ProgramInvocationException("Exception occurred during job execution.", e);
+ throw new ProgramInvocationException("Unexpected exception while program execution.", e);
}
finally {
actorSystem.shutdown();
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
index b5b3ac6..f68a7ae 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.client.program;
import akka.actor.ActorSystem;
import akka.actor.Props;
+import akka.actor.Status;
import akka.actor.UntypedActor;
import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.JobExecutionResult;
@@ -36,7 +37,6 @@ import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobID;
import org.apache.flink.runtime.jobmanager.JobManager;
-import org.apache.flink.runtime.messages.JobManagerMessages;
import org.apache.flink.runtime.net.NetUtils;
import org.junit.After;
@@ -174,7 +174,7 @@ public class ClientTest {
// bam!
}
catch (Exception e) {
- fail("wrong exception");
+ fail("wrong exception " + e);
}
verify(this.compilerMock, times(1)).compile(any(Plan.class));
@@ -225,7 +225,7 @@ public class ClientTest {
@Override
public void onReceive(Object message) throws Exception {
- getSender().tell(new JobManagerMessages.SubmissionSuccess(new JobID()), getSelf());
+ getSender().tell(new Status.Success(new JobID()), getSelf());
}
}
@@ -233,7 +233,7 @@ public class ClientTest {
@Override
public void onReceive(Object message) throws Exception {
- getSender().tell(new JobManagerMessages.SubmissionFailure(new JobID(), new Exception("test")), getSelf());
+ getSender().tell(new Status.Failure(new Exception("test")), getSelf());
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobCancellationException.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobCancellationException.java b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobCancellationException.java
index 1a9a1d0..9f72db0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobCancellationException.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobCancellationException.java
@@ -18,13 +18,17 @@
package org.apache.flink.runtime.client;
+import org.apache.flink.runtime.jobgraph.JobID;
+
/**
* An exception which is thrown by the JobClient if a job is aborted as a result of a user
* cancellation.
*/
-public class JobCancellationException extends Exception {
+public class JobCancellationException extends JobExecutionException {
+
+ private static final long serialVersionUID = 2818087325120827526L;
- public JobCancellationException(final String msg, final Throwable cause){
- super(msg, cause);
+ public JobCancellationException(final JobID jobID, final String msg, final Throwable cause){
+ super(jobID, msg, cause);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobExecutionException.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobExecutionException.java b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobExecutionException.java
index d5eb492..1a56a93 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobExecutionException.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobExecutionException.java
@@ -18,6 +18,8 @@
package org.apache.flink.runtime.client;
+import org.apache.flink.runtime.jobgraph.JobID;
+
/**
* This exception is thrown by the {@link JobClient} if a job has been aborted as a result of an
* error which occurred during the execution.
@@ -26,13 +28,33 @@ public class JobExecutionException extends Exception {
private static final long serialVersionUID = 2818087325120827525L;
+ private JobID jobID;
+
/**
* Constructs a new job execution exception.
*
* @param msg The cause for the execution exception.
* @param cause The cause of the exception
*/
- public JobExecutionException(String msg, Throwable cause) {
+ public JobExecutionException(final JobID jobID, final String msg, final Throwable cause) {
super(msg, cause);
+
+ this.jobID = jobID;
+ }
+
+ public JobExecutionException(final JobID jobID, final String msg) {
+ super(msg);
+
+ this.jobID = jobID;
+ }
+
+ public JobExecutionException(final JobID jobID, final Throwable cause) {
+ super(cause);
+
+ this.jobID = jobID;
+ }
+
+ public JobID getJobID() {
+ return jobID;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobSubmissionException.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobSubmissionException.java b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobSubmissionException.java
new file mode 100644
index 0000000..3d672a5
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobSubmissionException.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.client;
+
+import org.apache.flink.runtime.jobgraph.JobID;
+
+/**
+ * This exception denotes an error while submitting a job to the JobManager
+ */
+public class JobSubmissionException extends JobExecutionException {
+
+ private static final long serialVersionUID = 2818087325120827526L;
+
+ public JobSubmissionException(final JobID jobID, final String msg, final Throwable cause) {
+ super(jobID, msg, cause);
+ }
+
+ public JobSubmissionException(final JobID jobID, final String msg) {
+ super(jobID, msg);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobTimeoutException.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobTimeoutException.java b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobTimeoutException.java
index 2bd6ec5..10ef601 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobTimeoutException.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobTimeoutException.java
@@ -18,12 +18,16 @@
package org.apache.flink.runtime.client;
+import org.apache.flink.runtime.jobgraph.JobID;
+
/**
* An exception which is thrown by the JobClient if the job manager is no longer reachable.
*/
-public class JobTimeoutException extends Exception {
+public class JobTimeoutException extends JobExecutionException {
+
+ private static final long serialVersionUID = 2818087325120827529L;
- public JobTimeoutException(final String msg, final Throwable cause) {
- super(msg, cause);
+ public JobTimeoutException(final JobID jobID, final String msg, final Throwable cause) {
+ super(jobID, msg, cause);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
index 7198076..f7b13fd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
@@ -406,7 +406,7 @@ public class ExecutionGraph implements Serializable {
}
} else {
// set the state of the job to failed
- transitionState(current, JobStatus.FAILED, t);
+ transitionState(JobStatus.FAILING, JobStatus.FAILED, t);
}
return;
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-runtime/src/main/scala/org/apache/flink/runtime/client/JobClient.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/client/JobClient.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/client/JobClient.scala
index 4a64f87..d3ceeaf 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/client/JobClient.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/client/JobClient.scala
@@ -21,14 +21,14 @@ package org.apache.flink.runtime.client
import java.io.IOException
import java.net.{InetAddress, InetSocketAddress}
-import akka.actor.Status.Failure
+import akka.actor.Status.{Success, Failure}
import akka.actor._
import akka.pattern.{Patterns, ask}
import org.apache.flink.api.common.JobExecutionResult
import org.apache.flink.configuration.{ConfigConstants, Configuration}
import org.apache.flink.runtime.ActorLogMessages
import org.apache.flink.runtime.akka.AkkaUtils
-import org.apache.flink.runtime.jobgraph.JobGraph
+import org.apache.flink.runtime.jobgraph.{JobID, JobGraph}
import org.apache.flink.runtime.jobmanager.JobManager
import org.apache.flink.runtime.messages.JobClientMessages.{SubmitJobDetached, SubmitJobAndWait}
import org.apache.flink.runtime.messages.JobManagerMessages._
@@ -82,24 +82,16 @@ Actor with ActorLogMessages with ActorLogging {
class JobClientListener(jobSubmitter: ActorRef) extends Actor with ActorLogMessages with
ActorLogging {
override def receiveWithLogMessages: Receive = {
- case SubmissionFailure(jobID, t) =>
- System.out.println(s"Submission of job with ID $jobID was unsuccessful, " +
- s"because ${t.getMessage}.")
+ case failure: Failure =>
+ jobSubmitter ! failure
+ self ! PoisonPill
- case SubmissionSuccess(_) =>
+ case Success(_) =>
case JobResultSuccess(_, duration, accumulatorResults) =>
jobSubmitter ! new JobExecutionResult(duration, accumulatorResults)
self ! PoisonPill
- case JobResultCanceled(_, t) =>
- jobSubmitter ! Failure(new JobCancellationException("The job has been cancelled.", t))
- self ! PoisonPill
-
- case JobResultFailed(_, t) =>
- jobSubmitter ! Failure(new JobExecutionException("The job execution failed.", t))
- self ! PoisonPill
-
case msg =>
// we have to use System.out.println here to avoid erroneous behavior for output redirection
System.out.println(msg.toString)
@@ -211,7 +203,7 @@ object JobClient {
* execution fails.
* @return The job execution result
*/
- @throws(classOf[Exception])
+ @throws(classOf[JobExecutionException])
def submitJobAndWait(jobGraph: JobGraph, listenToStatusEvents: Boolean, jobClient: ActorRef)
(implicit timeout: FiniteDuration): JobExecutionResult = {
@@ -233,7 +225,8 @@ object JobClient {
Await.result(jmStatus, timeout)
} catch {
case t: Throwable =>
- throw new JobTimeoutException("Lost connection to job manager.", t)
+ throw new JobTimeoutException(jobGraph.getJobID, "Lost connection to " +
+ "job manager.", t)
}
}
}
@@ -244,26 +237,25 @@ object JobClient {
/**
* Submits a job in detached mode. The method sends the corresponding [[JobGraph]] to the
* JobClient specified by jobClient. The JobClient does not start a [[JobClientListener]] and
- * simply returns the [[SubmissionResponse]] of the [[JobManager]]. The SubmissionResponse is
- * then returned by this method.
+ * simply returns a possible failure on the [[JobManager]].
*
* @param jobGraph Flink job
* @param jobClient ActorRef to the JobClient
* @param timeout Tiemout for futures
* @return The submission response
*/
- @throws(classOf[Exception])
+ @throws(classOf[JobExecutionException])
def submitJobDetached(jobGraph: JobGraph, jobClient: ActorRef)(implicit timeout: FiniteDuration):
- SubmissionResponse = {
+ Unit = {
+
val response = (jobClient ? SubmitJobDetached(jobGraph))(timeout)
try {
- Await.result(response.mapTo[SubmissionResponse], timeout)
+ Await.result(response, timeout)
} catch {
case timeout: TimeoutException =>
- throw new JobTimeoutException("Timeout while waiting for the submission result.", timeout);
- case t: Throwable =>
- throw new JobExecutionException("Exception while waiting for the submission result.", t)
+ throw new JobTimeoutException(jobGraph.getJobID,
+ "Timeout while waiting for the submission result.", timeout);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
index cb8acae..1399bce 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
@@ -21,10 +21,12 @@ package org.apache.flink.runtime.jobmanager
import java.io.{IOException, File}
import java.net.InetSocketAddress
-import akka.actor.Status.Failure
+import akka.actor.Status.{Success, Failure}
import org.apache.flink.configuration.{ConfigConstants, GlobalConfiguration, Configuration}
import org.apache.flink.core.io.InputSplitAssigner
import org.apache.flink.runtime.blob.BlobServer
+import org.apache.flink.runtime.client.{JobSubmissionException, JobExecutionException,
+JobCancellationException}
import org.apache.flink.runtime.executiongraph.{ExecutionJobVertex, ExecutionGraph}
import org.apache.flink.runtime.jobmanager.web.WebInfoServer
import org.apache.flink.runtime.messages.ArchiveMessages.ArchiveExecutionGraph
@@ -254,31 +256,32 @@ class JobManager(val configuration: Configuration,
sender ! NextInputSplit(serializedInputSplit)
- case JobStatusChanged(jobID, newJobStatus, timeStamp, optionalMessage) =>
+ case JobStatusChanged(jobID, newJobStatus, timeStamp, error) =>
currentJobs.get(jobID) match {
case Some((executionGraph, jobInfo)) => executionGraph.getJobName
log.info("Status of job {} ({}) changed to {} {}.",
jobID, executionGraph.getJobName, newJobStatus,
- if(optionalMessage == null) "" else optionalMessage.getMessage)
+ if(error == null) "" else error.getMessage)
if(newJobStatus.isTerminalState) {
jobInfo.end = timeStamp
- // is the client waiting for the job result?
- if(!jobInfo.detached) {
- newJobStatus match {
- case JobStatus.FINISHED =>
- val accumulatorResults = accumulatorManager.getJobAccumulatorResults(jobID)
- jobInfo.client ! JobResultSuccess(jobID, jobInfo.duration, accumulatorResults)
- case JobStatus.CANCELED =>
- jobInfo.client ! JobResultCanceled(jobID, optionalMessage)
- case JobStatus.FAILED =>
- jobInfo.client ! JobResultFailed(jobID, optionalMessage)
- case x =>
- val exception = new IllegalStateException(s"$x is not a terminal state.")
- jobInfo.client ! JobResultFailed(jobID, exception)
- throw exception
- }
+ // is the client waiting for the job result?
+ newJobStatus match {
+ case JobStatus.FINISHED if !jobInfo.detached =>
+ val accumulatorResults = accumulatorManager.getJobAccumulatorResults(jobID)
+ jobInfo.client ! JobResultSuccess(jobID, jobInfo.duration, accumulatorResults)
+ case JobStatus.CANCELED =>
+ jobInfo.client ! Failure(new JobCancellationException(jobID,
+ "Job was cancelled.", error))
+ case JobStatus.FAILED =>
+ jobInfo.client ! Failure(new JobExecutionException(jobID,
+ "Job execution failed.", error))
+ case x =>
+ val exception = new JobExecutionException(jobID, s"$x is not a " +
+ "terminal state.")
+ jobInfo.client ! Failure(exception)
+ throw exception
}
removeJob(jobID)
@@ -381,14 +384,12 @@ class JobManager(val configuration: Configuration,
private def submitJob(jobGraph: JobGraph, listenToEvents: Boolean, detached: Boolean): Unit = {
try {
if (jobGraph == null) {
- sender ! akka.actor.Status.Failure(new IllegalArgumentException("JobGraph must not be" +
- " null."))
+ sender ! Failure(new JobSubmissionException(null, "JobGraph must not be null."))
} else {
log.info(s"Received job ${jobGraph.getJobID} (${jobGraph.getName}).")
if (jobGraph.getNumberOfVertices == 0) {
- sender ! SubmissionFailure(jobGraph.getJobID, new IllegalArgumentException("Job is " +
- "empty."))
+ sender ! Failure(new JobSubmissionException(jobGraph.getJobID ,"Job is empty."))
} else {
// Create the user code class loader
libraryCacheManager.registerJob(jobGraph.getJobID, jobGraph.getUserJarBlobKeys)
@@ -415,10 +416,8 @@ class JobManager(val configuration: Configuration,
}
if (log.isDebugEnabled) {
- log.debug(s"Running master initialization of job ${jobGraph.getJobID} (${
- jobGraph
- .getName
- }}).")
+ log.debug(s"Running master initialization of job ${jobGraph.getJobID} " +
+ s"(${jobGraph.getName}).")
}
for (vertex <- jobGraph.getVertices.asScala) {
@@ -464,32 +463,31 @@ class JobManager(val configuration: Configuration,
executionGraph.scheduleForExecution(scheduler)
- sender ! SubmissionSuccess(jobGraph.getJobID)
+ sender ! Success(jobGraph.getJobID)
}
}
} catch {
case t: Throwable =>
- log.error(t, "Job submission failed.")
+ log.error(t, "Job submission of job {} failed.", jobGraph.getJobID)
currentJobs.get(jobGraph.getJobID) match {
case Some((executionGraph, jobInfo)) =>
/*
- * Register self to be notified about job status changes in case that it did not happen
- * before. That way the proper cleanup of the job is triggered in the JobStatusChanged
- * handler.
- */
+ * Register self to be notified about job status changes in case that it did not happen
+ * before. That way the proper cleanup of the job is triggered in the JobStatusChanged
+ * handler.
+ */
if (!executionGraph.containsJobStatusListener(self)) {
executionGraph.registerJobStatusListener(self)
}
+ // let the execution graph fail, which will send a failure to the job client
executionGraph.fail(t)
-
case None =>
libraryCacheManager.unregisterJob(jobGraph.getJobID)
currentJobs.remove(jobGraph.getJobID)
+ sender ! Failure(t)
}
-
- sender ! SubmissionFailure(jobGraph.getJobID, t)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobmanagerMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobmanagerMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobmanagerMessages.scala
index f25083f..8aef552 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobmanagerMessages.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobmanagerMessages.scala
@@ -179,10 +179,6 @@ object JobManagerMessages {
*/
case class RequestFinalJobStatus(jobID: JobID)
- sealed trait JobResult{
- def jobID: JobID
- }
-
/**
* Denotes a successful job execution.
*
@@ -191,39 +187,7 @@ object JobManagerMessages {
* @param accumulatorResults
*/
case class JobResultSuccess(jobID: JobID, runtime: Long, accumulatorResults: java.util.Map[String,
- AnyRef]) extends JobResult {}
-
- /**
- * Denotes a cancellation of the job.
- * @param jobID
- * @param t
- */
- case class JobResultCanceled(jobID: JobID, t: Throwable) extends JobResult
-
- /**
- * Denotes a failed job execution.
- * @param jobID
- * @param t
- */
- case class JobResultFailed(jobID: JobID, t: Throwable) extends JobResult
-
- sealed trait SubmissionResponse{
- def jobID: JobID
- }
-
- /**
- * Denotes a successful job submission.
- * @param jobID
- */
- case class SubmissionSuccess(jobID: JobID) extends SubmissionResponse
-
- /**
- * Denotes a failed job submission. The cause of the failure is denoted by [[cause]].
- *
- * @param jobID
- * @param cause of the submission failure
- */
- case class SubmissionFailure(jobID: JobID, cause: Throwable) extends SubmissionResponse
+ AnyRef]) {}
sealed trait CancellationResponse{
def jobID: JobID
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/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 a83886c..e6a487d 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
@@ -55,8 +55,6 @@ import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
-import scala.None;
-import scala.None$;
import scala.Option;
import scala.concurrent.Await;
import scala.concurrent.Future;
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/CoLocationConstraintITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/CoLocationConstraintITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/CoLocationConstraintITCase.scala
index 2104034..28be994 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/CoLocationConstraintITCase.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/CoLocationConstraintITCase.scala
@@ -19,13 +19,13 @@
package org.apache.flink.runtime.jobmanager
import akka.actor.ActorSystem
+import akka.actor.Status.Success
import akka.testkit.{ImplicitSender, TestKit}
import org.apache.flink.runtime.jobgraph.{JobGraph, DistributionPattern,
AbstractJobVertex}
import org.apache.flink.runtime.jobmanager.Tasks.{Receiver, Sender}
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup
-import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultSuccess, SubmissionSuccess,
-SubmitJob}
+import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultSuccess, SubmitJob}
import org.apache.flink.runtime.testingUtils.TestingUtils
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner
@@ -70,7 +70,7 @@ ImplicitSender with WordSpecLike with Matchers with BeforeAndAfterAll with WrapA
try {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
expectMsgType[JobResultSuccess]
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala
index 89d5d43..7e2840a 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala
@@ -20,9 +20,11 @@ package org.apache.flink.runtime.jobmanager
import Tasks._
import akka.actor.ActorSystem
+import akka.actor.Status.{Success, Failure}
import akka.pattern.ask
import akka.testkit.{ImplicitSender, TestKit}
import akka.util.Timeout
+import org.apache.flink.runtime.client.JobExecutionException
import org.apache.flink.runtime.jobgraph.{AbstractJobVertex, DistributionPattern, JobGraph, ScheduleMode}
import org.apache.flink.runtime.messages.JobManagerMessages._
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages.NotifyWhenJobRemoved
@@ -70,9 +72,14 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
within(1 second) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionFailure(jobGraph.getJobID, new NoResourceAvailableException(1,1,0)))
+ val failure = expectMsgType[Failure]
- expectMsg(JobResultFailed(jobGraph.getJobID, new NoResourceAvailableException(1,1,0)))
+ failure.cause match {
+ case e: JobExecutionException =>
+ jobGraph.getJobID should equal(e.getJobID)
+ new NoResourceAvailableException(1,1,0) should equal(e.getCause)
+ case e => fail(s"Received wrong exception of type $e.")
+ }
}
jm ! NotifyWhenJobRemoved(jobGraph.getJobID)
@@ -103,7 +110,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
val result = expectMsgType[JobResultSuccess]
result.jobID should equal(jobGraph.getJobID)
@@ -133,7 +140,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
val result = expectMsgType[JobResultSuccess]
@@ -168,7 +175,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
val result = expectMsgType[JobResultSuccess]
@@ -203,7 +210,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
expectMsgType[JobResultSuccess]
}
@@ -240,8 +247,15 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
- expectMsgType[JobResultFailed]
+ expectMsg(Success(jobGraph.getJobID))
+ val failure = expectMsgType[Failure]
+
+ failure.cause match {
+ case e: JobExecutionException =>
+ jobGraph.getJobID should equal(e.getJobID)
+
+ case e => fail(s"Received wrong exception $e.")
+ }
}
jm ! NotifyWhenJobRemoved(jobGraph.getJobID)
@@ -276,7 +290,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
try {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
expectMsgType[JobResultSuccess]
}
@@ -321,7 +335,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
expectMsgType[JobResultSuccess]
@@ -359,8 +373,16 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
- expectMsgType[JobResultFailed]
+ expectMsg(Success(jobGraph.getJobID))
+
+ val failure = expectMsgType[Failure]
+
+ failure.cause match {
+ case e: JobExecutionException =>
+ jobGraph.getJobID should equal(e.getJobID)
+
+ case e => fail(s"Received wrong exception $e.")
+ }
}
jm ! NotifyWhenJobRemoved(jobGraph.getJobID)
@@ -399,8 +421,15 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
- expectMsgType[JobResultFailed]
+ expectMsg(Success(jobGraph.getJobID))
+ val failure = expectMsgType[Failure]
+
+ failure.cause match {
+ case e: JobExecutionException =>
+ jobGraph.getJobID should equal(e.getJobID)
+
+ case e => fail(s"Received wrong exception $e.")
+ }
}
jm ! NotifyWhenJobRemoved(jobGraph.getJobID)
@@ -431,8 +460,15 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
try {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
- expectMsgType[JobResultFailed]
+ expectMsg(Success(jobGraph.getJobID))
+ val failure = expectMsgType[Failure]
+
+ failure.cause match {
+ case e: JobExecutionException =>
+ jobGraph.getJobID should equal(e.getJobID)
+
+ case e => fail(s"Received wrong exception $e.")
+ }
}
jm ! NotifyWhenJobRemoved(jobGraph.getJobID)
@@ -466,8 +502,15 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
expectMsg(num_tasks)
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
- expectMsgType[JobResultFailed]
+ expectMsg(Success(jobGraph.getJobID))
+ val failure = expectMsgType[Failure]
+
+ failure.cause match {
+ case e: JobExecutionException =>
+ jobGraph.getJobID should equal(e.getJobID)
+
+ case e => fail(s"Received wrong exception $e.")
+ }
}
jm ! NotifyWhenJobRemoved(jobGraph.getJobID)
@@ -506,8 +549,15 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
expectMsg(num_tasks)
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
- expectMsgType[JobResultFailed]
+ expectMsg(Success(jobGraph.getJobID))
+ val failure = expectMsgType[Failure]
+
+ failure.cause match {
+ case e: JobExecutionException =>
+ jobGraph.getJobID should equal(e.getJobID)
+
+ case e => fail(s"Received wrong exception $e.")
+ }
}
jm ! NotifyWhenJobRemoved(jobGraph.getJobID)
@@ -539,7 +589,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
within(TestingUtils.TESTING_DURATION){
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
expectMsgType[JobResultSuccess]
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
index a7be14b..2c1f82f 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
@@ -18,14 +18,14 @@
package org.apache.flink.runtime.jobmanager
+import akka.actor.Status.Success
import akka.actor.{ActorRef, PoisonPill, ActorSystem}
import akka.testkit.{ImplicitSender, TestKit}
import org.apache.flink.runtime.jobgraph.{JobStatus, JobGraph, DistributionPattern,
AbstractJobVertex}
import org.apache.flink.runtime.jobmanager.Tasks.{BlockingOnceReceiver, FailingOnceReceiver}
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup
-import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultSuccess, SubmissionSuccess,
-SubmitJob}
+import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultSuccess, SubmitJob}
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages._
import org.apache.flink.runtime.testingUtils.TestingUtils
import org.junit.runner.RunWith
@@ -68,7 +68,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
within(TestingUtils.TESTING_DURATION){
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
val result = expectMsgType[JobResultSuccess]
@@ -111,7 +111,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
within(TestingUtils.TESTING_DURATION){
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
val result = expectMsgType[JobResultSuccess]
@@ -155,7 +155,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
within(TestingUtils.TESTING_DURATION){
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
jm ! WaitForAllVerticesToBeRunningOrFinished(jobGraph.getJobID)
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/SlotSharingITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/SlotSharingITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/SlotSharingITCase.scala
index 289f759..d719dc3 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/SlotSharingITCase.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/SlotSharingITCase.scala
@@ -19,11 +19,12 @@
package org.apache.flink.runtime.jobmanager
import akka.actor.ActorSystem
+import akka.actor.Status.Success
import akka.testkit.{ImplicitSender, TestKit}
import org.apache.flink.runtime.jobgraph.{AbstractJobVertex, DistributionPattern, JobGraph}
import org.apache.flink.runtime.jobmanager.Tasks.{Sender, AgnosticBinaryReceiver, Receiver}
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup
-import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultSuccess, SubmissionSuccess, SubmitJob}
+import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultSuccess, SubmitJob}
import org.apache.flink.runtime.testingUtils.TestingUtils
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner
@@ -65,7 +66,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
try {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
expectMsgType[JobResultSuccess]
}
@@ -108,7 +109,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
try {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
expectMsgType[JobResultSuccess]
}
} finally {
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/TaskManagerFailsWithSlotSharingITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/TaskManagerFailsWithSlotSharingITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/TaskManagerFailsWithSlotSharingITCase.scala
index 626c518..dcfc899 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/TaskManagerFailsWithSlotSharingITCase.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/TaskManagerFailsWithSlotSharingITCase.scala
@@ -18,12 +18,14 @@
package org.apache.flink.runtime.jobmanager
+import akka.actor.Status.{Failure, Success}
import akka.actor.{Kill, ActorSystem, PoisonPill}
import akka.testkit.{ImplicitSender, TestKit}
+import org.apache.flink.runtime.client.JobExecutionException
import org.apache.flink.runtime.jobgraph.{AbstractJobVertex, DistributionPattern, JobGraph}
import org.apache.flink.runtime.jobmanager.Tasks.{BlockingReceiver, Sender}
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup
-import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultFailed, SubmissionSuccess, SubmitJob}
+import org.apache.flink.runtime.messages.JobManagerMessages.SubmitJob
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages._
import org.apache.flink.runtime.testingUtils.TestingUtils
import org.junit.runner.RunWith
@@ -68,7 +70,7 @@ ImplicitSender with WordSpecLike with Matchers with BeforeAndAfterAll {
try{
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
jm ! WaitForAllVerticesToBeRunningOrFinished(jobID)
@@ -77,7 +79,13 @@ ImplicitSender with WordSpecLike with Matchers with BeforeAndAfterAll {
//kill task manager
taskManagers(0) ! PoisonPill
- expectMsgType[JobResultFailed]
+ val failure = expectMsgType[Failure]
+
+ failure.cause match {
+ case e: JobExecutionException =>
+ jobGraph.getJobID should equal(e.getJobID)
+ case e => fail(s"Received wrong exception $e.")
+ }
}
}finally{
cluster.stop()
@@ -111,7 +119,7 @@ ImplicitSender with WordSpecLike with Matchers with BeforeAndAfterAll {
try{
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
jm ! WaitForAllVerticesToBeRunningOrFinished(jobID)
expectMsg(AllVerticesRunning(jobID))
@@ -119,7 +127,14 @@ ImplicitSender with WordSpecLike with Matchers with BeforeAndAfterAll {
//kill task manager
taskManagers(0) ! Kill
- expectMsgType[JobResultFailed]
+ val failure = expectMsgType[Failure]
+
+ failure.cause match {
+ case e: JobExecutionException =>
+ jobGraph.getJobID should equal(e.getJobID)
+
+ case e => fail(s"Received wrong exception $e.")
+ }
}
}finally{
cluster.stop()
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
index 15c55ea..566f661 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
@@ -114,6 +114,7 @@ trait TestingJobManager extends ActorLogMessages with WrapAsScala {
case NotifyWhenTaskManagerTerminated(taskManager) =>
val waiting = waitForTaskManagerToBeTerminated.getOrElse(taskManager.path.name, Set())
waitForTaskManagerToBeTerminated += taskManager.path.name -> (waiting + sender)
+
case msg@Terminated(taskManager) =>
super.receiveWithLogMessages(msg)
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/flink-tests/src/test/java/org/apache/flink/test/failingPrograms/JobSubmissionFailsITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/failingPrograms/JobSubmissionFailsITCase.java b/flink-tests/src/test/java/org/apache/flink/test/failingPrograms/JobSubmissionFailsITCase.java
new file mode 100644
index 0000000..7da4125
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/failingPrograms/JobSubmissionFailsITCase.java
@@ -0,0 +1,227 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.failingPrograms;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.testkit.JavaTestKit;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.client.JobClient;
+import org.apache.flink.runtime.client.JobExecutionException;
+import org.apache.flink.runtime.client.JobSubmissionException;
+import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmanager.Tasks;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.test.util.ForkableFlinkMiniCluster;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+@RunWith(Parameterized.class)
+public class JobSubmissionFailsITCase {
+
+ private static ActorSystem system;
+
+ private static JobGraph workingJobGraph;
+
+ @BeforeClass
+ public static void setup() {
+ system = ActorSystem.create("TestingActorSystem", AkkaUtils.getDefaultAkkaConfig());
+
+ final AbstractJobVertex jobVertex = new AbstractJobVertex("Working job vertex.");
+ jobVertex.setInvokableClass(Tasks.NoOpInvokable.class);
+
+ workingJobGraph = new JobGraph("Working testing job", jobVertex);
+ }
+
+ @AfterClass
+ public static void teardown() {
+ JavaTestKit.shutdownActorSystem(system);
+ system = null;
+ }
+
+ private boolean detached;
+
+ public JobSubmissionFailsITCase(boolean detached) {
+ this.detached = detached;
+ }
+
+ @Parameterized.Parameters(name = "Detached mode = {0}")
+ public static Collection<Boolean[]> executionModes(){
+ return Arrays.asList(new Boolean[]{false},
+ new Boolean[]{true});
+ }
+
+ private JobExecutionResult submitJob(JobGraph jobGraph, ActorRef jobClient) throws Exception {
+ if(detached) {
+ JobClient.submitJobDetached(jobGraph, jobClient, TestingUtils.TESTING_DURATION());
+ return null;
+ } else {
+ return JobClient.submitJobAndWait(jobGraph, false, jobClient, TestingUtils.TESTING_DURATION());
+ }
+ }
+
+ @Test
+ public void testExceptionInInitializeOnMaster() {
+ new JavaTestKit(system) {{
+ final int numSlots = 20;
+
+ final ForkableFlinkMiniCluster cluster =
+ ForkableFlinkMiniCluster.startCluster(numSlots/2, 2,
+ TestingUtils.TESTING_DURATION().toString());
+
+ final ActorRef jobClient = cluster.getJobClient();
+
+ final AbstractJobVertex failingJobVertex = new FailingJobVertex("Failing job vertex");
+ failingJobVertex.setInvokableClass(Tasks.NoOpInvokable.class);
+
+ final JobGraph failingJobGraph = new JobGraph("Failing testing job", failingJobVertex);
+
+ try {
+ new Within(TestingUtils.TESTING_DURATION()) {
+
+ @Override
+ protected void run() {
+ try {
+ submitJob(failingJobGraph, jobClient);
+ fail("Expected JobExecutionException.");
+ } catch (JobExecutionException e) {
+ assertEquals("Test exception.", e.getCause().getMessage());
+ } catch (Throwable t) {
+ fail("Caught wrong exception of type " + t.getClass() + ".");
+ t.printStackTrace();
+ }
+
+ try {
+ JobClient.submitJobAndWait(workingJobGraph, false, jobClient,
+ TestingUtils.TESTING_DURATION());
+ } catch (Throwable t) {
+ fail("Caught unexpected exception " + t.getMessage() + ".");
+ }
+ }
+ };
+ } finally {
+ cluster.stop();
+ }
+ }};
+ }
+
+ @Test
+ public void testSubmitEmptyJobGraph() {
+ new JavaTestKit(system) {{
+ final int numSlots = 20;
+
+ final ForkableFlinkMiniCluster cluster =
+ ForkableFlinkMiniCluster.startCluster(numSlots/2, 2,
+ TestingUtils.TESTING_DURATION().toString());
+
+ final ActorRef jobClient = cluster.getJobClient();
+
+ final JobGraph jobGraph = new JobGraph("Testing job");
+
+ try {
+ new Within(TestingUtils.TESTING_DURATION()) {
+
+ @Override
+ protected void run() {
+ try {
+ submitJob(jobGraph, jobClient);
+ fail("Expected JobSubmissionException.");
+ } catch (JobSubmissionException e) {
+ assertEquals("Job is empty.", e.getMessage());
+ } catch (Throwable t) {
+ fail("Caught wrong exception of type " + t.getClass() + ".");
+ t.printStackTrace();
+ }
+
+ try {
+ JobClient.submitJobAndWait(workingJobGraph, false, jobClient,
+ TestingUtils.TESTING_DURATION());
+ } catch (Throwable t) {
+ fail("Caught unexpected exception " + t.getMessage() + ".");
+ }
+ }
+ };
+ } finally {
+ cluster.stop();
+ }
+ }};
+ }
+
+ @Test
+ public void testSubmitNullJobGraph() {
+ new JavaTestKit(system) {{
+ final int numSlots = 20;
+
+ final ForkableFlinkMiniCluster cluster =
+ ForkableFlinkMiniCluster.startCluster(numSlots/2, 2,
+ TestingUtils.TESTING_DURATION().toString());
+
+ final ActorRef jobClient = cluster.getJobClient();
+
+ try {
+ new Within(TestingUtils.TESTING_DURATION()) {
+
+ @Override
+ protected void run() {
+ try {
+ submitJob(null, jobClient);
+ fail("Expected JobSubmissionException.");
+ } catch (JobSubmissionException e) {
+ assertEquals("JobGraph must not be null.", e.getMessage());
+ } catch (Throwable t) {
+ fail("Caught wrong exception of type " + t.getClass() + ".");
+ t.printStackTrace();
+ }
+
+ try {
+ JobClient.submitJobAndWait(workingJobGraph, false, jobClient,
+ TestingUtils.TESTING_DURATION());
+ } catch (Throwable t) {
+ fail("Caught unexpected exception " + t.getMessage() + ".");
+ }
+ }
+ };
+ } finally {
+ cluster.stop();
+ }
+ }};
+ }
+
+ public static class FailingJobVertex extends AbstractJobVertex {
+ public FailingJobVertex(final String msg) {
+ super(msg);
+ }
+
+ @Override
+ public void initializeOnMaster(ClassLoader loader) throws Exception {
+ throw new Exception("Test exception.");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/4ff91cd7/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 a6ee119..60c67fb 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
@@ -18,15 +18,17 @@
package org.apache.flink.api.scala.runtime.taskmanager
+import akka.actor.Status.{Failure, Success}
import akka.actor.{ActorSystem, Kill, PoisonPill}
import akka.testkit.{ImplicitSender, TestKit}
import org.apache.flink.configuration.Configuration
import org.apache.flink.configuration.ConfigConstants
import org.apache.flink.runtime.akka.AkkaUtils
+import org.apache.flink.runtime.client.JobExecutionException
import org.apache.flink.runtime.jobgraph.{AbstractJobVertex, DistributionPattern, JobGraph}
import org.apache.flink.runtime.jobmanager.Tasks.{BlockingReceiver, Sender}
-import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultFailed, RequestNumberRegisteredTaskManager, SubmissionSuccess, SubmitJob}
+import org.apache.flink.runtime.messages.JobManagerMessages.{RequestNumberRegisteredTaskManager, SubmitJob}
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages._
import org.apache.flink.runtime.testingUtils.TestingUtils
import org.apache.flink.test.util.ForkableFlinkMiniCluster
@@ -103,7 +105,7 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
try {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
jm ! WaitForAllVerticesToBeRunningOrFinished(jobID)
@@ -114,9 +116,17 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
val tm = expectMsgType[WorkingTaskManager].taskManager
// kill one task manager
tm ! PoisonPill
- expectMsgType[JobResultFailed]
+
+ val failure = expectMsgType[Failure]
+
+ failure.cause match {
+ case e: JobExecutionException =>
+ jobGraph.getJobID should equal(e.getJobID)
+
+ case e => fail(s"Received wrong exception $e.")
+ }
}
- }finally{
+ } finally {
cluster.stop()
}
}
@@ -142,16 +152,24 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
try {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
jm ! WaitForAllVerticesToBeRunningOrFinished(jobID)
expectMsg(AllVerticesRunning(jobID))
// kill one task manager
taskManagers(0) ! Kill
- expectMsgType[JobResultFailed]
+
+ val failure = expectMsgType[Failure]
+
+ failure.cause match {
+ case e: JobExecutionException =>
+ jobGraph.getJobID should equal(e.getJobID)
+
+ case e => fail(s"Received wrong exception $e.")
+ }
}
- }finally{
+ } finally {
cluster.stop()
}
}
[5/9] flink git commit: [tests] Add integration test for restart
recovery
Posted by se...@apache.org.
[tests] Add integration test for restart recovery
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7d80711d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7d80711d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7d80711d
Branch: refs/heads/master
Commit: 7d80711d366c5a2127e50051e6c0d5b347b638c6
Parents: a110449
Author: Stephan Ewen <se...@apache.org>
Authored: Mon Feb 16 21:40:06 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Feb 19 18:54:14 2015 +0100
----------------------------------------------------------------------
.../flink/configuration/ConfigConstants.java | 60 -----
.../apache/flink/runtime/instance/Instance.java | 4 +-
.../flink/runtime/instance/InstanceManager.java | 42 +---
.../flink/runtime/jobmanager/JobManager.scala | 6 +-
.../flink/runtime/taskmanager/TaskManager.scala | 2 +-
.../runtime/instance/InstanceManagerTest.java | 10 +-
.../runtime/testingUtils/TestingUtils.scala | 1 -
.../test/recovery/SimpleRecoveryITCase.java | 252 +++++++++++++++++++
.../TaskManagerFailureRecoveryITCase.java | 184 ++++++++++++++
9 files changed, 456 insertions(+), 105 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7d80711d/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
index 42a3c9a..767648a 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
@@ -56,12 +56,6 @@ public final class ConfigConstants {
public static final String JOB_MANAGER_IPC_PORT_KEY = "jobmanager.rpc.port";
/**
- * The config parameter defining the number of seconds that a task manager heartbeat may be missing before it is
- * marked as failed.
- */
- public static final String JOB_MANAGER_DEAD_TASKMANAGER_TIMEOUT_KEY = "jobmanager.max-heartbeat-delay-before-failure.msecs";
-
- /**
* The config parameter defining the storage directory to be used by the blob server.
*/
public static final String BLOB_STORAGE_DIRECTORY_KEY = "blob.storage.directory";
@@ -136,34 +130,6 @@ public final class ConfigConstants {
public static final String TASK_MANAGER_NUM_TASK_SLOTS = "taskmanager.numberOfTaskSlots";
/**
- * The number of incoming network IO threads (e.g. incoming connection threads used in NettyConnectionManager
- * for the ServerBootstrap.)
- */
- public static final String TASK_MANAGER_NET_NUM_IN_THREADS_KEY = "taskmanager.net.numInThreads";
-
- /**
- * The number of outgoing network IO threads (e.g. outgoing connection threads used in NettyConnectionManager for
- * the Bootstrap.)
- */
- public static final String TASK_MANAGER_NET_NUM_OUT_THREADS_KEY = "taskmanager.net.numOutThreads";
-
- /**
- * The low water mark used in NettyConnectionManager for the Bootstrap.
- */
- public static final String TASK_MANAGER_NET_NETTY_LOW_WATER_MARK = "taskmanager.net.nettyLowWaterMark";
-
- /**
- * The high water mark used in NettyConnectionManager for the Bootstrap.
- */
- public static final String TASK_MANAGER_NET_NETTY_HIGH_WATER_MARK = "taskmanager.net.nettyHighWaterMark";
-
- /**
- * Parameter for the interval in which the TaskManager sends the periodic heart beat messages
- * to the JobManager (in msecs).
- */
- public static final String TASK_MANAGER_HEARTBEAT_INTERVAL_KEY = "taskmanager.heartbeat-interval";
-
- /**
* Flag indicating whether to start a thread, which repeatedly logs the memory usage of the JVM.
*/
public static final String TASK_MANAGER_DEBUG_MEMORY_USAGE_START_LOG_THREAD = "taskmanager.debug.memory.startLogThread";
@@ -473,32 +439,6 @@ public final class ConfigConstants {
public static final int DEFAULT_TASK_MANAGER_NETWORK_BUFFER_SIZE = 32768;
/**
- * Default number of incoming network IO threads (e.g. number of incoming connection threads used in
- * NettyConnectionManager for the ServerBootstrap). If set to -1, a reasonable default depending on the number of
- * cores will be picked.
- */
- public static final int DEFAULT_TASK_MANAGER_NET_NUM_IN_THREADS = -1;
-
- /**
- * Default number of outgoing network IO threads (e.g. number of outgoing connection threads used in
- * NettyConnectionManager for the Bootstrap). If set to -1, a reasonable default depending on the number of cores
- * will be picked.
- */
- public static final int DEFAULT_TASK_MANAGER_NET_NUM_OUT_THREADS = -1;
-
- /**
- * Default low water mark used in NettyConnectionManager for the Bootstrap. If set to -1, NettyConnectionManager
- * will use half of the network buffer size as the low water mark.
- */
- public static final int DEFAULT_TASK_MANAGER_NET_NETTY_LOW_WATER_MARK = -1;
-
- /**
- * Default high water mark used in NettyConnectionManager for the Bootstrap. If set to -1, NettyConnectionManager
- * will use the network buffer size as the high water mark.
- */
- public static final int DEFAULT_TASK_MANAGER_NET_NETTY_HIGH_WATER_MARK = -1;
-
- /**
* Flag indicating whether to start a thread, which repeatedly logs the memory usage of the JVM.
*/
public static final boolean DEFAULT_TASK_MANAGER_DEBUG_MEMORY_USAGE_START_LOG_THREAD = false;
http://git-wip-us.apache.org/repos/asf/flink/blob/7d80711d/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
index a5a9263..4e08389 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
@@ -329,7 +329,7 @@ public class Instance {
@Override
public String toString() {
- return instanceId + " @ " + (taskManager != null ? taskManager.path() : "ActorRef.noSender") + " - " +
- numberOfSlots + " slots" + " - " + hashCode();
+ return String.format("%s @ %s - %d slots - URL: %s", instanceId, connectionInfo.getHostname(),
+ numberOfSlots, (taskManager != null ? taskManager.path() : "ActorRef.noSender"));
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7d80711d/flink-runtime/src/main/java/org/apache/flink/runtime/instance/InstanceManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/InstanceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/InstanceManager.java
index b28da35..2ee41da 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/InstanceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/InstanceManager.java
@@ -27,8 +27,6 @@ import java.util.Map;
import java.util.Set;
import akka.actor.ActorRef;
-import org.apache.flink.configuration.ConfigConstants;
-import org.apache.flink.configuration.GlobalConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -57,9 +55,6 @@ public class InstanceManager {
/** Listeners that want to be notified about availability and disappearance of instances */
private final List<InstanceListener> instanceListeners = new ArrayList<InstanceListener>();
-
- /** Duration after which a task manager is considered dead if it did not send a heart-beat message. */
- private final long heartbeatTimeout;
/** The total number of task slots that the system has */
private int totalNumberOfAliveTaskSlots;
@@ -72,32 +67,12 @@ public class InstanceManager {
// ------------------------------------------------------------------------
/**
- * Creates an instance manager, using the global configuration value for maximum interval between heartbeats
- * where a task manager is still considered alive.
+ * Creates an new instance manager.
*/
public InstanceManager() {
- this(GlobalConfiguration.getLong(
- ConfigConstants.JOB_MANAGER_DEAD_TASKMANAGER_TIMEOUT_KEY,
- ConfigConstants.DEFAULT_JOB_MANAGER_DEAD_TASKMANAGER_TIMEOUT));
- }
-
- public InstanceManager(long heartbeatTimeout) {
- this(heartbeatTimeout, heartbeatTimeout);
- }
-
- public InstanceManager(long heartbeatTimeout, long cleanupInterval) {
- if (heartbeatTimeout <= 0 || cleanupInterval <= 0) {
- throw new IllegalArgumentException("Heartbeat timeout and cleanup interval must be positive.");
- }
-
this.registeredHostsById = new HashMap<InstanceID, Instance>();
this.registeredHostsByConnection = new HashMap<ActorRef, Instance>();
this.deadHosts = new HashSet<ActorRef>();
- this.heartbeatTimeout = heartbeatTimeout;
- }
-
- public long getHeartbeatTimeout() {
- return heartbeatTimeout;
}
public void shutdown() {
@@ -132,14 +107,19 @@ public class InstanceManager {
if (host == null){
if (LOG.isDebugEnabled()) {
- LOG.debug("Received hearbeat from unknown TaskManager with instance ID " + instanceId.toString() +
- " Possibly TaskManager was maked as dead (timed-out) earlier. " +
+ LOG.debug("Received heartbeat from unknown TaskManager with instance ID " + instanceId.toString() +
+ " Possibly TaskManager was marked as dead (timed-out) earlier. " +
"Reporting back that task manager is no longer known.");
}
return false;
}
host.reportHeartBeat();
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Received heartbeat from TaskManager " + host);
+ }
+
return true;
}
}
@@ -164,7 +144,7 @@ public class InstanceManager {
" which was marked as dead earlier because of a heart-beat timeout.");
}
- InstanceID id = null;
+ InstanceID id;
do {
id = new InstanceID();
} while (registeredHostsById.containsKey(id));
@@ -178,8 +158,8 @@ public class InstanceManager {
totalNumberOfAliveTaskSlots += numberOfSlots;
if (LOG.isInfoEnabled()) {
- LOG.info(String.format("Registered TaskManager at %s as %s. Current number of registered hosts is %d.",
- taskManager.path(), id, registeredHostsById.size()));
+ LOG.info(String.format("Registered TaskManager at %s (%s) as %s. Current number of registered hosts is %d.",
+ connectionInfo.getHostname(), taskManager.path(), id, registeredHostsById.size()));
}
host.reportHeartBeat();
http://git-wip-us.apache.org/repos/asf/flink/blob/7d80711d/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
index 1741cdb..cb8acae 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
@@ -723,9 +723,9 @@ object JobManager {
ConfigConstants.DEFAULT_EXECUTION_RETRIES_KEY,
ConfigConstants.DEFAULT_EXECUTION_RETRIES)
- val delayBetweenRetries = 2 * configuration.getLong(
- ConfigConstants.JOB_MANAGER_DEAD_TASKMANAGER_TIMEOUT_KEY,
- ConfigConstants.DEFAULT_JOB_MANAGER_DEAD_TASKMANAGER_TIMEOUT)
+ val delayBetweenRetries = 2 * Duration(configuration.getString(
+ ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE,
+ ConfigConstants.DEFAULT_AKKA_ASK_TIMEOUT)).toMillis
(archiveCount, profilingEnabled, cleanupInterval, executionRetries, delayBetweenRetries)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7d80711d/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 99d824b..c14ebca 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
@@ -433,7 +433,7 @@ import scala.collection.JavaConverters._
if (jobConfig.getBoolean(ProfilingUtils.PROFILE_JOB_KEY, true)) {
profiler match {
case Some(profilerActorRef) => profilerActorRef ! MonitorTask(task)
- case None => log.info("There is no profiling enabled for the task manager.")
+ case None => // no log message here - floods the log
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7d80711d/flink-runtime/src/test/java/org/apache/flink/runtime/instance/InstanceManagerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/InstanceManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/InstanceManagerTest.java
index dff3dd3..b9aa674 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/InstanceManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/InstanceManagerTest.java
@@ -65,10 +65,6 @@ public class InstanceManagerTest{
try {
InstanceManager cm = new InstanceManager();
- // catches error that some parts assumed config values in seconds, others in
- // milliseconds by verifying that the timeout is not larger than 2 minutes.
- assertTrue(cm.getHeartbeatTimeout() < 2 * 60 * 1000);
-
final int dataPort = 20000;
HardwareDescription hardwareDescription = HardwareDescription.extractFromSystem(4096);
@@ -76,7 +72,7 @@ public class InstanceManagerTest{
InetAddress address = InetAddress.getByName("127.0.0.1");
// register three instances
- InstanceConnectionInfo ici1 = new InstanceConnectionInfo(address, dataPort + 0);
+ InstanceConnectionInfo ici1 = new InstanceConnectionInfo(address, dataPort);
InstanceConnectionInfo ici2 = new InstanceConnectionInfo(address, dataPort + 15);
InstanceConnectionInfo ici3 = new InstanceConnectionInfo(address, dataPort + 30);
@@ -121,7 +117,7 @@ public class InstanceManagerTest{
HardwareDescription resources = HardwareDescription.extractFromSystem(4096);
InetAddress address = InetAddress.getByName("127.0.0.1");
- InstanceConnectionInfo ici = new InstanceConnectionInfo(address, dataPort + 0);
+ InstanceConnectionInfo ici = new InstanceConnectionInfo(address, dataPort);
JavaTestKit probe = new JavaTestKit(system);
InstanceID i = cm.registerTaskManager(probe.getRef(), ici, resources, 1);
@@ -157,7 +153,7 @@ public class InstanceManagerTest{
InetAddress address = InetAddress.getByName("127.0.0.1");
// register three instances
- InstanceConnectionInfo ici1 = new InstanceConnectionInfo(address, dataPort + 0);
+ InstanceConnectionInfo ici1 = new InstanceConnectionInfo(address, dataPort);
InstanceConnectionInfo ici2 = new InstanceConnectionInfo(address, dataPort + 1);
InstanceConnectionInfo ici3 = new InstanceConnectionInfo(address, dataPort + 2);
http://git-wip-us.apache.org/repos/asf/flink/blob/7d80711d/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
index 4416ba6..c26b255 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
@@ -101,7 +101,6 @@ object TestingUtils {
val config = new Configuration()
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlots)
config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, numTMs)
- config.setInteger(ConfigConstants.JOB_MANAGER_DEAD_TASKMANAGER_TIMEOUT_KEY, 1000)
config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, timeout)
new TestingCluster(config)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7d80711d/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryITCase.java
new file mode 100644
index 0000000..911edb3
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryITCase.java
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.recovery;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.LocalCollectionOutputFormat;
+import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.client.JobExecutionException;
+import org.apache.flink.test.util.ForkableFlinkMiniCluster;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.Assert.*;
+
+public class SimpleRecoveryITCase {
+
+
+ private static ForkableFlinkMiniCluster cluster;
+
+ @BeforeClass
+ public static void setupCluster() {
+ Configuration config = new Configuration();
+ config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 2);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 2);
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "2 s");
+
+ cluster = new ForkableFlinkMiniCluster(config, false);
+ }
+
+ @AfterClass
+ public static void teardownCluster() {
+ try {
+ cluster.stop();
+ }
+ catch (Throwable t) {
+ System.err.println("Error stopping cluster on shutdown");
+ t.printStackTrace();
+ fail("Cluster shutdown caused an exception: " + t.getMessage());
+ }
+ }
+
+ @Test
+ public void testFailedRunThenSuccessfulRun() {
+
+ FailOnceMapper.failuresBeforeSuccess = 1;
+
+ try {
+ List<Long> resultCollection = new ArrayList<Long>();
+
+ // attempt 1
+ {
+ ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment(
+ "localhost", cluster.getJobManagerRPCPort());
+
+ env.setDegreeOfParallelism(4);
+ env.setNumberOfExecutionRetries(0);
+
+ env.generateSequence(1, 10)
+ .map(new FailOnceMapper<Long>())
+ .reduce(new ReduceFunction<Long>() {
+ @Override
+ public Long reduce(Long value1, Long value2) {
+ return value1 + value2;
+ }
+ })
+ .output(new LocalCollectionOutputFormat<Long>(resultCollection));
+
+ try {
+ env.execute();
+ fail("The program should have failed");
+ }
+ catch (ProgramInvocationException e) {
+ // expected
+ }
+ }
+
+ // attempt 2
+ {
+ ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment(
+ "localhost", cluster.getJobManagerRPCPort());
+
+ env.setDegreeOfParallelism(4);
+ env.setNumberOfExecutionRetries(0);
+
+ env.generateSequence(1, 10)
+ .map(new FailOnceMapper<Long>())
+ .reduce(new ReduceFunction<Long>() {
+ @Override
+ public Long reduce(Long value1, Long value2) {
+ return value1 + value2;
+ }
+ })
+ .output(new LocalCollectionOutputFormat<Long>(resultCollection));
+
+ try {
+ JobExecutionResult result = env.execute();
+ assertTrue(result.getNetRuntime() >= 0);
+ assertNotNull(result.getAllAccumulatorResults());
+ assertTrue(result.getAllAccumulatorResults().isEmpty());
+ }
+ catch (JobExecutionException e) {
+ fail("The program should have succeeded on the second run");
+ }
+
+ long sum = 0;
+ for (long l : resultCollection) {
+ sum += l;
+ }
+ assertEquals(55, sum);
+ }
+
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testRestart() {
+
+ FailOnceMapper.failuresBeforeSuccess = 1;
+
+ try {
+ List<Long> resultCollection = new ArrayList<Long>();
+
+ ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment(
+ "localhost", cluster.getJobManagerRPCPort());
+
+ env.setDegreeOfParallelism(4);
+ env.setNumberOfExecutionRetries(1);
+
+ env.generateSequence(1, 10)
+ .map(new FailOnceMapper<Long>())
+ .reduce(new ReduceFunction<Long>() {
+ @Override
+ public Long reduce(Long value1, Long value2) {
+ return value1 + value2;
+ }
+ })
+ .output(new LocalCollectionOutputFormat<Long>(resultCollection));
+
+ try {
+ JobExecutionResult result = env.execute();
+ assertTrue(result.getNetRuntime() >= 0);
+ assertNotNull(result.getAllAccumulatorResults());
+ assertTrue(result.getAllAccumulatorResults().isEmpty());
+ }
+ catch (JobExecutionException e) {
+ fail("The program should have succeeded on the second run");
+ }
+
+ long sum = 0;
+ for (long l : resultCollection) {
+ sum += l;
+ }
+ assertEquals(55, sum);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testRestartMultipleTimes() {
+
+ FailOnceMapper.failuresBeforeSuccess = 3;
+
+ try {
+ List<Long> resultCollection = new ArrayList<Long>();
+
+ ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment(
+ "localhost", cluster.getJobManagerRPCPort());
+
+ env.setDegreeOfParallelism(4);
+ env.setNumberOfExecutionRetries(3);
+
+ env.generateSequence(1, 10)
+ .map(new FailOnceMapper<Long>())
+ .reduce(new ReduceFunction<Long>() {
+ @Override
+ public Long reduce(Long value1, Long value2) {
+ return value1 + value2;
+ }
+ })
+ .output(new LocalCollectionOutputFormat<Long>(resultCollection));
+
+ try {
+ JobExecutionResult result = env.execute();
+ assertTrue(result.getNetRuntime() >= 0);
+ assertNotNull(result.getAllAccumulatorResults());
+ assertTrue(result.getAllAccumulatorResults().isEmpty());
+ }
+ catch (JobExecutionException e) {
+ fail("The program should have succeeded on the second run");
+ }
+
+ long sum = 0;
+ for (long l : resultCollection) {
+ sum += l;
+ }
+ assertEquals(55, sum);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ // ------------------------------------------------------------------------------------
+
+ private static class FailOnceMapper<T> extends RichMapFunction<T, T> {
+
+ private static int failuresBeforeSuccess = 0;
+
+ @Override
+ public T map(T value) throws Exception {
+ if (failuresBeforeSuccess > 0 && getRuntimeContext().getIndexOfThisSubtask() == 1) {
+ failuresBeforeSuccess--;
+ throw new Exception("Test Failure");
+ }
+
+ return value;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7d80711d/flink-tests/src/test/java/org/apache/flink/test/recovery/TaskManagerFailureRecoveryITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/TaskManagerFailureRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/TaskManagerFailureRecoveryITCase.java
new file mode 100644
index 0000000..85856ba
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/TaskManagerFailureRecoveryITCase.java
@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.recovery;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.PoisonPill;
+import akka.pattern.Patterns;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.LocalCollectionOutputFormat;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.messages.TaskManagerMessages;
+import org.apache.flink.test.util.ForkableFlinkMiniCluster;
+import org.junit.Test;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.*;
+
+public class TaskManagerFailureRecoveryITCase {
+
+ @Test
+ public void testRestartWithFailingTaskManager() {
+
+ final int PARALLELISM = 4;
+
+ ForkableFlinkMiniCluster cluster = null;
+ ActorSystem additionalSystem = null;
+
+ try {
+ Configuration config = new Configuration();
+ config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 2);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, PARALLELISM);
+ config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 16);
+
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "500 ms");
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "2 s");
+ config.setInteger(ConfigConstants.AKKA_WATCH_THRESHOLD, 2);
+
+ cluster = new ForkableFlinkMiniCluster(config, false);
+
+ // for the result
+ List<Long> resultCollection = new ArrayList<Long>();
+
+ final ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment(
+ "localhost", cluster.getJobManagerRPCPort());
+
+ env.setDegreeOfParallelism(PARALLELISM);
+ env.setNumberOfExecutionRetries(1);
+
+ env.generateSequence(1, 10)
+ .map(new FailingMapper<Long>())
+ .reduce(new ReduceFunction<Long>() {
+ @Override
+ public Long reduce(Long value1, Long value2) {
+ return value1 + value2;
+ }
+ })
+ .output(new LocalCollectionOutputFormat<Long>(resultCollection));
+
+
+ // simple reference (atomic does not matter) to pass back an exception from the trigger thread
+ final AtomicReference<Throwable> ref = new AtomicReference<Throwable>();
+
+ // trigger the execution from a separate thread, so we are available to temper with the
+ // cluster during the execution
+ Thread trigger = new Thread("program trigger") {
+ @Override
+ public void run() {
+ try {
+ env.execute();
+ }
+ catch (Throwable t) {
+ ref.set(t);
+ }
+ }
+ };
+ trigger.setDaemon(true);
+ trigger.start();
+
+ // block until all the mappers are actually deployed
+ // the mappers in turn are waiting
+ for (int i = 0; i < PARALLELISM; i++) {
+ FailingMapper.TASK_TO_COORD_QUEUE.take();
+ }
+
+ // bring up one more task manager and wait for it to appear
+ {
+ additionalSystem = cluster.startTaskManagerActorSystem(2);
+ ActorRef additionalTaskManager = cluster.startTaskManager(2, additionalSystem);
+ Object message = TaskManagerMessages.getNotifyWhenRegisteredAtJobManagerMessage();
+ Future<Object> future = Patterns.ask(additionalTaskManager, message, 30000);
+
+ try {
+ Await.result(future, new FiniteDuration(30000, TimeUnit.MILLISECONDS));
+ }
+ catch (TimeoutException e) {
+ fail ("The additional TaskManager did not come up within 30 seconds");
+ }
+ }
+
+ // kill the two other TaskManagers
+ for (ActorRef tm : cluster.getTaskManagersAsJava()) {
+ tm.tell(PoisonPill.getInstance(), null);
+ }
+
+ // wait for the next set of mappers (the recovery ones) to come online
+ for (int i = 0; i < PARALLELISM; i++) {
+ FailingMapper.TASK_TO_COORD_QUEUE.take();
+ }
+
+ // tell the mappers that they may continue this time
+ for (int i = 0; i < PARALLELISM; i++) {
+ FailingMapper.COORD_TO_TASK_QUEUE.add(new Object());
+ }
+
+ // wait for the program to finish
+ trigger.join();
+ if (ref.get() != null) {
+ Throwable t = ref.get();
+ t.printStackTrace();
+ fail("Program execution caused an exception: " + t.getMessage());
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ finally {
+ if (additionalSystem != null) {
+ additionalSystem.shutdown();
+ }
+ if (cluster != null) {
+ cluster.stop();
+ }
+ }
+ }
+
+ private static class FailingMapper<T> extends RichMapFunction<T, T> {
+
+ private static final BlockingQueue<Object> TASK_TO_COORD_QUEUE = new LinkedBlockingQueue<Object>();
+
+ private static final BlockingQueue<Object> COORD_TO_TASK_QUEUE = new LinkedBlockingQueue<Object>();
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ TASK_TO_COORD_QUEUE.add(new Object());
+ COORD_TO_TASK_QUEUE.take();
+ }
+
+ @Override
+ public T map(T value) throws Exception {
+ return value;
+ }
+ }
+}
[4/9] flink git commit: [FLINK-1584] [tests] Move logic for failure
detection test cluster into test classes
Posted by se...@apache.org.
[FLINK-1584] [tests] Move logic for failure detection test cluster into test classes
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/dbf589ad
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/dbf589ad
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/dbf589ad
Branch: refs/heads/master
Commit: dbf589ad3b1f2061eb83112f184fd43d14908d3b
Parents: 52b40de
Author: Stephan Ewen <se...@apache.org>
Authored: Thu Feb 19 18:50:16 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Feb 19 18:54:14 2015 +0100
----------------------------------------------------------------------
.../test/util/ForkableFlinkMiniCluster.scala | 20 +++-----------------
.../jobmanager/JobManagerFailsITCase.scala | 10 +++++++++-
.../taskmanager/TaskManagerFailsITCase.scala | 17 ++++++++++++++---
3 files changed, 26 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/dbf589ad/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
index 38095b9..fdb13b2 100644
--- a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
+++ b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
@@ -115,27 +115,13 @@ object ForkableFlinkMiniCluster {
import org.apache.flink.runtime.testingUtils.TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT
- def startClusterDeathWatch(numSlots: Int, numTaskManagers: Int,
- timeout: String = DEFAULT_AKKA_ASK_TIMEOUT):
- ForkableFlinkMiniCluster = {
+ def startCluster(numSlots: Int,
+ numTaskManagers: Int,
+ timeout: String = DEFAULT_AKKA_ASK_TIMEOUT): ForkableFlinkMiniCluster = {
val config = new Configuration()
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlots)
config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, numTaskManagers)
config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, timeout)
- config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "1000 ms")
- config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "4000 ms")
- config.setDouble(ConfigConstants.AKKA_WATCH_THRESHOLD, 5)
-
- new ForkableFlinkMiniCluster(config, singleActorSystem = false)
- }
-
- def startCluster(numSlots: Int, numTaskManagers: Int, timeout: String = DEFAULT_AKKA_ASK_TIMEOUT):
- ForkableFlinkMiniCluster = {
- val config = new Configuration()
- config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlots)
- config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, numTaskManagers)
- config.setInteger(ConfigConstants.JOB_MANAGER_DEAD_TASKMANAGER_TIMEOUT_KEY, 1000)
- config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, timeout)
new ForkableFlinkMiniCluster(config)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/dbf589ad/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 1e97f7b..66b2438 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
@@ -20,6 +20,7 @@ package org.apache.flink.api.scala.runtime.jobmanager
import akka.actor.{ActorSystem, PoisonPill}
import akka.testkit.{ImplicitSender, TestKit}
+import org.apache.flink.configuration.{ConfigConstants, Configuration}
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.messages.JobManagerMessages.RequestNumberRegisteredTaskManager
import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages.{JobManagerTerminated, NotifyWhenJobManagerTerminated}
@@ -42,7 +43,14 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
"detect a lost connection to the JobManager and try to reconnect to it" in {
val num_slots = 11
- val cluster = ForkableFlinkMiniCluster.startClusterDeathWatch(num_slots, 1)
+ val config = new Configuration()
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, num_slots)
+ config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 1)
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "1000 ms")
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "4000 ms")
+ config.setDouble(ConfigConstants.AKKA_WATCH_THRESHOLD, 5)
+
+ val cluster = new ForkableFlinkMiniCluster(config, singleActorSystem = false)
val tm = cluster.getTaskManagers(0)
val jm = cluster.getJobManager
http://git-wip-us.apache.org/repos/asf/flink/blob/dbf589ad/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 5ae464f..a6ee119 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
@@ -20,6 +20,9 @@ package org.apache.flink.api.scala.runtime.taskmanager
import akka.actor.{ActorSystem, Kill, PoisonPill}
import akka.testkit.{ImplicitSender, TestKit}
+
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.configuration.ConfigConstants
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.jobgraph.{AbstractJobVertex, DistributionPattern, JobGraph}
import org.apache.flink.runtime.jobmanager.Tasks.{BlockingReceiver, Sender}
@@ -27,6 +30,7 @@ import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultFailed, Re
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages._
import org.apache.flink.runtime.testingUtils.TestingUtils
import org.apache.flink.test.util.ForkableFlinkMiniCluster
+
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner
import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
@@ -46,7 +50,14 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
"detect a failing task manager" in {
val num_slots = 11
- val cluster = ForkableFlinkMiniCluster.startClusterDeathWatch(num_slots, 2)
+ val config = new Configuration()
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, num_slots)
+ config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 2)
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "1000 ms")
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "4000 ms")
+ config.setDouble(ConfigConstants.AKKA_WATCH_THRESHOLD, 5)
+
+ val cluster = new ForkableFlinkMiniCluster(config, singleActorSystem = false)
val taskManagers = cluster.getTaskManagers
val jm = cluster.getJobManager
@@ -65,7 +76,8 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
jm ! RequestNumberRegisteredTaskManager
expectMsg(1)
}
- }finally{
+ }
+ finally {
cluster.stop()
}
@@ -144,5 +156,4 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
}
}
}
-
}
[8/9] flink git commit: [FLINK-1499] [runtime] TaskManager sends
explicit disconnect message to JobManager in case of shutdown
Posted by se...@apache.org.
[FLINK-1499] [runtime] TaskManager sends explicit disconnect message to JobManager in case of shutdown
This closes #423
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/665e6012
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/665e6012
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/665e6012
Branch: refs/heads/master
Commit: 665e60128e02cf1990d5a8b13e4704e180b5054f
Parents: 5ac1911
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue Feb 17 16:36:34 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Feb 19 19:37:27 2015 +0100
----------------------------------------------------------------------
.../flink/runtime/jobmanager/JobManager.scala | 19 ++++-
.../runtime/minicluster/FlinkMiniCluster.scala | 3 +-
.../flink/runtime/taskmanager/TaskManager.scala | 8 ++-
.../runtime/testingUtils/TestingCluster.scala | 10 ---
.../testingUtils/TestingJobManager.scala | 22 ++++++
.../testingUtils/TestingTaskManager.scala | 17 ++---
.../flink/api/avro/EncoderDecoderTest.java | 1 -
.../test/util/ForkableFlinkMiniCluster.scala | 29 +++++++-
.../jobmanager/JobManagerFailsITCase.scala | 26 +++----
.../taskmanager/TaskManagerFailsITCase.scala | 73 ++++++++++++++++++--
10 files changed, 160 insertions(+), 48 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/665e6012/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
index 32a4e37..ad5c9e8 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
@@ -368,12 +368,25 @@ class JobManager(val configuration: Configuration,
taskManager forward SendStackTrace
case Terminated(taskManager) =>
- log.info("Task manager {} terminated.", taskManager.path)
- instanceManager.unregisterTaskManager(taskManager)
- context.unwatch(taskManager)
+ if(instanceManager.isRegistered(taskManager)) {
+ log.info("Task manager {} terminated.", taskManager.path)
+
+ instanceManager.unregisterTaskManager(taskManager)
+ context.unwatch(taskManager)
+ }
case RequestJobManagerStatus =>
sender ! JobManagerStatusAlive
+
+ case Disconnect(msg) =>
+ val taskManager = sender
+
+ if(instanceManager.isRegistered(taskManager)){
+ log.info("Task manager {} wants to disconnect, because {}.", taskManager.path, msg)
+
+ instanceManager.unregisterTaskManager(taskManager)
+ context.unwatch(taskManager)
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/665e6012/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
index 3eb1d1e..3679f02 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
@@ -62,8 +62,7 @@ abstract class FlinkMiniCluster(userConfiguration: Configuration,
val actorSystemsTaskManagers = for(i <- 0 until numTaskManagers) yield {
val actorSystem = if(singleActorSystem) {
jobManagerActorSystem
- }
- else{
+ } else {
startTaskManagerActorSystem(i)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/665e6012/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 ae192ce..9d15723 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
@@ -152,6 +152,10 @@ import scala.collection.JavaConverters._
override def postStop(): Unit = {
log.info("Stopping task manager {}.", self.path)
+ currentJobManager foreach {
+ _ ! Disconnect(s"TaskManager ${self.path} is shutting down.")
+ }
+
cancelAndClearEverything(new Exception("Task Manager is shutting down."))
cleanupTaskManager()
@@ -415,8 +419,8 @@ import scala.collection.JavaConverters._
new RuntimeEnvironment(jobManager, task, tdd, userCodeClassLoader,
memoryManager, ioManager, splitProvider, bcVarManager, networkEnvironment.get)
- case None => throw new IllegalStateException("TaskManager has not yet registered at " +
- "the JobManager.")
+ case None => throw new IllegalStateException("TaskManager has not yet been registered at " +
+ "a JobManager.")
}
task.setEnvironment(env)
http://git-wip-us.apache.org/repos/asf/flink/blob/665e6012/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
index 895da46..87fca99 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
@@ -19,15 +19,12 @@
package org.apache.flink.runtime.testingUtils
import akka.actor.{ActorRef, Props, ActorSystem}
-import akka.pattern.Patterns.gracefulStop
import org.apache.flink.configuration.{ConfigConstants, Configuration}
import org.apache.flink.runtime.jobmanager.{MemoryArchivist, JobManager}
import org.apache.flink.runtime.minicluster.FlinkMiniCluster
import org.apache.flink.runtime.net.NetUtils
import org.apache.flink.runtime.taskmanager.TaskManager
-import scala.concurrent.Await
-
/**
* Testing cluster which starts the [[JobManager]] and [[TaskManager]] actors with testing support
* in the same [[ActorSystem]].
@@ -74,11 +71,4 @@ FlinkMiniCluster(userConfiguration, singleActorSystem) {
networkConnectionConfig) with TestingTaskManager), TaskManager.TASK_MANAGER_NAME + "_" +
(index + 1))
}
- val stopped = gracefulStop(jobManagerActor, TestingUtils.TESTING_DURATION)
- Await.result(stopped, TestingUtils.TESTING_DURATION)
-
- jobManagerActorSystem.shutdown()
- jobManagerActorSystem.awaitTermination()
-
- jobManagerActorSystem = startJobManagerActorSystem()
}
http://git-wip-us.apache.org/repos/asf/flink/blob/665e6012/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
index f41e114..89cbfe6 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
@@ -25,7 +25,9 @@ import org.apache.flink.runtime.execution.ExecutionState
import org.apache.flink.runtime.jobgraph.{JobStatus, JobID}
import org.apache.flink.runtime.jobmanager.{JobManager, MemoryArchivist}
import org.apache.flink.runtime.messages.ExecutionGraphMessages.JobStatusChanged
+import org.apache.flink.runtime.messages.Messages.Disconnect
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages._
+import org.apache.flink.runtime.testingUtils.TestingMessages.DisableDisconnect
import scala.collection.convert.WrapAsScala
import scala.concurrent.Future
@@ -52,6 +54,8 @@ trait TestingJobManager extends ActorLogMessages with WrapAsScala {
val waitForJobStatus = scala.collection.mutable.HashMap[JobID,
collection.mutable.HashMap[JobStatus, Set[ActorRef]]]()
+ var disconnectDisabled = false
+
abstract override def receiveWithLogMessages: Receive = {
receiveTestingMessages orElse super.receiveWithLogMessages
}
@@ -125,6 +129,7 @@ trait TestingJobManager extends ActorLogMessages with WrapAsScala {
listener ! TaskManagerTerminated(taskManager)
}
}
+
case RequestWorkingTaskManager(jobID) =>
currentJobs.get(jobID) match {
case Some((eg, _)) =>
@@ -170,6 +175,23 @@ trait TestingJobManager extends ActorLogMessages with WrapAsScala {
if (cleanup) {
waitForJobStatus.remove(jobID)
}
+
+ case DisableDisconnect =>
+ disconnectDisabled = true
+
+ case msg: Disconnect =>
+ if (!disconnectDisabled) {
+ super.receiveWithLogMessages(msg)
+
+ val taskManager = sender
+
+ waitForTaskManagerToBeTerminated.remove(taskManager.path.name) foreach {
+ _ foreach {
+ listener =>
+ listener ! TaskManagerTerminated(taskManager)
+ }
+ }
+ }
}
def checkIfAllVerticesRunning(jobID: JobID): Boolean = {
http://git-wip-us.apache.org/repos/asf/flink/blob/665e6012/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
index 0ecbd04..59106d3 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
@@ -119,21 +119,18 @@ trait TestingTaskManager extends ActorLogMessages {
}
case msg:Disconnect =>
- super.receiveWithLogMessages(msg)
+ if (!disconnectDisabled) {
+ super.receiveWithLogMessages(msg)
- val jobManager = sender
+ val jobManager = sender
- waitForJobManagerToBeTerminated.remove(jobManager.path.name) foreach {
- _ foreach {
- _ ! JobManagerTerminated(jobManager)
+ waitForJobManagerToBeTerminated.remove(jobManager.path.name) foreach {
+ _ foreach {
+ _ ! JobManagerTerminated(jobManager)
+ }
}
}
- case msg:Disconnect =>
- if (!disconnectDisabled) {
- super.receiveWithLogMessages(msg)
- }
-
case DisableDisconnect =>
disconnectDisabled = true
}
http://git-wip-us.apache.org/repos/asf/flink/blob/665e6012/flink-staging/flink-avro/src/test/java/org/apache/flink/api/avro/EncoderDecoderTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-avro/src/test/java/org/apache/flink/api/avro/EncoderDecoderTest.java b/flink-staging/flink-avro/src/test/java/org/apache/flink/api/avro/EncoderDecoderTest.java
index 8f14cb3..8f29dca 100644
--- a/flink-staging/flink-avro/src/test/java/org/apache/flink/api/avro/EncoderDecoderTest.java
+++ b/flink-staging/flink-avro/src/test/java/org/apache/flink/api/avro/EncoderDecoderTest.java
@@ -29,7 +29,6 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
-import org.apache.avro.generic.GenericData;
import org.apache.avro.reflect.ReflectDatumReader;
import org.apache.avro.reflect.ReflectDatumWriter;
import org.apache.flink.api.io.avro.generated.Colors;
http://git-wip-us.apache.org/repos/asf/flink/blob/665e6012/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
index fdb13b2..eeb622a 100644
--- a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
+++ b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
@@ -19,12 +19,15 @@
package org.apache.flink.test.util
import akka.actor.{Props, ActorRef, ActorSystem}
+import akka.pattern.Patterns._
import org.apache.flink.configuration.{ConfigConstants, Configuration}
import org.apache.flink.runtime.jobmanager.{MemoryArchivist, JobManager}
import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster
import org.apache.flink.runtime.taskmanager.TaskManager
-import org.apache.flink.runtime.testingUtils.{TestingJobManager, TestingMemoryArchivist,
-TestingTaskManager}
+import org.apache.flink.runtime.testingUtils.{TestingUtils, TestingJobManager,
+TestingMemoryArchivist, TestingTaskManager}
+
+import scala.concurrent.Await
/**
* A forkable mini cluster is a special case of the mini cluster, used for parallel test execution
@@ -106,9 +109,29 @@ class ForkableFlinkMiniCluster(userConfiguration: Configuration, singleActorSyst
}
def restartJobManager(): Unit = {
- jobManagerActorSystem.stop(jobManagerActor)
+ val stopped = gracefulStop(jobManagerActor, TestingUtils.TESTING_DURATION)
+ Await.result(stopped, TestingUtils.TESTING_DURATION)
+
+ jobManagerActorSystem.shutdown()
+ jobManagerActorSystem.awaitTermination()
+
+ jobManagerActorSystem = startJobManagerActorSystem()
jobManagerActor = startJobManager(jobManagerActorSystem)
}
+
+ def restartTaskManager(index: Int): Unit = {
+ val stopped = gracefulStop(taskManagerActors(index), TestingUtils.TESTING_DURATION)
+ Await.result(stopped, TestingUtils.TESTING_DURATION)
+
+ taskManagerActorSystems(index).shutdown()
+ taskManagerActorSystems(index).awaitTermination()
+
+ val taskManagerActorSystem = startTaskManagerActorSystem(index)
+ val taskManagerActor = startTaskManager(index)(jobManagerActorSystem)
+
+ taskManagerActors = taskManagerActors.patch(index, Seq(taskManagerActor), 1)
+ taskManagerActorSystems = taskManagerActorSystems.patch(index, Seq(taskManagerActorSystem), 1)
+ }
}
object ForkableFlinkMiniCluster {
http://git-wip-us.apache.org/repos/asf/flink/blob/665e6012/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 3813beb..ac1864c 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
@@ -18,12 +18,19 @@
package org.apache.flink.api.scala.runtime.jobmanager
+import akka.actor.Status.{Success, Failure}
import akka.actor.{ActorSystem, PoisonPill}
import akka.testkit.{ImplicitSender, TestKit}
import org.apache.flink.configuration.{ConfigConstants, Configuration}
import org.apache.flink.runtime.akka.AkkaUtils
+import org.apache.flink.runtime.client.JobExecutionException
+import org.apache.flink.runtime.jobgraph.{JobGraph, AbstractJobVertex}
import org.apache.flink.runtime.jobmanager.Tasks.{NoOpInvokable, BlockingNoOpInvokable}
-import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages.{JobManagerTerminated, NotifyWhenJobManagerTerminated}
+import org.apache.flink.runtime.messages.JobManagerMessages._
+import org.apache.flink.runtime.testingUtils.TestingMessages.DisableDisconnect
+import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages.{JobManagerTerminated,
+NotifyWhenJobManagerTerminated}
+import org.apache.flink.runtime.testingUtils.TestingUtils
import org.apache.flink.test.util.ForkableFlinkMiniCluster
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner
@@ -39,7 +46,7 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
TestKit.shutdownActorSystem(system)
}
- "The TaskManager" should {
+ "A TaskManager" should {
"detect a lost connection to the JobManager and try to reconnect to it" in {
val num_slots = 11
@@ -79,9 +86,7 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
cluster.stop()
}
}
- }
- "The system" should {
"go into a clean state in case of a JobManager failure" in {
val num_slots = 20
@@ -95,15 +100,15 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
noOp.setInvokableClass(classOf[NoOpInvokable])
val jobGraph2 = new JobGraph("NoOp Testjob", noOp)
- val cluster = TestingUtils.startTestingClusterDeathWatch(num_slots/2, 2)
+ val cluster = ForkableFlinkMiniCluster.startClusterDeathWatch(num_slots / 2, 2)
var jm = cluster.getJobManager
val tm = cluster.getTaskManagers(0)
- try{
+ try {
within(TestingUtils.TESTING_DURATION) {
jm ! SubmitJob(jobGraph)
- expectMsg(SubmissionSuccess(jobGraph.getJobID))
+ expectMsg(Success(jobGraph.getJobID))
tm ! NotifyWhenJobManagerTerminated(jm)
@@ -118,13 +123,8 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
cluster.waitForTaskManagersToBeRegistered()
jm ! SubmitJob(jobGraph2)
- val response = expectMsgType[SubmissionResponse]
- response match {
- case SubmissionSuccess(jobID) => jobID should equal(jobGraph2.getJobID)
- case SubmissionFailure(jobID, t) =>
- fail("Submission of the second job failed.", t)
- }
+ val failure = expectMsgType[Success]
val result = expectMsgType[JobResultSuccess]
http://git-wip-us.apache.org/repos/asf/flink/blob/665e6012/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 60c67fb..c81ec88 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
@@ -22,14 +22,16 @@ import akka.actor.Status.{Failure, Success}
import akka.actor.{ActorSystem, Kill, PoisonPill}
import akka.testkit.{ImplicitSender, TestKit}
-import org.apache.flink.configuration.Configuration
import org.apache.flink.configuration.ConfigConstants
+import org.apache.flink.configuration.Configuration
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.client.JobExecutionException
import org.apache.flink.runtime.jobgraph.{AbstractJobVertex, DistributionPattern, JobGraph}
-import org.apache.flink.runtime.jobmanager.Tasks.{BlockingReceiver, Sender}
-import org.apache.flink.runtime.messages.JobManagerMessages.{RequestNumberRegisteredTaskManager, SubmitJob}
+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.TaskManagerMessages.{RegisteredAtJobManager, NotifyWhenRegisteredAtJobManager}
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages._
+import org.apache.flink.runtime.testingUtils.TestingMessages.DisableDisconnect
import org.apache.flink.runtime.testingUtils.TestingUtils
import org.apache.flink.test.util.ForkableFlinkMiniCluster
@@ -64,6 +66,8 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
val taskManagers = cluster.getTaskManagers
val jm = cluster.getJobManager
+ jm ! DisableDisconnect
+
try{
within(TestingUtils.TESTING_DURATION){
jm ! RequestNumberRegisteredTaskManager
@@ -82,7 +86,6 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
finally {
cluster.stop()
}
-
}
"handle gracefully failing task manager" in {
@@ -173,5 +176,67 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
cluster.stop()
}
}
+
+ "go into a clean state in case of a TaskManager failure" in {
+ val num_slots = 20
+
+ val sender = new AbstractJobVertex("BlockingSender")
+ sender.setParallelism(num_slots)
+ sender.setInvokableClass(classOf[BlockingNoOpInvokable])
+ val jobGraph = new JobGraph("Blocking Testjob", sender)
+
+ val noOp = new AbstractJobVertex("NoOpInvokable")
+ noOp.setParallelism(num_slots)
+ noOp.setInvokableClass(classOf[NoOpInvokable])
+ val jobGraph2 = new JobGraph("NoOp Testjob", noOp)
+
+ val config = new Configuration()
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, num_slots/2)
+ config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 2)
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "1000 ms")
+ config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "4000 ms")
+ config.setDouble(ConfigConstants.AKKA_WATCH_THRESHOLD, 5)
+
+ val cluster = new ForkableFlinkMiniCluster(config, singleActorSystem = false)
+
+ var tm = cluster.getTaskManagers(0)
+ val jm = cluster.getJobManager
+
+ try{
+ within(TestingUtils.TESTING_DURATION){
+ jm ! SubmitJob(jobGraph)
+ expectMsg(Success(jobGraph.getJobID))
+
+ tm ! PoisonPill
+
+ val failure = expectMsgType[Failure]
+
+ failure.cause match {
+ case e: JobExecutionException =>
+ jobGraph.getJobID should equal(e.getJobID)
+
+ case e => fail(s"Received wrong exception $e.")
+ }
+
+ cluster.restartTaskManager(0)
+
+ tm = cluster.getTaskManagers(0)
+
+ tm ! NotifyWhenRegisteredAtJobManager
+
+ expectMsg(RegisteredAtJobManager)
+
+ jm ! SubmitJob(jobGraph2)
+
+ expectMsgType[Success]
+
+ val result = expectMsgType[JobResultSuccess]
+ result.jobID should equal(jobGraph2.getJobID)
+ }
+ } finally {
+ cluster.stop()
+ }
+ }
}
+
}