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/03/10 15:00:02 UTC
[02/14] flink git commit: [FLINK-1638] [streaming] Fault tolerance
logic in JM and TM
[FLINK-1638] [streaming] Fault tolerance logic in JM and TM
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a34869c0
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a34869c0
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a34869c0
Branch: refs/heads/master
Commit: a34869c0b40ac9055f68f1c431f7fb28b936e9f8
Parents: 5061edb
Author: Paris Carbone <se...@gmail.com>
Authored: Fri Feb 20 23:12:57 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Mar 10 14:58:48 2015 +0100
----------------------------------------------------------------------
.../flink/runtime/jobmanager/JobManager.scala | 37 +++++++++++++++++---
.../flink/runtime/taskmanager/TaskManager.scala | 17 ++++++++-
2 files changed, 48 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a34869c0/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 7ba06e7..4f58ba7 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
@@ -41,7 +41,7 @@ import org.apache.flink.runtime.ActorLogMessages
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager
import org.apache.flink.runtime.instance.InstanceManager
-import org.apache.flink.runtime.jobgraph.{JobGraph, JobStatus, JobID}
+import org.apache.flink.runtime.jobgraph.{ScheduleMode,JobGraph,JobStatus,JobID}
import org.apache.flink.runtime.jobmanager.accumulators.AccumulatorManager
import org.apache.flink.runtime.jobmanager.scheduler.{Scheduler => FlinkScheduler}
import org.apache.flink.runtime.messages.JobManagerMessages._
@@ -97,12 +97,14 @@ class JobManager(val configuration: Configuration,
val delayBetweenRetries: Long,
val timeout: FiniteDuration)
extends Actor with ActorLogMessages with ActorLogging {
-
+
/** Reference to the log, for debugging */
- protected val LOG = JobManager.LOG
+ val LOG = JobManager.LOG
/** List of current jobs running jobs */
- protected val currentJobs = scala.collection.mutable.HashMap[JobID, (ExecutionGraph, JobInfo)]()
+ val currentJobs = scala.collection.mutable.HashMap[JobID, (ExecutionGraph, JobInfo)]()
+
+ val barrierMonitors = scala.collection.mutable.HashMap[JobID, ActorRef]()
/**
* Run when the job manager is started. Simply logs an informational message.
@@ -299,13 +301,38 @@ class JobManager(val configuration: Configuration,
jobInfo.client ! Failure(exception)
throw exception
}
-
+ barrierMonitors.get(jobID) match {
+ case Some(monitor) =>
+ monitor ! PoisonPill
+ barrierMonitors.remove(jobID)
+ case None =>
+ }
removeJob(jobID)
}
+ else {
+ newJobStatus match {
+ case JobStatus.RUNNING => currentJobs.get(jobID) match {
+ case Some((executionGraph, _)) =>
+ //FIXME this is just a fast n dirty check for determining streaming jobs
+ if (executionGraph.getScheduleMode == ScheduleMode.ALL) {
+ barrierMonitors += jobID -> StreamStateMonitor.props(context, executionGraph)
+ }
+ case None =>
+ log.error("Cannot create state monitor for job ID {}.", jobID)
+ new IllegalStateException("Cannot find execution graph for job ID " + jobID)
+ }
+ }
+ }
case None =>
removeJob(jobID)
}
+ case BarrierAck(jobID, jobVertex, instanceID, checkpoint) =>
+ barrierMonitors.get(jobID) match {
+ case Some(monitor) => monitor ! BarrierAck(jobID, jobVertex, instanceID, checkpoint)
+ case None =>
+ }
+
case ScheduleOrUpdateConsumers(jobId, executionId, partitionIndex) =>
currentJobs.get(jobId) match {
case Some((executionGraph, _)) =>
http://git-wip-us.apache.org/repos/asf/flink/blob/a34869c0/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 a48fc8d..4271141 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
@@ -43,7 +43,8 @@ import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync
import org.apache.flink.runtime.io.network.NetworkEnvironment
import org.apache.flink.runtime.io.network.netty.NettyConfig
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID
-import org.apache.flink.runtime.jobmanager.JobManager
+import org.apache.flink.runtime.jobgraph.tasks.BarrierTransceiver
+import org.apache.flink.runtime.jobmanager.{BarrierReq,JobManager}
import org.apache.flink.runtime.memorymanager.DefaultMemoryManager
import org.apache.flink.runtime.messages.JobManagerMessages.UpdateTaskExecutionState
import org.apache.flink.runtime.messages.Messages.{Disconnect, Acknowledge}
@@ -349,6 +350,20 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo,
networkEnvironment foreach {
_.getPartitionManager.failIntermediateResultPartitions(executionID)
}
+
+ case BarrierReq(attemptID, checkpointID) =>
+ log.debug("[FT-TaskManager] Barrier request received for attempt {}", attemptID)
+ runningTasks.get(attemptID) match {
+ case Some(i) =>
+ if (i.getExecutionState == ExecutionState.RUNNING) {
+ i.getEnvironment.getInvokable match {
+ case barrierTransceiver: BarrierTransceiver =>
+ barrierTransceiver.broadcastBarrier(checkpointID)
+ case _ => log.error("[FT-TaskManager] Received a barrier for the wrong vertex")
+ }
+ }
+ case None => log.error("[FT-TaskManager] Received a barrier for an unknown vertex")
+ }
}
/**