You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2016/11/01 08:37:37 UTC
[2/4] flink git commit: [FLINK-4887] [execution graph] Introduce
TaskManagerGateway to encapsulate communcation logic
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/flink-runtime/src/main/scala/org/apache/flink/runtime/clusterframework/ContaineredJobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/clusterframework/ContaineredJobManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/clusterframework/ContaineredJobManager.scala
index f426254..72df671 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/clusterframework/ContaineredJobManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/clusterframework/ContaineredJobManager.scala
@@ -32,8 +32,8 @@ import org.apache.flink.runtime.jobgraph.JobStatus
import org.apache.flink.runtime.jobmanager.scheduler.{Scheduler => FlinkScheduler}
import org.apache.flink.runtime.jobmanager.{JobManager, SubmittedJobGraphStore}
import org.apache.flink.runtime.leaderelection.LeaderElectionService
+import org.apache.flink.runtime.messages.Acknowledge
import org.apache.flink.runtime.messages.JobManagerMessages.{CurrentJobStatus, JobNotFound, RequestJobStatus}
-import org.apache.flink.runtime.messages.Messages.Acknowledge
import org.apache.flink.runtime.metrics.{MetricRegistry => FlinkMetricRegistry}
import scala.concurrent.duration._
@@ -122,7 +122,7 @@ abstract class ContaineredJobManager(
)(context.dispatcher)
}
- sender() ! decorateMessage(Acknowledge)
+ sender() ! decorateMessage(Acknowledge.get())
case msg: GetClusterStatus =>
sender() ! decorateMessage(
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/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 2bae7fe..5a5d7a5 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
@@ -44,7 +44,7 @@ import org.apache.flink.runtime.clusterframework.FlinkResourceManager
import org.apache.flink.runtime.clusterframework.messages._
import org.apache.flink.runtime.clusterframework.standalone.StandaloneResourceManager
import org.apache.flink.runtime.clusterframework.types.ResourceID
-import org.apache.flink.runtime.concurrent.BiFunction
+import org.apache.flink.runtime.concurrent.{AcceptFunction, BiFunction}
import org.apache.flink.runtime.execution.SuppressRestartsException
import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager
import org.apache.flink.runtime.executiongraph.restart.RestartStrategyFactory
@@ -54,12 +54,14 @@ import org.apache.flink.runtime.io.network.PartitionState
import org.apache.flink.runtime.jobgraph.{JobGraph, JobStatus}
import org.apache.flink.runtime.jobmanager.SubmittedJobGraphStore.SubmittedJobGraphListener
import org.apache.flink.runtime.jobmanager.scheduler.{Scheduler => FlinkScheduler}
+import org.apache.flink.runtime.jobmanager.slots.ActorTaskManagerGateway
import org.apache.flink.runtime.leaderelection.{LeaderContender, LeaderElectionService, StandaloneLeaderElectionService}
import org.apache.flink.runtime.messages.ArchiveMessages.ArchiveExecutionGraph
import org.apache.flink.runtime.messages.ExecutionGraphMessages.JobStatusChanged
import org.apache.flink.runtime.messages.JobManagerMessages._
-import org.apache.flink.runtime.messages.Messages.{Acknowledge, Disconnect}
+import org.apache.flink.runtime.messages.Messages.Disconnect
import org.apache.flink.runtime.messages.RegistrationMessages._
+import org.apache.flink.runtime.messages.{Acknowledge, StackTrace}
import org.apache.flink.runtime.messages.TaskManagerMessages.{Heartbeat, SendStackTrace}
import org.apache.flink.runtime.messages.TaskMessages.UpdateTaskExecutionState
import org.apache.flink.runtime.messages.accumulators._
@@ -137,15 +139,6 @@ class JobManager(
override val log = Logger(getClass)
- /** The extra execution context, for futures, with a custom logging reporter */
- protected val executionContext: ExecutionContext = ExecutionContext.fromExecutor(
- executorService,
- (t: Throwable) => {
- if (!context.system.isTerminated) {
- log.error("Executor could not execute task", t)
- }
- })
-
/** Either running or not yet archived jobs (session hasn't been ended). */
protected val currentJobs = scala.collection.mutable.HashMap[JobID, (ExecutionGraph, JobInfo)]()
@@ -243,9 +236,9 @@ class JobManager(
// disconnect the registered task managers
instanceManager.getAllRegisteredInstances.asScala.foreach {
- instance => instance.getActorGateway().tell(
- Disconnect(instance.getId, "JobManager is shutting down"),
- new AkkaActorGateway(self, leaderSessionID.orNull))
+ instance => instance.getTaskManagerGateway().disconnectFromJobManager(
+ instance.getId,
+ new Exception("JobManager is shuttind down."))
}
try {
@@ -333,9 +326,9 @@ class JobManager(
// disconnect the registered task managers
instanceManager.getAllRegisteredInstances.asScala.foreach {
- instance => instance.getActorGateway().tell(
- Disconnect(instance.getId, "JobManager is no longer the leader"),
- new AkkaActorGateway(self, leaderSessionID.orNull))
+ instance => instance.getTaskManagerGateway().disconnectFromJobManager(
+ instance.getId(),
+ new Exception("JobManager is no longer the leader"))
}
instanceManager.unregisterAllTaskManagers()
@@ -424,12 +417,14 @@ class JobManager(
libraryCacheManager.getBlobServerPort))
} else {
try {
+ val actorGateway = new AkkaActorGateway(taskManager, leaderSessionID.orNull)
+ val taskManagerGateway = new ActorTaskManagerGateway(actorGateway)
+
val instanceID = instanceManager.registerTaskManager(
- taskManager,
+ taskManagerGateway,
connectionInfo,
hardwareInformation,
- numberOfSlots,
- leaderSessionID.orNull)
+ numberOfSlots)
taskManagerMap.put(taskManager, instanceID)
@@ -458,7 +453,15 @@ class JobManager(
Option(instanceManager.getRegisteredInstance(resourceID)) match {
case Some(instance) =>
// trigger removal of task manager
- handleTaskManagerTerminated(instance.getActorGateway.actor(), instance.getId)
+ val taskManagerGateway = instance.getTaskManagerGateway
+
+ taskManagerGateway match {
+ case x: ActorTaskManagerGateway =>
+ handleTaskManagerTerminated(x.getActorGateway().actor(), instance.getId)
+ case _ => log.debug(s"Cannot remove reosurce ${resourceID}, because there is " +
+ s"no ActorRef registered.")
+ }
+
case None =>
log.debug(s"Resource $resourceID has not been registered at job manager.")
}
@@ -924,7 +927,7 @@ class JobManager(
case Some((executionGraph, _)) =>
try {
executionGraph.scheduleOrUpdateConsumers(partitionId)
- sender ! decorateMessage(Acknowledge)
+ sender ! decorateMessage(Acknowledge.get())
} catch {
case e: Exception =>
sender ! decorateMessage(
@@ -1047,8 +1050,20 @@ class JobManager(
case message: InfoMessage => handleInfoRequestMessage(message, sender())
case RequestStackTrace(instanceID) =>
- val gateway = instanceManager.getRegisteredInstanceById(instanceID).getActorGateway
- gateway.forward(SendStackTrace, new AkkaActorGateway(sender, leaderSessionID.orNull))
+ val taskManagerGateway = instanceManager
+ .getRegisteredInstanceById(instanceID)
+ .getTaskManagerGateway
+
+ val stackTraceFuture = taskManagerGateway
+ .requestStackTrace(Time.milliseconds(timeout.toMillis))
+
+ val originalSender = new AkkaActorGateway(sender(), leaderSessionID.orNull)
+
+ stackTraceFuture.thenAccept(new AcceptFunction[StackTrace] {
+ override def accept(value: StackTrace): Unit = {
+ originalSender.tell(value)
+ }
+ })
case Terminated(taskManagerActorRef) =>
taskManagerMap.get(taskManagerActorRef) match {
@@ -1088,11 +1103,12 @@ class JobManager(
case None =>
}
- case Disconnect(instanceId, msg) =>
+ case Disconnect(instanceId, cause) =>
val taskManager = sender()
if (instanceManager.isRegistered(instanceId)) {
- log.info(s"Task manager ${taskManager.path} wants to disconnect, because $msg.")
+ log.info(s"Task manager ${taskManager.path} wants to disconnect, " +
+ s"because ${cause.getMessage}.")
instanceManager.unregisterTaskManager(instanceId, false)
taskManagerMap.remove(taskManager)
@@ -1107,7 +1123,7 @@ class JobManager(
// stop all task managers
instanceManager.getAllRegisteredInstances.asScala foreach {
instance =>
- instance.getActorGateway.tell(msg)
+ instance.getTaskManagerGateway.stopCluster(msg.finalStatus(), msg.message())
}
// send resource manager the ok
@@ -1238,7 +1254,7 @@ class JobManager(
executionGraph,
jobGraph,
flinkConfiguration,
- executionContext,
+ executorService,
userCodeLoader,
checkpointRecoveryFactory,
Time.of(timeout.length, timeout.unit),
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/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 aa22914..1103a69 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
@@ -26,11 +26,6 @@ import org.apache.flink.runtime.instance.InstanceID
object Messages {
/**
- * Message to signal the successful reception of another message
- */
- case object Acknowledge
-
- /**
* Signals that the receiver (JobManager/TaskManager) shall disconnect the sender.
*
* The TaskManager may send this on shutdown to let the JobManager realize the TaskManager
@@ -39,14 +34,7 @@ object Messages {
* The JobManager may send this message to its TaskManagers to let them clean up their
* tasks that depend on the JobManager and go into a clean state.
*
- * @param reason The reason for disconnecting, to be displayed in log and error messages.
- */
- case class Disconnect(instanceId: InstanceID, reason: String) extends RequiresLeaderSessionID
-
- /**
- * Accessor for the case object instance, to simplify Java interoperability.
- *
- * @return The Acknowledge case object instance.
+ * @param cause The reason for disconnecting, to be displayed in log and error messages.
*/
- def getAcknowledge(): Acknowledge.type = Acknowledge
+ case class Disconnect(instanceId: InstanceID, cause: Exception) extends RequiresLeaderSessionID
}
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/StackTraceSampleMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/StackTraceSampleMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/StackTraceSampleMessages.scala
index c3c26dc..3a7c057 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/StackTraceSampleMessages.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/StackTraceSampleMessages.scala
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.messages
import akka.actor.ActorRef
+import org.apache.flink.api.common.time.Time
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID
import scala.concurrent.duration.FiniteDuration
@@ -47,43 +48,11 @@ object StackTraceSampleMessages {
sampleId: Int,
executionId: ExecutionAttemptID,
numSamples: Int,
- delayBetweenSamples: FiniteDuration,
+ delayBetweenSamples: Time,
maxStackTraceDepth: Int = 0)
extends StackTraceSampleMessages with java.io.Serializable
/**
- * Response after a successful stack trace sample (sent by the task managers
- * to the job manager).
- *
- * @param sampleId ID of the this sample.
- * @param executionId ID of the sampled task.
- * @param samples Stack trace samples (head is most recent sample).
- */
- case class ResponseStackTraceSampleSuccess(
- sampleId: Int,
- executionId: ExecutionAttemptID,
- samples: java.util.List[Array[StackTraceElement]])
- extends StackTraceSampleMessages {
-
- override def toString: String =
- s"ResponseStackTraceSampleSuccess($sampleId, $executionId, ${samples.size()} samples)"
- }
-
- /**
- * Response after a failed stack trace sample (sent by the task managers to
- * the job manager).
- *
- * @param sampleId ID of the this sample.
- * @param executionId ID of the sampled task.
- * @param cause Failure cause.
- */
- case class ResponseStackTraceSampleFailure(
- sampleId: Int,
- executionId: ExecutionAttemptID,
- cause: Exception)
- extends StackTraceSampleMessages
-
- /**
* Task manager internal sample message.
*
* @param sampleId ID of the this sample.
@@ -100,7 +69,7 @@ object StackTraceSampleMessages {
case class SampleTaskStackTrace(
sampleId: Int,
executionId: ExecutionAttemptID,
- delayBetweenSamples: FiniteDuration,
+ delayBetweenSamples: Time,
maxStackTraceDepth: Int,
numRemainingSamples: Int,
currentTraces: java.util.List[Array[StackTraceElement]],
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskControlMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskControlMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskControlMessages.scala
index 158024d..6c0da2b 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskControlMessages.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskControlMessages.scala
@@ -18,8 +18,10 @@
package org.apache.flink.runtime.messages
+import java.util
+
import org.apache.flink.runtime.deployment.{InputChannelDeploymentDescriptor, TaskDeploymentDescriptor}
-import org.apache.flink.runtime.executiongraph.ExecutionAttemptID
+import org.apache.flink.runtime.executiongraph.{ExecutionAttemptID, PartitionInfo}
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID
import org.apache.flink.runtime.taskmanager.TaskExecutionState
@@ -115,7 +117,7 @@ object TaskMessages {
*/
case class UpdateTaskMultiplePartitionInfos(
executionID: ExecutionAttemptID,
- partitionInfos: Seq[(IntermediateDataSetID, InputChannelDeploymentDescriptor)])
+ partitionInfos: java.lang.Iterable[PartitionInfo])
extends UpdatePartitionInfo
/**
@@ -141,25 +143,6 @@ object TaskMessages {
case class UpdateTaskExecutionState(taskExecutionState: TaskExecutionState)
extends TaskMessage with RequiresLeaderSessionID
- /**
- * Response message to updates in the task state. Send for example as a response to
- *
- * - [[SubmitTask]]
- * - [[CancelTask]]
- *
- * @param executionID identifying the respective task
- * @param success indicating whether the operation has been successful
- * @param description Optional description for unsuccessful results.
- */
- case class TaskOperationResult(
- executionID: ExecutionAttemptID,
- success: Boolean,
- description: String)
- extends TaskMessage {
- def this(executionID: ExecutionAttemptID, success: Boolean) = this(executionID, success, "")
- }
-
-
// --------------------------------------------------------------------------
// Utility Functions
// --------------------------------------------------------------------------
@@ -173,9 +156,14 @@ object TaskMessages {
require(resultIDs.size() == partitionInfos.size(),
"ResultIDs must have the same length as partitionInfos.")
- import scala.collection.JavaConverters.asScalaBufferConverter
+ val partitionInfoList = new util.ArrayList[PartitionInfo](resultIDs.size())
+
+ for (i <- 0 until resultIDs.size()) {
+ partitionInfoList.add(new PartitionInfo(resultIDs.get(i), partitionInfos.get(i)))
+ }
- new UpdateTaskMultiplePartitionInfos(executionID,
- resultIDs.asScala.zip(partitionInfos.asScala))
+ new UpdateTaskMultiplePartitionInfos(
+ executionID,
+ partitionInfoList)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/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 74ed560..5073bb0 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
@@ -22,7 +22,7 @@ import java.io.{File, FileInputStream, IOException}
import java.lang.management.ManagementFactory
import java.net.{InetAddress, InetSocketAddress}
import java.util
-import java.util.UUID
+import java.util.{Collections, UUID}
import _root_.akka.actor._
import _root_.akka.pattern.ask
@@ -38,10 +38,10 @@ import org.apache.flink.runtime.clusterframework.types.ResourceID
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.blob.{BlobCache, BlobClient, BlobService}
import org.apache.flink.runtime.broadcast.BroadcastVariableManager
-import org.apache.flink.runtime.deployment.{InputChannelDeploymentDescriptor, TaskDeploymentDescriptor}
+import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor
import org.apache.flink.runtime.execution.ExecutionState
import org.apache.flink.runtime.execution.librarycache.{BlobLibraryCacheManager, FallbackLibraryCacheManager, LibraryCacheManager}
-import org.apache.flink.runtime.executiongraph.ExecutionAttemptID
+import org.apache.flink.runtime.executiongraph.{ExecutionAttemptID, PartitionInfo}
import org.apache.flink.runtime.filecache.FileCache
import org.apache.flink.runtime.instance.{AkkaActorGateway, HardwareDescription, InstanceID}
import org.apache.flink.runtime.io.disk.iomanager.IOManager.IOMode
@@ -50,12 +50,12 @@ import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool
import org.apache.flink.runtime.io.network.{LocalConnectionManager, NetworkEnvironment, TaskEventDispatcher}
import org.apache.flink.runtime.io.network.netty.{NettyConfig, NettyConnectionManager, PartitionStateChecker}
import org.apache.flink.runtime.io.network.partition.{ResultPartitionConsumableNotifier, ResultPartitionManager}
-import org.apache.flink.runtime.jobgraph.IntermediateDataSetID
import org.apache.flink.runtime.leaderretrieval.{LeaderRetrievalListener, LeaderRetrievalService}
import org.apache.flink.runtime.memory.MemoryManager
+import org.apache.flink.runtime.messages.{Acknowledge, StackTraceSampleResponse}
import org.apache.flink.runtime.messages.Messages._
import org.apache.flink.runtime.messages.RegistrationMessages._
-import org.apache.flink.runtime.messages.StackTraceSampleMessages.{ResponseStackTraceSampleFailure, ResponseStackTraceSampleSuccess, SampleTaskStackTrace, StackTraceSampleMessages, TriggerStackTraceSample}
+import org.apache.flink.runtime.messages.StackTraceSampleMessages.{SampleTaskStackTrace, StackTraceSampleMessages, TriggerStackTraceSample}
import org.apache.flink.runtime.messages.TaskManagerMessages._
import org.apache.flink.runtime.messages.TaskMessages._
import org.apache.flink.runtime.messages.checkpoint.{AbstractCheckpointMessage, NotifyCheckpointComplete, TriggerCheckpoint}
@@ -310,16 +310,16 @@ class TaskManager(
// this message indicates that some actor watched by this TaskManager has died
case Terminated(actor: ActorRef) =>
if (isConnected && actor == currentJobManager.orNull) {
- handleJobManagerDisconnect(sender(), "JobManager is no longer reachable")
+ handleJobManagerDisconnect("JobManager is no longer reachable")
triggerTaskManagerRegistration()
} else {
log.warn(s"Received unrecognized disconnect message " +
s"from ${if (actor == null) null else actor.path}.")
}
- case Disconnect(instanceIdToDisconnect, msg) =>
+ case Disconnect(instanceIdToDisconnect, cause) =>
if (instanceIdToDisconnect.equals(instanceID)) {
- handleJobManagerDisconnect(sender(), s"JobManager requested disconnect: $msg")
+ handleJobManagerDisconnect(s"JobManager requested disconnect: ${cause.getMessage()}")
triggerTaskManagerRegistration()
} else {
log.debug(s"Received disconnect message for wrong instance id ${instanceIdToDisconnect}.")
@@ -392,7 +392,9 @@ class TaskManager(
// tell the task about the availability of a new input partition
case UpdateTaskSinglePartitionInfo(executionID, resultID, partitionInfo) =>
- updateTaskInputPartitions(executionID, List((resultID, partitionInfo)))
+ updateTaskInputPartitions(
+ executionID,
+ Collections.singletonList(new PartitionInfo(resultID, partitionInfo)))
// tell the task about the availability of some new input partitions
case UpdateTaskMultiplePartitionInfos(executionID, partitionInfos) =>
@@ -469,24 +471,14 @@ class TaskManager(
if (task != null) {
try {
task.stopExecution()
- sender ! decorateMessage(new TaskOperationResult(executionID, true))
+ sender ! decorateMessage(Acknowledge.get())
} catch {
case t: Throwable =>
- sender ! decorateMessage(
- new TaskOperationResult(
- executionID,
- false,
- t.getClass().getSimpleName() + ": " + t.getLocalizedMessage())
- )
+ sender ! decorateMessage(Failure(t))
}
} else {
log.debug(s"Cannot find task to stop for execution ${executionID})")
- sender ! decorateMessage(
- new TaskOperationResult(
- executionID,
- false,
- "No task with that execution ID was found.")
- )
+ sender ! decorateMessage(Acknowledge.get())
}
// cancels a task
@@ -494,15 +486,10 @@ class TaskManager(
val task = runningTasks.get(executionID)
if (task != null) {
task.cancelExecution()
- sender ! decorateMessage(new TaskOperationResult(executionID, true))
+ sender ! decorateMessage(Acknowledge.get())
} else {
log.debug(s"Cannot find task to cancel for execution $executionID)")
- sender ! decorateMessage(
- new TaskOperationResult(
- executionID,
- false,
- "No task with that execution ID was found.")
- )
+ sender ! decorateMessage(Acknowledge.get())
}
}
}
@@ -764,14 +751,16 @@ class TaskManager(
sender)
context.system.scheduler.scheduleOnce(
- delayBetweenSamples,
+ new FiniteDuration(delayBetweenSamples.getSize, delayBetweenSamples.getUnit),
self,
msg)(context.dispatcher)
} else {
// ---- Done ----
log.debug(s"Done with stack trace sample $sampleId.")
- sender ! ResponseStackTraceSampleSuccess(sampleId, executionId, currentTraces)
+
+
+ sender ! new StackTraceSampleResponse(sampleId, executionId, currentTraces)
}
case None =>
@@ -780,7 +769,7 @@ class TaskManager(
s"Either the task is not known to the task manager or it is not running.")
} else {
// Task removed during sampling. Reply with partial result.
- sender ! ResponseStackTraceSampleSuccess(sampleId, executionId, currentTraces)
+ sender ! new StackTraceSampleResponse(sampleId, executionId, currentTraces)
}
}
} else {
@@ -788,7 +777,7 @@ class TaskManager(
}
} catch {
case e: Exception =>
- sender ! ResponseStackTraceSampleFailure(sampleId, executionId, e)
+ sender ! Failure(e)
}
case _ => unhandled(message)
@@ -1029,7 +1018,10 @@ class TaskManager(
// de-register from the JobManager (faster detection of disconnect)
currentJobManager foreach {
- _ ! decorateMessage(Disconnect(instanceID, s"TaskManager ${self.path} is disassociating"))
+ _ ! decorateMessage(
+ Disconnect(
+ instanceID,
+ new Exception(s"TaskManager ${self.path} is disassociating")))
}
currentJobManager = None
@@ -1061,31 +1053,26 @@ class TaskManager(
}
}
- protected def handleJobManagerDisconnect(jobManager: ActorRef, msg: String): Unit = {
- if (isConnected && jobManager != null) {
-
+ protected def handleJobManagerDisconnect(msg: String): Unit = {
+ if (isConnected) {
+ val jobManager = currentJobManager.orNull
// check if it comes from our JobManager
- if (jobManager == currentJobManager.orNull) {
- try {
- val message = s"TaskManager ${self.path} disconnects from JobManager " +
- s"${jobManager.path}: " + msg
- log.info(message)
+ try {
+ val message = s"TaskManager ${self.path} disconnects from JobManager " +
+ s"${jobManager.path}: " + msg
+ log.info(message)
- // cancel all our tasks with a proper error message
- cancelAndClearEverything(new Exception(message))
+ // cancel all our tasks with a proper error message
+ cancelAndClearEverything(new Exception(message))
- // reset our state to disassociated
- disassociateFromJobManager()
- }
- catch {
- // this is pretty bad, it leaves the TaskManager in a state where it cannot
- // cleanly reconnect
- case t: Throwable =>
- killTaskManagerFatal("Failed to disassociate from the JobManager", t)
- }
+ // reset our state to disassociated
+ disassociateFromJobManager()
}
- else {
- log.warn(s"Received erroneous JobManager disconnect message for ${jobManager.path}.")
+ catch {
+ // this is pretty bad, it leaves the TaskManager in a state where it cannot
+ // cleanly reconnect
+ case t: Throwable =>
+ killTaskManagerFatal("Failed to disassociate from the JobManager", t)
}
}
}
@@ -1180,7 +1167,7 @@ class TaskManager(
// all good, we kick off the task, which performs its own initialization
task.startTaskThread()
- sender ! decorateMessage(Acknowledge)
+ sender ! decorateMessage(Acknowledge.get())
}
catch {
case t: Throwable =>
@@ -1197,15 +1184,16 @@ class TaskManager(
*/
private def updateTaskInputPartitions(
executionId: ExecutionAttemptID,
- partitionInfos: Seq[(IntermediateDataSetID, InputChannelDeploymentDescriptor)])
+ partitionInfos: java.lang.Iterable[PartitionInfo])
: Unit = {
Option(runningTasks.get(executionId)) match {
case Some(task) =>
- val errors: Seq[String] = partitionInfos.flatMap { info =>
+ val errors: Iterable[String] = partitionInfos.asScala.flatMap { info =>
- val (resultID, partitionInfo) = info
+ val resultID = info.getIntermediateDataSetID
+ val partitionInfo = info.getInputChannelDeploymentDescriptor
val reader = task.getInputGateById(resultID)
if (reader != null) {
@@ -1236,7 +1224,7 @@ class TaskManager(
}
if (errors.isEmpty) {
- sender ! decorateMessage(Acknowledge)
+ sender ! decorateMessage(Acknowledge.get())
} else {
sender ! decorateMessage(Failure(new Exception(errors.mkString("\n"))))
}
@@ -1244,7 +1232,7 @@ class TaskManager(
case None =>
log.debug(s"Discard update for input partitions of task $executionId : " +
s"task is no longer running.")
- sender ! decorateMessage(Acknowledge)
+ sender ! decorateMessage(Acknowledge.get())
}
}
@@ -1359,7 +1347,7 @@ class TaskManager(
"Thread: " + thread.getName + '\n' + elements.mkString("\n")
}.mkString("\n\n")
- recipient ! decorateMessage(StackTrace(instanceID, stackTraceStr))
+ recipient ! decorateMessage(new StackTrace(instanceID, stackTraceStr))
}
catch {
case e: Exception => log.error("Failed to send stack trace to " + recipient.path, e)
@@ -1402,9 +1390,9 @@ class TaskManager(
case Some(jm) =>
Option(newJobManagerAkkaURL) match {
case Some(newJMAkkaURL) =>
- handleJobManagerDisconnect(jm, s"JobManager $newJMAkkaURL was elected as leader.")
+ handleJobManagerDisconnect(s"JobManager $newJMAkkaURL was elected as leader.")
case None =>
- handleJobManagerDisconnect(jm, s"Old JobManager lost its leadership.")
+ handleJobManagerDisconnect(s"Old JobManager lost its leadership.")
}
case None =>
}
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
index fd425f3..ab41113 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.checkpoint;
import com.google.common.collect.Iterables;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.fs.Path;
@@ -57,7 +58,6 @@ import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
-import scala.concurrent.ExecutionContext;
import java.io.IOException;
import java.io.Serializable;
@@ -71,6 +71,7 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
+import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -81,7 +82,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import static org.mockito.Mockito.any;
+import static org.mockito.Matchers.anyLong;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
@@ -311,10 +312,9 @@ public class CheckpointCoordinatorTest {
// check that the vertices received the trigger checkpoint message
{
- TriggerCheckpoint expectedMessage1 = new TriggerCheckpoint(jid, attemptID1, checkpointId, timestamp);
TriggerCheckpoint expectedMessage2 = new TriggerCheckpoint(jid, attemptID2, checkpointId, timestamp);
- verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(expectedMessage1), eq(attemptID1));
- verify(vertex2, times(1)).sendMessageToCurrentExecution(eq(expectedMessage2), eq(attemptID2));
+ verify(vertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointId), eq(timestamp));
+ verify(vertex2.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointId), eq(timestamp));
}
CheckpointMetaData checkpointMetaData = new CheckpointMetaData(checkpointId, 0L);
@@ -356,10 +356,8 @@ public class CheckpointCoordinatorTest {
// check that the vertices received the new trigger checkpoint message
{
- TriggerCheckpoint expectedMessage1 = new TriggerCheckpoint(jid, attemptID1, checkpointIdNew, checkpointNew.getCheckpointTimestamp());
- TriggerCheckpoint expectedMessage2 = new TriggerCheckpoint(jid, attemptID2, checkpointIdNew, checkpointNew.getCheckpointTimestamp());
- verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(expectedMessage1), eq(attemptID1));
- verify(vertex2, times(1)).sendMessageToCurrentExecution(eq(expectedMessage2), eq(attemptID2));
+ verify(vertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointIdNew), eq(checkpointNew.getCheckpointTimestamp()));
+ verify(vertex2.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointIdNew), eq(checkpointNew.getCheckpointTimestamp()));
}
// decline again, nothing should happen
@@ -454,18 +452,14 @@ public class CheckpointCoordinatorTest {
// check that the vertices received the trigger checkpoint message
{
- TriggerCheckpoint expectedMessage1 = new TriggerCheckpoint(jid, attemptID1, checkpoint1Id, timestamp);
- TriggerCheckpoint expectedMessage2 = new TriggerCheckpoint(jid, attemptID2, checkpoint1Id, timestamp);
- verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(expectedMessage1), eq(attemptID1));
- verify(vertex2, times(1)).sendMessageToCurrentExecution(eq(expectedMessage2), eq(attemptID2));
+ verify(vertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpoint1Id), eq(timestamp));
+ verify(vertex2.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpoint1Id), eq(timestamp));
}
// check that the vertices received the trigger checkpoint message for the second checkpoint
{
- TriggerCheckpoint expectedMessage1 = new TriggerCheckpoint(jid, attemptID1, checkpoint2Id, timestamp + 2);
- TriggerCheckpoint expectedMessage2 = new TriggerCheckpoint(jid, attemptID2, checkpoint2Id, timestamp + 2);
- verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(expectedMessage1), eq(attemptID1));
- verify(vertex2, times(1)).sendMessageToCurrentExecution(eq(expectedMessage2), eq(attemptID2));
+ verify(vertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpoint2Id), eq(timestamp + 2));
+ verify(vertex2.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpoint2Id), eq(timestamp + 2));
}
// decline checkpoint from one of the tasks, this should cancel the checkpoint
@@ -559,10 +553,8 @@ public class CheckpointCoordinatorTest {
// check that the vertices received the trigger checkpoint message
{
- TriggerCheckpoint expectedMessage1 = new TriggerCheckpoint(jid, attemptID1, checkpointId, timestamp);
- TriggerCheckpoint expectedMessage2 = new TriggerCheckpoint(jid, attemptID2, checkpointId, timestamp);
- verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(expectedMessage1), eq(attemptID1));
- verify(vertex2, times(1)).sendMessageToCurrentExecution(eq(expectedMessage2), eq(attemptID2));
+ verify(vertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointId), eq(timestamp));
+ verify(vertex2.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointId), eq(timestamp));
}
CheckpointMetaData checkpointMetaData = new CheckpointMetaData(checkpointId, 0L);
@@ -592,10 +584,8 @@ public class CheckpointCoordinatorTest {
// validate that the relevant tasks got a confirmation message
{
- NotifyCheckpointComplete confirmMessage1 = new NotifyCheckpointComplete(jid, attemptID1, checkpointId, timestamp);
- NotifyCheckpointComplete confirmMessage2 = new NotifyCheckpointComplete(jid, attemptID2, checkpointId, timestamp);
- verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(confirmMessage1), eq(attemptID1));
- verify(vertex2, times(1)).sendMessageToCurrentExecution(eq(confirmMessage2), eq(attemptID2));
+ verify(vertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointId), eq(timestamp));
+ verify(vertex2.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointId), eq(timestamp));
}
CompletedCheckpoint success = coord.getSuccessfulCheckpoints().get(0);
@@ -626,15 +616,11 @@ public class CheckpointCoordinatorTest {
// validate that the relevant tasks got a confirmation message
{
- TriggerCheckpoint expectedMessage1 = new TriggerCheckpoint(jid, attemptID1, checkpointIdNew, timestampNew);
- TriggerCheckpoint expectedMessage2 = new TriggerCheckpoint(jid, attemptID2, checkpointIdNew, timestampNew);
- verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(expectedMessage1), eq(attemptID1));
- verify(vertex2, times(1)).sendMessageToCurrentExecution(eq(expectedMessage2), eq(attemptID2));
-
- NotifyCheckpointComplete confirmMessage1 = new NotifyCheckpointComplete(jid, attemptID1, checkpointIdNew, timestampNew);
- NotifyCheckpointComplete confirmMessage2 = new NotifyCheckpointComplete(jid, attemptID2, checkpointIdNew, timestampNew);
- verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(confirmMessage1), eq(attemptID1));
- verify(vertex2, times(1)).sendMessageToCurrentExecution(eq(confirmMessage2), eq(attemptID2));
+ verify(vertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointIdNew), eq(timestampNew));
+ verify(vertex2.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointIdNew), eq(timestampNew));
+
+ verify(vertex1.getCurrentExecutionAttempt(), times(1)).notifyCheckpointComplete(eq(checkpointIdNew), eq(timestampNew));
+ verify(vertex2.getCurrentExecutionAttempt(), times(1)).notifyCheckpointComplete(eq(checkpointIdNew), eq(timestampNew));
}
coord.shutdown(JobStatus.FINISHED);
@@ -701,10 +687,8 @@ public class CheckpointCoordinatorTest {
long checkpointId1 = pending1.getCheckpointId();
// trigger messages should have been sent
- verify(triggerVertex1, times(1)).sendMessageToCurrentExecution(
- new TriggerCheckpoint(jid, triggerAttemptID1, checkpointId1, timestamp1), triggerAttemptID1);
- verify(triggerVertex2, times(1)).sendMessageToCurrentExecution(
- new TriggerCheckpoint(jid, triggerAttemptID2, checkpointId1, timestamp1), triggerAttemptID2);
+ verify(triggerVertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointId1), eq(timestamp1));
+ verify(triggerVertex2.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointId1), eq(timestamp1));
CheckpointMetaData checkpointMetaData1 = new CheckpointMetaData(checkpointId1, 0L);
@@ -730,10 +714,8 @@ public class CheckpointCoordinatorTest {
CheckpointMetaData checkpointMetaData2 = new CheckpointMetaData(checkpointId2, 0L);
// trigger messages should have been sent
- verify(triggerVertex1, times(1)).sendMessageToCurrentExecution(
- new TriggerCheckpoint(jid, triggerAttemptID1, checkpointId2, timestamp2), triggerAttemptID1);
- verify(triggerVertex2, times(1)).sendMessageToCurrentExecution(
- new TriggerCheckpoint(jid, triggerAttemptID2, checkpointId2, timestamp2), triggerAttemptID2);
+ verify(triggerVertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointId2), eq(timestamp2));
+ verify(triggerVertex2.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointId2), eq(timestamp2));
// we acknowledge the remaining two tasks from the first
// checkpoint and two tasks from the second checkpoint
@@ -748,8 +730,7 @@ public class CheckpointCoordinatorTest {
assertTrue(pending1.isDiscarded());
// the first confirm message should be out
- verify(commitVertex, times(1)).sendMessageToCurrentExecution(
- new NotifyCheckpointComplete(jid, commitAttemptID, checkpointId1, timestamp1), commitAttemptID);
+ verify(commitVertex.getCurrentExecutionAttempt(), times(1)).notifyCheckpointComplete(eq(checkpointId1), eq(timestamp1));
// send the last remaining ack for the second checkpoint
coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, ackAttemptID3, checkpointMetaData2));
@@ -760,8 +741,7 @@ public class CheckpointCoordinatorTest {
assertTrue(pending2.isDiscarded());
// the second commit message should be out
- verify(commitVertex, times(1)).sendMessageToCurrentExecution(
- new NotifyCheckpointComplete(jid, commitAttemptID, checkpointId2, timestamp2), commitAttemptID);
+ verify(commitVertex.getCurrentExecutionAttempt(), times(1)).notifyCheckpointComplete(eq(checkpointId2), eq(timestamp2));
// validate the committed checkpoints
List<CompletedCheckpoint> scs = coord.getSuccessfulCheckpoints();
@@ -841,10 +821,8 @@ public class CheckpointCoordinatorTest {
long checkpointId1 = pending1.getCheckpointId();
// trigger messages should have been sent
- verify(triggerVertex1, times(1)).sendMessageToCurrentExecution(
- new TriggerCheckpoint(jid, triggerAttemptID1, checkpointId1, timestamp1), triggerAttemptID1);
- verify(triggerVertex2, times(1)).sendMessageToCurrentExecution(
- new TriggerCheckpoint(jid, triggerAttemptID2, checkpointId1, timestamp1), triggerAttemptID2);
+ verify(triggerVertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointId1), eq(timestamp1));
+ verify(triggerVertex2.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointId1), eq(timestamp1));
CheckpointMetaData checkpointMetaData1 = new CheckpointMetaData(checkpointId1, 0L);
@@ -868,10 +846,8 @@ public class CheckpointCoordinatorTest {
long checkpointId2 = pending2.getCheckpointId();
// trigger messages should have been sent
- verify(triggerVertex1, times(1)).sendMessageToCurrentExecution(
- new TriggerCheckpoint(jid, triggerAttemptID1, checkpointId2, timestamp2), triggerAttemptID1);
- verify(triggerVertex2, times(1)).sendMessageToCurrentExecution(
- new TriggerCheckpoint(jid, triggerAttemptID2, checkpointId2, timestamp2), triggerAttemptID2);
+ verify(triggerVertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointId2), eq(timestamp2));
+ verify(triggerVertex2.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointId2), eq(timestamp2));
// we acknowledge one more task from the first checkpoint and the second
// checkpoint completely. The second checkpoint should then subsume the first checkpoint
@@ -900,8 +876,7 @@ public class CheckpointCoordinatorTest {
assertTrue(success.getTaskStates().isEmpty());
// the first confirm message should be out
- verify(commitVertex, times(1)).sendMessageToCurrentExecution(
- new NotifyCheckpointComplete(jid, commitAttemptID, checkpointId2, timestamp2), commitAttemptID);
+ verify(commitVertex.getCurrentExecutionAttempt(), times(1)).notifyCheckpointComplete(eq(checkpointId2), eq(timestamp2));
// send the last remaining ack for the first checkpoint. This should not do anything
coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, ackAttemptID3, new CheckpointMetaData(checkpointId1, 0L)));
@@ -979,8 +954,7 @@ public class CheckpointCoordinatorTest {
assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints());
// no confirm message must have been sent
- verify(commitVertex, times(0))
- .sendMessageToCurrentExecution(any(NotifyCheckpointComplete.class), any(ExecutionAttemptID.class));
+ verify(commitVertex.getCurrentExecutionAttempt(), times(0)).notifyCheckpointComplete(anyLong(), anyLong());
coord.shutdown(JobStatus.FINISHED);
}
@@ -1067,6 +1041,8 @@ public class CheckpointCoordinatorTest {
ExecutionVertex commitVertex = mockExecutionVertex(commitAttemptID);
final AtomicInteger numCalls = new AtomicInteger();
+
+ final Execution execution = triggerVertex.getCurrentExecutionAttempt();
doAnswer(new Answer<Void>() {
@@ -1075,9 +1051,8 @@ public class CheckpointCoordinatorTest {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
- TriggerCheckpoint message = (TriggerCheckpoint) invocation.getArguments()[0];
- long id = message.getCheckpointId();
- long ts = message.getTimestamp();
+ long id = (Long) invocation.getArguments()[0];
+ long ts = (Long) invocation.getArguments()[1];
assertTrue(id > lastId);
assertTrue(ts >= lastTs);
@@ -1088,7 +1063,7 @@ public class CheckpointCoordinatorTest {
numCalls.incrementAndGet();
return null;
}
- }).when(triggerVertex).sendMessageToCurrentExecution(any(Serializable.class), any(ExecutionAttemptID.class));
+ }).when(execution).triggerCheckpoint(anyLong(), anyLong());
CheckpointCoordinator coord = new CheckpointCoordinator(
jid,
@@ -1171,16 +1146,15 @@ public class CheckpointCoordinatorTest {
ExecutionVertex vertex1 = mockExecutionVertex(attemptID1);
final AtomicInteger numCalls = new AtomicInteger();
+ final Execution execution = vertex1.getCurrentExecutionAttempt();
doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
- if (invocation.getArguments()[0] instanceof TriggerCheckpoint) {
- numCalls.incrementAndGet();
- }
+ numCalls.incrementAndGet();
return null;
}
- }).when(vertex1).sendMessageToCurrentExecution(any(Serializable.class), any(ExecutionAttemptID.class));
+ }).when(execution).triggerCheckpoint(anyLong(), anyLong());
CheckpointCoordinator coord = new CheckpointCoordinator(
jid,
@@ -1338,10 +1312,8 @@ public class CheckpointCoordinatorTest {
// validate that the relevant tasks got a confirmation message
{
- NotifyCheckpointComplete confirmMessage1 = new NotifyCheckpointComplete(jid, attemptID1, checkpointId, timestamp);
- NotifyCheckpointComplete confirmMessage2 = new NotifyCheckpointComplete(jid, attemptID2, checkpointId, timestamp);
- verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(confirmMessage1), eq(attemptID1));
- verify(vertex2, times(1)).sendMessageToCurrentExecution(eq(confirmMessage2), eq(attemptID2));
+ verify(vertex1.getCurrentExecutionAttempt(), times(1)).notifyCheckpointComplete(eq(checkpointId), eq(timestamp));
+ verify(vertex2.getCurrentExecutionAttempt(), times(1)).notifyCheckpointComplete(eq(checkpointId), eq(timestamp));
}
CompletedCheckpoint success = coord.getSuccessfulCheckpoints().get(0);
@@ -1374,15 +1346,13 @@ public class CheckpointCoordinatorTest {
// validate that the relevant tasks got a confirmation message
{
- TriggerCheckpoint expectedMessage1 = new TriggerCheckpoint(jid, attemptID1, checkpointIdNew, timestampNew);
- TriggerCheckpoint expectedMessage2 = new TriggerCheckpoint(jid, attemptID2, checkpointIdNew, timestampNew);
- verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(expectedMessage1), eq(attemptID1));
- verify(vertex2, times(1)).sendMessageToCurrentExecution(eq(expectedMessage2), eq(attemptID2));
+ verify(vertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointIdNew), eq(timestampNew));
+ verify(vertex2.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointIdNew), eq(timestampNew));
NotifyCheckpointComplete confirmMessage1 = new NotifyCheckpointComplete(jid, attemptID1, checkpointIdNew, timestampNew);
NotifyCheckpointComplete confirmMessage2 = new NotifyCheckpointComplete(jid, attemptID2, checkpointIdNew, timestampNew);
- verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(confirmMessage1), eq(attemptID1));
- verify(vertex2, times(1)).sendMessageToCurrentExecution(eq(confirmMessage2), eq(attemptID2));
+ verify(vertex1.getCurrentExecutionAttempt(), times(1)).notifyCheckpointComplete(eq(checkpointIdNew), eq(timestampNew));
+ verify(vertex2.getCurrentExecutionAttempt(), times(1)).notifyCheckpointComplete(eq(checkpointIdNew), eq(timestampNew));
}
coord.shutdown(JobStatus.FINISHED);
@@ -1493,13 +1463,23 @@ public class CheckpointCoordinatorTest {
final AtomicInteger numCalls = new AtomicInteger();
+ final Execution execution = triggerVertex.getCurrentExecutionAttempt();
+
+ doAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocation) throws Throwable {
+ numCalls.incrementAndGet();
+ return null;
+ }
+ }).when(execution).triggerCheckpoint(anyLong(), anyLong());
+
doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
numCalls.incrementAndGet();
return null;
}
- }).when(triggerVertex).sendMessageToCurrentExecution(any(Serializable.class), any(ExecutionAttemptID.class));
+ }).when(execution).notifyCheckpointComplete(anyLong(), anyLong());
CheckpointCoordinator coord = new CheckpointCoordinator(
jid,
@@ -1531,8 +1511,8 @@ public class CheckpointCoordinatorTest {
assertEquals(maxConcurrentAttempts, numCalls.get());
- verify(triggerVertex, times(maxConcurrentAttempts))
- .sendMessageToCurrentExecution(any(TriggerCheckpoint.class), eq(triggerAttemptID));
+ verify(triggerVertex.getCurrentExecutionAttempt(), times(maxConcurrentAttempts))
+ .triggerCheckpoint(anyLong(), anyLong());
// now, once we acknowledge one checkpoint, it should trigger the next one
coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, ackAttemptID, new CheckpointMetaData(1L, 0L)));
@@ -2562,11 +2542,11 @@ public class CheckpointCoordinatorTest {
ExecutionVertex vertex = mock(ExecutionVertex.class);
final Execution exec = spy(new Execution(
- mock(ExecutionContext.class),
+ mock(Executor.class),
vertex,
1,
1L,
- null
+ Time.milliseconds(500L)
));
when(exec.getAttemptId()).thenReturn(attemptID);
when(exec.getState()).thenReturn(state, successiveStates);
@@ -2924,8 +2904,7 @@ public class CheckpointCoordinatorTest {
{
// check that the vertices received the trigger checkpoint message
- TriggerCheckpoint expectedMessage1 = new TriggerCheckpoint(jid, attemptID1, checkpointId, timestamp);
- verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(expectedMessage1), eq(attemptID1));
+ verify(vertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointId), eq(timestamp));
}
// decline checkpoint, this should cancel the checkpoint and re-trigger with correct properties
@@ -2954,8 +2933,7 @@ public class CheckpointCoordinatorTest {
// check that the vertices received the new trigger checkpoint message
{
- TriggerCheckpoint expectedMessage1 = new TriggerCheckpoint(jid, attemptID1, checkpointIdNew, checkpointNew.getCheckpointTimestamp());
- verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(expectedMessage1), eq(attemptID1));
+ verify(vertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpointIdNew), eq(checkpointNew.getCheckpointTimestamp()));
}
coord.shutdown(JobStatus.FINISHED);
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java
index a8bed46..d352b6d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java
@@ -22,6 +22,7 @@ import akka.actor.ActorSystem;
import akka.testkit.JavaTestKit;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.akka.AkkaUtils;
@@ -94,17 +95,17 @@ public class ExecutionGraphCheckpointCoordinatorTest {
CheckpointIDCounter counter,
CompletedCheckpointStore store) throws Exception {
ExecutionGraph executionGraph = new ExecutionGraph(
- TestingUtils.defaultExecutionContext(),
- new JobID(),
- "test",
- new Configuration(),
- new SerializedValue<>(new ExecutionConfig()),
- new FiniteDuration(1, TimeUnit.DAYS),
- new NoRestartStrategy(),
- Collections.<BlobKey>emptyList(),
- Collections.<URL>emptyList(),
- ClassLoader.getSystemClassLoader(),
- new UnregisteredMetricsGroup());
+ TestingUtils.defaultExecutionContext(),
+ new JobID(),
+ "test",
+ new Configuration(),
+ new SerializedValue<>(new ExecutionConfig()),
+ Time.days(1L),
+ new NoRestartStrategy(),
+ Collections.<BlobKey>emptyList(),
+ Collections.<URL>emptyList(),
+ ClassLoader.getSystemClassLoader(),
+ new UnregisteredMetricsGroup());
executionGraph.enableSnapshotCheckpointing(
100,
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/flink-runtime/src/test/java/org/apache/flink/runtime/client/JobClientActorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/client/JobClientActorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/client/JobClientActorTest.java
index 5792c9a..2e3384f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/client/JobClientActorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/client/JobClientActorTest.java
@@ -30,9 +30,9 @@ import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.akka.FlinkUntypedActor;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.leaderelection.TestingLeaderRetrievalService;
+import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.messages.JobClientMessages;
import org.apache.flink.runtime.messages.JobClientMessages.AttachToJobAndWait;
-import org.apache.flink.runtime.messages.Messages;
import org.apache.flink.util.TestLogger;
import org.junit.AfterClass;
import org.junit.Assert;
@@ -363,7 +363,7 @@ public class JobClientActorTest extends TestLogger {
jobAccepted = true;
if (testFuture != ActorRef.noSender()) {
- testFuture.tell(Messages.getAcknowledge(), getSelf());
+ testFuture.tell(Acknowledge.get(), getSelf());
}
}
else if (message instanceof RegisterJobClient) {
@@ -374,14 +374,14 @@ public class JobClientActorTest extends TestLogger {
jobAccepted = true;
if (testFuture != ActorRef.noSender()) {
- testFuture.tell(Messages.getAcknowledge(), getSelf());
+ testFuture.tell(Acknowledge.get(), getSelf());
}
}
else if (message instanceof RegisterTest) {
testFuture = getSender();
if (jobAccepted) {
- testFuture.tell(Messages.getAcknowledge(), getSelf());
+ testFuture.tell(Acknowledge.get(), getSelf());
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/AllVerticesIteratorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/AllVerticesIteratorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/AllVerticesIteratorTest.java
index 1e66d81..d8ada35 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/AllVerticesIteratorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/AllVerticesIteratorTest.java
@@ -44,7 +44,7 @@ public class AllVerticesIteratorTest {
v4.setParallelism(2);
ExecutionGraph eg = Mockito.mock(ExecutionGraph.class);
- Mockito.when(eg.getExecutionContext()).thenReturn(TestingUtils.directExecutionContext());
+ Mockito.when(eg.getExecutor()).thenReturn(TestingUtils.directExecutionContext());
ExecutionJobVertex ejv1 = new ExecutionJobVertex(eg, v1, 1,
AkkaUtils.getDefaultTimeout());
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
index 2a287af..acccd98 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
@@ -47,6 +47,7 @@ import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
+import org.apache.flink.runtime.jobmanager.slots.ActorTaskManagerGateway;
import org.apache.flink.runtime.operators.BatchTask;
import org.apache.flink.runtime.taskmanager.TaskExecutionState;
import org.apache.flink.runtime.testingUtils.TestingUtils;
@@ -103,7 +104,7 @@ public class ExecutionGraphDeploymentTest {
ExecutionGraphTestUtils.SimpleActorGateway instanceGateway = new ExecutionGraphTestUtils.SimpleActorGateway(TestingUtils.directExecutionContext());
- final Instance instance = getInstance(instanceGateway);
+ final Instance instance = getInstance(new ActorTaskManagerGateway(instanceGateway));
final SimpleSlot slot = instance.allocateSimpleSlot(jobId);
@@ -320,8 +321,9 @@ public class ExecutionGraphDeploymentTest {
for (int i = 0; i < dop1; i++) {
scheduler.newInstanceAvailable(
ExecutionGraphTestUtils.getInstance(
- new ExecutionGraphTestUtils.SimpleActorGateway(
- TestingUtils.directExecutionContext())));
+ new ActorTaskManagerGateway(
+ new ExecutionGraphTestUtils.SimpleActorGateway(
+ TestingUtils.directExecutionContext()))));
}
assertEquals(dop1, scheduler.getNumberOfAvailableSlots());
@@ -362,9 +364,10 @@ public class ExecutionGraphDeploymentTest {
Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
for (int i = 0; i < dop1 + dop2; i++) {
scheduler.newInstanceAvailable(
- ExecutionGraphTestUtils.getInstance(
- new ExecutionGraphTestUtils.SimpleActorGateway(
- TestingUtils.directExecutionContext())));
+ ExecutionGraphTestUtils.getInstance(
+ new ActorTaskManagerGateway(
+ new ExecutionGraphTestUtils.SimpleActorGateway(
+ TestingUtils.directExecutionContext()))));
}
assertEquals(dop1 + dop2, scheduler.getNumberOfAvailableSlots());
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphMetricsTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphMetricsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphMetricsTest.java
index 5b59471..09e7c3e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphMetricsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphMetricsTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.executiongraph;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Gauge;
@@ -30,11 +31,13 @@ import org.apache.flink.runtime.JobException;
import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.concurrent.impl.FlinkCompletableFuture;
+import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.execution.SuppressRestartsException;
import org.apache.flink.runtime.executiongraph.restart.RestartStrategy;
-import org.apache.flink.runtime.instance.ActorGateway;
import org.apache.flink.runtime.instance.Instance;
+import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway;
+import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.metrics.MetricRegistryConfiguration;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
import org.apache.flink.runtime.instance.SimpleSlot;
@@ -45,7 +48,6 @@ import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobmanager.Tasks;
import org.apache.flink.runtime.jobmanager.scheduler.ScheduledUnit;
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
-import org.apache.flink.runtime.messages.Messages;
import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
import org.apache.flink.runtime.taskmanager.TaskExecutionState;
@@ -55,10 +57,6 @@ import org.apache.flink.util.TestLogger;
import org.junit.Test;
import org.mockito.Matchers;
-import scala.concurrent.ExecutionContext$;
-import scala.concurrent.Future$;
-import scala.concurrent.duration.FiniteDuration;
-
import java.io.IOException;
import java.net.URL;
import java.util.ArrayList;
@@ -68,7 +66,6 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
@@ -99,7 +96,7 @@ public class ExecutionGraphMetricsTest extends TestLogger {
Configuration jobConfig = new Configuration();
- FiniteDuration timeout = new FiniteDuration(10, TimeUnit.SECONDS);
+ Time timeout = Time.seconds(10L);
MetricRegistry metricRegistry = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(config));
@@ -121,12 +118,12 @@ public class ExecutionGraphMetricsTest extends TestLogger {
when(taskManagerLocation.getResourceID()).thenReturn(taskManagerId);
when(taskManagerLocation.getHostname()).thenReturn("localhost");
- ActorGateway actorGateway = mock(ActorGateway.class);
+ TaskManagerGateway taskManagerGateway = mock(TaskManagerGateway.class);
Instance instance = mock(Instance.class);
when(instance.getTaskManagerLocation()).thenReturn(taskManagerLocation);
when(instance.getTaskManagerID()).thenReturn(taskManagerId);
- when(instance.getActorGateway()).thenReturn(actorGateway);
+ when(instance.getTaskManagerGateway()).thenReturn(taskManagerGateway);
Slot rootSlot = mock(Slot.class);
@@ -134,7 +131,7 @@ public class ExecutionGraphMetricsTest extends TestLogger {
when(simpleSlot.isAlive()).thenReturn(true);
when(simpleSlot.getTaskManagerLocation()).thenReturn(taskManagerLocation);
when(simpleSlot.getTaskManagerID()).thenReturn(taskManagerId);
- when(simpleSlot.getTaskManagerActorGateway()).thenReturn(actorGateway);
+ when(simpleSlot.getTaskManagerGateway()).thenReturn(taskManagerGateway);
when(simpleSlot.setExecutedVertex(Matchers.any(Execution.class))).thenReturn(true);
when(simpleSlot.getRoot()).thenReturn(rootSlot);
@@ -144,12 +141,12 @@ public class ExecutionGraphMetricsTest extends TestLogger {
when(rootSlot.getSlotNumber()).thenReturn(0);
- when(actorGateway.ask(Matchers.any(Object.class), Matchers.any(FiniteDuration.class))).thenReturn(Future$.MODULE$.<Object>successful(Messages.getAcknowledge()));
+ when(taskManagerGateway.submitTask(any(TaskDeploymentDescriptor.class), any(Time.class))).thenReturn(FlinkCompletableFuture.completed(Acknowledge.get()));
TestingRestartStrategy testingRestartStrategy = new TestingRestartStrategy();
ExecutionGraph executionGraph = new ExecutionGraph(
- ExecutionContext$.MODULE$.fromExecutor(executor),
+ executor,
jobGraph.getJobID(),
jobGraph.getName(),
jobConfig,
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java
index 7b2e20d..d6770a6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java
@@ -41,6 +41,7 @@ import org.apache.flink.runtime.jobmanager.Tasks;
import org.apache.flink.runtime.jobmanager.scheduler.CoLocationConstraint;
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.runtime.jobmanager.slots.ActorTaskManagerGateway;
import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.flink.util.SerializedValue;
import org.apache.flink.util.TestLogger;
@@ -97,7 +98,8 @@ public class ExecutionGraphRestartTest extends TestLogger {
//setting up
Instance instance = ExecutionGraphTestUtils.getInstance(
- new SimpleActorGateway(TestingUtils.directExecutionContext()),
+ new ActorTaskManagerGateway(
+ new SimpleActorGateway(TestingUtils.directExecutionContext())),
NUM_TASKS);
Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
@@ -221,7 +223,8 @@ public class ExecutionGraphRestartTest extends TestLogger {
Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
Instance instance = ExecutionGraphTestUtils.getInstance(
- new SimpleActorGateway(TestingUtils.directExecutionContext()),
+ new ActorTaskManagerGateway(
+ new SimpleActorGateway(TestingUtils.directExecutionContext())),
NUM_TASKS);
scheduler.newInstanceAvailable(instance);
@@ -369,7 +372,8 @@ public class ExecutionGraphRestartTest extends TestLogger {
@Test
public void testFailingExecutionAfterRestart() throws Exception {
Instance instance = ExecutionGraphTestUtils.getInstance(
- new SimpleActorGateway(TestingUtils.directExecutionContext()),
+ new ActorTaskManagerGateway(
+ new SimpleActorGateway(TestingUtils.directExecutionContext())),
2);
Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
@@ -432,8 +436,9 @@ public class ExecutionGraphRestartTest extends TestLogger {
@Test
public void testFailExecutionAfterCancel() throws Exception {
Instance instance = ExecutionGraphTestUtils.getInstance(
- new SimpleActorGateway(TestingUtils.directExecutionContext()),
- 2);
+ new ActorTaskManagerGateway(
+ new SimpleActorGateway(TestingUtils.directExecutionContext())),
+ 2);
Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
scheduler.newInstanceAvailable(instance);
@@ -477,8 +482,9 @@ public class ExecutionGraphRestartTest extends TestLogger {
@Test
public void testFailExecutionGraphAfterCancel() throws Exception {
Instance instance = ExecutionGraphTestUtils.getInstance(
- new SimpleActorGateway(TestingUtils.directExecutionContext()),
- 2);
+ new ActorTaskManagerGateway(
+ new SimpleActorGateway(TestingUtils.directExecutionContext())),
+ 2);
Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
scheduler.newInstanceAvailable(instance);
@@ -524,7 +530,8 @@ public class ExecutionGraphRestartTest extends TestLogger {
Deadline deadline = timeout.fromNow();
Instance instance = ExecutionGraphTestUtils.getInstance(
- new SimpleActorGateway(TestingUtils.directExecutionContext()),
+ new ActorTaskManagerGateway(
+ new SimpleActorGateway(TestingUtils.directExecutionContext())),
NUM_TASKS);
Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
@@ -639,8 +646,9 @@ public class ExecutionGraphRestartTest extends TestLogger {
private static Tuple2<ExecutionGraph, Instance> createExecutionGraph(RestartStrategy restartStrategy, boolean isSpy) throws Exception {
Instance instance = ExecutionGraphTestUtils.getInstance(
- new SimpleActorGateway(TestingUtils.directExecutionContext()),
- NUM_TASKS);
+ new ActorTaskManagerGateway(
+ new SimpleActorGateway(TestingUtils.directExecutionContext())),
+ NUM_TASKS);
Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
scheduler.newInstanceAvailable(instance);
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
index df47c3a..28dff02 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
@@ -24,6 +24,7 @@ import static org.mockito.Mockito.spy;
import java.lang.reflect.Field;
import java.net.InetAddress;
+import java.util.concurrent.Executor;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.configuration.Configuration;
@@ -35,8 +36,9 @@ import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy;
import org.apache.flink.runtime.instance.BaseTestingActorGateway;
import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.instance.Instance;
+import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway;
+import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
-import org.apache.flink.runtime.instance.ActorGateway;
import org.apache.flink.runtime.instance.InstanceID;
import org.apache.flink.runtime.instance.SimpleSlot;
import org.apache.flink.runtime.jobgraph.JobVertex;
@@ -44,11 +46,9 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-import org.apache.flink.runtime.messages.Messages;
import org.apache.flink.runtime.messages.TaskMessages.SubmitTask;
import org.apache.flink.runtime.messages.TaskMessages.FailIntermediateResultPartitions;
import org.apache.flink.runtime.messages.TaskMessages.CancelTask;
-import org.apache.flink.runtime.messages.TaskMessages.TaskOperationResult;
import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.flink.util.SerializedValue;
import org.mockito.Matchers;
@@ -103,11 +103,11 @@ public class ExecutionGraphTestUtils {
// utility mocking methods
// --------------------------------------------------------------------------------------------
- public static Instance getInstance(final ActorGateway gateway) throws Exception {
+ public static Instance getInstance(final TaskManagerGateway gateway) throws Exception {
return getInstance(gateway, 1);
}
- public static Instance getInstance(final ActorGateway gateway, final int numberOfSlots) throws Exception {
+ public static Instance getInstance(final TaskManagerGateway gateway, final int numberOfSlots) throws Exception {
ResourceID resourceID = ResourceID.generate();
HardwareDescription hardwareDescription = new HardwareDescription(4, 2L*1024*1024*1024, 1024*1024*1024, 512*1024*1024);
InetAddress address = InetAddress.getByName("127.0.0.1");
@@ -132,11 +132,11 @@ public class ExecutionGraphTestUtils {
SubmitTask submitTask = (SubmitTask) message;
lastTDD = submitTask.tasks();
- result = Messages.getAcknowledge();
+ result = Acknowledge.get();
} else if(message instanceof CancelTask) {
CancelTask cancelTask = (CancelTask) message;
- result = new TaskOperationResult(cancelTask.attemptID(), true);
+ result = Acknowledge.get();
} else if(message instanceof FailIntermediateResultPartitions) {
result = new Object();
}
@@ -159,7 +159,7 @@ public class ExecutionGraphTestUtils {
} else if (message instanceof CancelTask) {
CancelTask cancelTask = (CancelTask) message;
- return new TaskOperationResult(cancelTask.attemptID(), true);
+ return Acknowledge.get();
} else {
return null;
}
@@ -168,12 +168,12 @@ public class ExecutionGraphTestUtils {
public static final String ERROR_MESSAGE = "test_failure_error_message";
- public static ExecutionJobVertex getExecutionVertex(JobVertexID id, ExecutionContext executionContext) throws Exception {
+ public static ExecutionJobVertex getExecutionVertex(JobVertexID id, Executor executor) throws Exception {
JobVertex ajv = new JobVertex("TestVertex", id);
ajv.setInvokableClass(mock(AbstractInvokable.class).getClass());
ExecutionGraph graph = new ExecutionGraph(
- executionContext,
+ executor,
new JobID(),
"test job",
new Configuration(),
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
index f3743c1..8434ed7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
@@ -34,6 +34,7 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.jobmanager.slots.ActorTaskManagerGateway;
import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.flink.util.SerializedValue;
import org.junit.Test;
@@ -67,8 +68,9 @@ public class ExecutionStateProgressTest {
// mock resources and mock taskmanager
for (ExecutionVertex ee : ejv.getTaskVertices()) {
SimpleSlot slot = getInstance(
+ new ActorTaskManagerGateway(
new SimpleActorGateway(
- TestingUtils.defaultExecutionContext())
+ TestingUtils.defaultExecutionContext()))
).allocateSimpleSlot(jid);
ee.deployToSlot(slot);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/cfb3790f/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java
index 5c87315..b10b796 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java
@@ -34,10 +34,10 @@ import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
-import org.apache.flink.runtime.messages.Messages;
+import org.apache.flink.runtime.jobmanager.slots.ActorTaskManagerGateway;
+import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.messages.TaskMessages.SubmitTask;
import org.apache.flink.runtime.messages.TaskMessages.CancelTask;
-import org.apache.flink.runtime.messages.TaskMessages.TaskOperationResult;
import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.junit.Test;
@@ -127,11 +127,9 @@ public class ExecutionVertexCancelTest {
assertEquals(ExecutionState.SCHEDULED, vertex.getExecutionState());
ActorGateway actorGateway = new CancelSequenceActorGateway(
- executionContext,
- new TaskOperationResult(execId, true),
- new TaskOperationResult(execId, false));
+ executionContext, 2);
- Instance instance = getInstance(actorGateway);
+ Instance instance = getInstance(new ActorTaskManagerGateway(actorGateway));
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
vertex.deployToSlot(slot);
@@ -198,11 +196,9 @@ public class ExecutionVertexCancelTest {
// first return NOT SUCCESS (task not found, cancel call overtook deploy call), then success (cancel call after deploy call)
ActorGateway actorGateway = new CancelSequenceActorGateway(
executionContext,
- new TaskOperationResult(execId, false),
- new TaskOperationResult(execId, true)
- );
+ 2);
- Instance instance = getInstance(actorGateway);
+ Instance instance = getInstance(new ActorTaskManagerGateway(actorGateway));
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
vertex.deployToSlot(slot);
@@ -261,9 +257,9 @@ public class ExecutionVertexCancelTest {
ActorGateway actorGateway = new CancelSequenceActorGateway(
TestingUtils.directExecutionContext(),
- new TaskOperationResult(execId, true));
+ 1);
- Instance instance = getInstance(actorGateway);
+ Instance instance = getInstance(new ActorTaskManagerGateway(actorGateway));
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
setVertexState(vertex, ExecutionState.RUNNING);
@@ -302,10 +298,9 @@ public class ExecutionVertexCancelTest {
final ActorGateway actorGateway = new CancelSequenceActorGateway(
TestingUtils.directExecutionContext(),
- new TaskOperationResult(execId, true)
- );
+ 1);
- Instance instance = getInstance(actorGateway);
+ Instance instance = getInstance(new ActorTaskManagerGateway(actorGateway));
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
setVertexState(vertex, ExecutionState.RUNNING);
@@ -353,10 +348,9 @@ public class ExecutionVertexCancelTest {
final ActorGateway actorGateway = new CancelSequenceActorGateway(
TestingUtils.directExecutionContext(),
- new TaskOperationResult(execId, false)
- );
+ 1);
- Instance instance = getInstance(actorGateway);
+ Instance instance = getInstance(new ActorTaskManagerGateway(actorGateway));
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
setVertexState(vertex, ExecutionState.RUNNING);
@@ -387,9 +381,9 @@ public class ExecutionVertexCancelTest {
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
- final ActorGateway gateway = new CancelSequenceActorGateway(TestingUtils.directExecutionContext());
+ final ActorGateway gateway = new CancelSequenceActorGateway(TestingUtils.directExecutionContext(), 0);
- Instance instance = getInstance(gateway);
+ Instance instance = getInstance(new ActorTaskManagerGateway(gateway));
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
setVertexState(vertex, ExecutionState.RUNNING);
@@ -420,13 +414,12 @@ public class ExecutionVertexCancelTest {
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
- final ExecutionAttemptID execID = vertex.getCurrentExecutionAttempt().getAttemptId();
final ActorGateway gateway = new CancelSequenceActorGateway(
TestingUtils.defaultExecutionContext(),
- new TaskOperationResult(execID, true));
+ 1);
- Instance instance = getInstance(gateway);
+ Instance instance = getInstance(new ActorTaskManagerGateway(gateway));
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
setVertexState(vertex, ExecutionState.RUNNING);
@@ -481,7 +474,7 @@ public class ExecutionVertexCancelTest {
// deploying after canceling from CREATED needs to raise an exception, because
// the scheduler (or any caller) needs to know that the slot should be released
try {
- Instance instance = getInstance(DummyActorGateway.INSTANCE);
+ Instance instance = getInstance(new ActorTaskManagerGateway(DummyActorGateway.INSTANCE));
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
vertex.deployToSlot(slot);
@@ -524,7 +517,7 @@ public class ExecutionVertexCancelTest {
AkkaUtils.getDefaultTimeout());
setVertexState(vertex, ExecutionState.CANCELING);
- Instance instance = getInstance(DummyActorGateway.INSTANCE);
+ Instance instance = getInstance(new ActorTaskManagerGateway(DummyActorGateway.INSTANCE));
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
vertex.deployToSlot(slot);
@@ -540,7 +533,7 @@ public class ExecutionVertexCancelTest {
ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
- Instance instance = getInstance(DummyActorGateway.INSTANCE);
+ Instance instance = getInstance(new ActorTaskManagerGateway(DummyActorGateway.INSTANCE));
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
setVertexResource(vertex, slot);
@@ -561,26 +554,26 @@ public class ExecutionVertexCancelTest {
}
public static class CancelSequenceActorGateway extends BaseTestingActorGateway {
- private final TaskOperationResult[] results;
+ private final int successfulOperations;
private int index = -1;
- public CancelSequenceActorGateway(ExecutionContext executionContext, TaskOperationResult... result) {
+ public CancelSequenceActorGateway(ExecutionContext executionContext, int successfulOperations) {
super(executionContext);
- this.results = result;
+ this.successfulOperations = successfulOperations;
}
@Override
public Object handleMessage(Object message) throws Exception {
Object result;
if(message instanceof SubmitTask) {
- result = Messages.getAcknowledge();
+ result = Acknowledge.get();
} else if(message instanceof CancelTask) {
index++;
- if(index >= results.length){
+ if(index >= successfulOperations){
throw new IOException("RPC call failed.");
} else {
- result = results[index];
+ result = Acknowledge.get();
}
} else {
result = null;