http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java index f166c36..5e276bf 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java @@ -18,8 +18,6 @@ package org.apache.flink.runtime.taskmanager; -import akka.actor.ActorRef; - import org.apache.flink.api.common.JobID; import org.apache.flink.configuration.Configuration; import org.apache.flink.core.fs.Path; @@ -27,6 +25,7 @@ import org.apache.flink.runtime.accumulators.AccumulatorRegistry; import org.apache.flink.runtime.broadcast.BroadcastVariableManager; import org.apache.flink.runtime.execution.Environment; import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; +import org.apache.flink.runtime.instance.ActorGateway; import org.apache.flink.runtime.io.disk.iomanager.IOManager; import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter; import org.apache.flink.runtime.io.network.partition.consumer.InputGate; @@ -72,25 +71,32 @@ public class RuntimeEnvironment implements Environment { private final ResultPartitionWriter[] writers; private final InputGate[] inputGates; - private final ActorRef jobManagerActor; + private final ActorGateway jobManager; private final AccumulatorRegistry accumulatorRegistry; // ------------------------------------------------------------------------ - public RuntimeEnvironment(JobID jobId, JobVertexID jobVertexId, ExecutionAttemptID executionId, - String taskName, String taskNameWithSubtasks, - int subtaskIndex, int parallelism, - Configuration jobConfiguration, Configuration taskConfiguration, - ClassLoader userCodeClassLoader, - MemoryManager memManager, IOManager ioManager, - BroadcastVariableManager bcVarManager, + public RuntimeEnvironment( + JobID jobId, + JobVertexID jobVertexId, + ExecutionAttemptID executionId, + String taskName, + String taskNameWithSubtasks, + int subtaskIndex, + int parallelism, + Configuration jobConfiguration, + Configuration taskConfiguration, + ClassLoader userCodeClassLoader, + MemoryManager memManager, + IOManager ioManager, + BroadcastVariableManager bcVarManager, AccumulatorRegistry accumulatorRegistry, - InputSplitProvider splitProvider, - Map<String, Future<Path>> distCacheEntries, - ResultPartitionWriter[] writers, - InputGate[] inputGates, - ActorRef jobManagerActor) { + InputSplitProvider splitProvider, + Map<String, Future<Path>> distCacheEntries, + ResultPartitionWriter[] writers, + InputGate[] inputGates, + ActorGateway jobManager) { checkArgument(parallelism > 0 && subtaskIndex >= 0 && subtaskIndex < parallelism); @@ -112,7 +118,7 @@ public class RuntimeEnvironment implements Environment { this.distCacheEntries = checkNotNull(distCacheEntries); this.writers = checkNotNull(writers); this.inputGates = checkNotNull(inputGates); - this.jobManagerActor = checkNotNull(jobManagerActor); + this.jobManager = checkNotNull(jobManager); } @@ -238,6 +244,6 @@ public class RuntimeEnvironment implements Environment { } AcknowledgeCheckpoint message = new AcknowledgeCheckpoint(jobId, executionId, checkpointId, serializedState); - jobManagerActor.tell(message, ActorRef.noSender()); + jobManager.tell(message); } }
http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java index 13a2ace..c4f62fb 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java @@ -18,8 +18,6 @@ package org.apache.flink.runtime.taskmanager; -import akka.actor.ActorRef; -import akka.util.Timeout; import org.apache.flink.api.common.JobID; import org.apache.flink.api.common.cache.DistributedCache; @@ -37,6 +35,7 @@ import org.apache.flink.runtime.execution.ExecutionState; import org.apache.flink.runtime.execution.librarycache.LibraryCacheManager; import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; import org.apache.flink.runtime.filecache.FileCache; +import org.apache.flink.runtime.instance.ActorGateway; import org.apache.flink.runtime.io.disk.iomanager.IOManager; import org.apache.flink.runtime.io.network.NetworkEnvironment; import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter; @@ -163,17 +162,17 @@ public class Task implements Runnable { private final Map<IntermediateDataSetID, SingleInputGate> inputGatesById; - /** The TaskManager actor that spawned this task */ - private final ActorRef taskManager; + /** Gateway to the TaskManager that spawned this task */ + private final ActorGateway taskManager; - /** The JobManager actor */ - private final ActorRef jobManager; + /** Gateway to the JobManager */ + private final ActorGateway jobManager; /** All actors that want to be notified about changes in the task's execution state */ - private final List<ActorRef> executionListenerActors; + private final List<ActorGateway> executionListenerActors; /** The timeout for all ask operations on actors */ - private final Timeout actorAskTimeout; + private final FiniteDuration actorAskTimeout; /** The library cache, from which the task can request its required JAR files */ private final LibraryCacheManager libraryCache; @@ -224,8 +223,8 @@ public class Task implements Runnable { IOManager ioManager, NetworkEnvironment networkEnvironment, BroadcastVariableManager bcVarManager, - ActorRef taskManagerActor, - ActorRef jobManagerActor, + ActorGateway taskManagerActor, + ActorGateway jobManagerActor, FiniteDuration actorAskTimeout, LibraryCacheManager libraryCache, FileCache fileCache) @@ -254,13 +253,13 @@ public class Task implements Runnable { this.jobManager = checkNotNull(jobManagerActor); this.taskManager = checkNotNull(taskManagerActor); - this.actorAskTimeout = new Timeout(checkNotNull(actorAskTimeout)); + this.actorAskTimeout = checkNotNull(actorAskTimeout); this.libraryCache = checkNotNull(libraryCache); this.fileCache = checkNotNull(fileCache); this.network = checkNotNull(networkEnvironment); - this.executionListenerActors = new CopyOnWriteArrayList<ActorRef>(); + this.executionListenerActors = new CopyOnWriteArrayList<ActorGateway>(); // create the reader and writer structures @@ -568,7 +567,7 @@ public class Task implements Runnable { // to know this! notifyObservers(ExecutionState.RUNNING, null); taskManager.tell(new TaskMessages.UpdateTaskExecutionState( - new TaskExecutionState(jobId, executionId, ExecutionState.RUNNING)), ActorRef.noSender()); + new TaskExecutionState(jobId, executionId, ExecutionState.RUNNING))); // make sure the user code classloader is accessible thread-locally executingThread.setContextClassLoader(userCodeClassLoader); @@ -750,11 +749,11 @@ public class Task implements Runnable { } private void notifyFinalState() { - taskManager.tell(new TaskInFinalState(executionId), ActorRef.noSender()); + taskManager.tell(new TaskInFinalState(executionId)); } private void notifyFatalError(String message, Throwable cause) { - taskManager.tell(new FatalError(message, cause), ActorRef.noSender()); + taskManager.tell(new FatalError(message, cause)); } // ---------------------------------------------------------------------------------------------------------------- @@ -839,14 +838,10 @@ public class Task implements Runnable { // State Listeners // ------------------------------------------------------------------------ - public void registerExecutionListener(ActorRef listener) { + public void registerExecutionListener(ActorGateway listener) { executionListenerActors.add(listener); } - public void unregisterExecutionListener(ActorRef listener) { - executionListenerActors.remove(listener); - } - private void notifyObservers(ExecutionState newState, Throwable error) { if (error == null) { LOG.info(taskNameWithSubtask + " switched to " + newState); @@ -859,8 +854,8 @@ public class Task implements Runnable { TaskMessages.UpdateTaskExecutionState actorMessage = new TaskMessages.UpdateTaskExecutionState(stateUpdate); - for (ActorRef listener : executionListenerActors) { - listener.tell(actorMessage, ActorRef.noSender()); + for (ActorGateway listener : executionListenerActors) { + listener.tell(actorMessage); } } http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskInputSplitProvider.java ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskInputSplitProvider.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskInputSplitProvider.java index 5a69850..cb78c16 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskInputSplitProvider.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskInputSplitProvider.java @@ -18,13 +18,10 @@ package org.apache.flink.runtime.taskmanager; -import akka.actor.ActorRef; - -import akka.pattern.Patterns; -import akka.util.Timeout; import org.apache.flink.core.io.InputSplit; import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; import org.apache.flink.api.common.JobID; +import org.apache.flink.runtime.instance.ActorGateway; import org.apache.flink.runtime.jobgraph.JobVertexID; import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider; import org.apache.flink.runtime.messages.JobManagerMessages; @@ -32,10 +29,11 @@ import org.apache.flink.util.InstantiationUtil; import scala.concurrent.Await; import scala.concurrent.Future; +import scala.concurrent.duration.FiniteDuration; public class TaskInputSplitProvider implements InputSplitProvider { - private final ActorRef jobManager; + private final ActorGateway jobManager; private final JobID jobId; @@ -45,11 +43,15 @@ public class TaskInputSplitProvider implements InputSplitProvider { private final ClassLoader usercodeClassLoader; - private final Timeout timeout; + private final FiniteDuration timeout; - public TaskInputSplitProvider(ActorRef jobManager, JobID jobId, JobVertexID vertexId, - ExecutionAttemptID executionID, ClassLoader userCodeClassLoader, - Timeout timeout) + public TaskInputSplitProvider( + ActorGateway jobManager, + JobID jobId, + JobVertexID vertexId, + ExecutionAttemptID executionID, + ClassLoader userCodeClassLoader, + FiniteDuration timeout) { this.jobManager = jobManager; this.jobId = jobId; @@ -62,11 +64,11 @@ public class TaskInputSplitProvider implements InputSplitProvider { @Override public InputSplit getNextInputSplit() { try { - final Future<Object> response = Patterns.ask(jobManager, + final Future<Object> response = jobManager.ask( new JobManagerMessages.RequestNextInputSplit(jobId, vertexId, executionID), timeout); - final Object result = Await.result(response, timeout.duration()); + final Object result = Await.result(response, timeout); if(!(result instanceof JobManagerMessages.NextInputSplit)){ throw new RuntimeException("RequestNextInputSplit requires a response of type " + http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorLogMessages.scala ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorLogMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorLogMessages.scala deleted file mode 100644 index c74c339..0000000 --- a/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorLogMessages.scala +++ /dev/null @@ -1,52 +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 - -import _root_.akka.actor.Actor - -/** - * Mixin to add debug message logging - */ -trait ActorLogMessages { - that: Actor with ActorSynchronousLogging => - - override def receive: Receive = new Actor.Receive { - private val _receiveWithLogMessages = receiveWithLogMessages - - override def isDefinedAt(x: Any): Boolean = _receiveWithLogMessages.isDefinedAt(x) - - override def apply(x: Any): Unit = { - if (!log.isDebugEnabled) { - _receiveWithLogMessages(x) - } - else { - log.debug(s"Received message $x at ${that.self.path} from ${that.sender()}.") - - val start = System.nanoTime() - - _receiveWithLogMessages(x) - - val duration = (System.nanoTime() - start) / 1000000 - log.debug(s"Handled message $x in $duration ms from ${that.sender()}.") - } - } - } - - def receiveWithLogMessages: Receive -} http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorSynchronousLogging.scala ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorSynchronousLogging.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorSynchronousLogging.scala deleted file mode 100644 index 4d3a988..0000000 --- a/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorSynchronousLogging.scala +++ /dev/null @@ -1,31 +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 - -import _root_.akka.actor.Actor -import grizzled.slf4j.Logger - -/** Adds a logger to an [[akka.actor.Actor]] implementation - * - */ -trait ActorSynchronousLogging { - self: Actor => - - lazy val log = Logger(getClass) -} http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/flink-runtime/src/main/scala/org/apache/flink/runtime/FlinkActor.scala ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/FlinkActor.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/FlinkActor.scala new file mode 100644 index 0000000..74a9b07 --- /dev/null +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/FlinkActor.scala @@ -0,0 +1,48 @@ +/* + * 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 + +import _root_.akka.actor.Actor +import grizzled.slf4j.Logger + +/** Base trait for Flink's actors. + * + * The message handling logic is defined in the handleMessage method. This allows to mixin + * stackable traits which change the message receiving behaviour. + */ +trait FlinkActor extends Actor { + val log: Logger + + override def receive: Receive = handleMessage + + /** Handle incoming messages + * + * @return + */ + def handleMessage: Receive + + /** Factory method for messages. This method can be used by mixins to decorate messages + * + * @param message The message to decorate + * @return The decorated message + */ + def decorateMessage(message: Any): Any = { + message + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/flink-runtime/src/main/scala/org/apache/flink/runtime/LeaderSessionMessageDecorator.scala ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/LeaderSessionMessageDecorator.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/LeaderSessionMessageDecorator.scala new file mode 100644 index 0000000..c6793ed --- /dev/null +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/LeaderSessionMessageDecorator.scala @@ -0,0 +1,46 @@ +/* + * 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 + +import java.util.UUID + +import org.apache.flink.runtime.messages.JobManagerMessages.LeaderSessionMessage +import org.apache.flink.runtime.messages.RequiresLeaderSessionID + +/** [[MessageDecorator]] which wraps [[RequiresLeaderSessionID]] messages in a + * [[LeaderSessionMessage]] with the given leader session ID. + * + * @param leaderSessionID Leader session ID which is associated with the + * [[RequiresLeaderSessionID]] message + */ +class LeaderSessionMessageDecorator(val leaderSessionID: Option[UUID]) extends MessageDecorator { + + /** Wraps [[RequiresLeaderSessionID]] messages in a [[LeaderSessionMessage]]. + * + * @param message Message to decorate + * @return Decorated message + */ + override def decorate(message: Any): Any = { + message match { + case msg: RequiresLeaderSessionID => + LeaderSessionMessage(leaderSessionID, msg) + case msg => msg + } + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/flink-runtime/src/main/scala/org/apache/flink/runtime/LeaderSessionMessages.scala ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/LeaderSessionMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/LeaderSessionMessages.scala new file mode 100644 index 0000000..d54926d --- /dev/null +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/LeaderSessionMessages.scala @@ -0,0 +1,73 @@ +/* + * 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 + +import java.util.UUID + +import org.apache.flink.runtime.messages.JobManagerMessages.LeaderSessionMessage +import org.apache.flink.runtime.messages.RequiresLeaderSessionID + +/** Mixin to filter out [[LeaderSessionMessage]] which contain an invalid leader session id. + * Messages which contain a valid leader session ID are unwrapped and forwarded to the actor. + * + */ +trait LeaderSessionMessages extends FlinkActor { + protected def leaderSessionID: Option[UUID] + + abstract override def receive: Receive = { + case LeaderSessionMessage(id, msg) => + // Filter out messages which have not the correct leader session ID + (leaderSessionID, id) match { + case (Some(currentID), Some(msgID)) => + if(currentID.equals(msgID)) { + // correct leader session ID + super.receive(msg) + } else { + // discard message because of incorrect leader session ID + handleDiscardedMessage(msg) + } + + case _ => handleDiscardedMessage(msg) + } + case msg: RequiresLeaderSessionID => + throw new Exception(s"Received a message $msg without a leader session ID, even though" + + " it requires to have one.") + case msg => + // pass the message to the parent's receive method for further processing + super.receive(msg) + } + + private def handleDiscardedMessage(msg: Any): Unit = { + log.debug(s"Discard message $msg because the leader session ID was not correct.") + } + + /** Wrap [[RequiresLeaderSessionID]] messages in a [[LeaderSessionMessage]] + * + * @param message The message to decorate + * @return The decorated message + */ + override def decorateMessage(message: Any): Any = { + message match { + case msg: RequiresLeaderSessionID => + LeaderSessionMessage(leaderSessionID, super.decorateMessage(msg)) + + case msg => super.decorateMessage(msg) + } + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/flink-runtime/src/main/scala/org/apache/flink/runtime/LogMessages.scala ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/LogMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/LogMessages.scala new file mode 100644 index 0000000..e1e8961 --- /dev/null +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/LogMessages.scala @@ -0,0 +1,48 @@ +/* + * 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 + +/** Mixin to add message logging if the debug log level is activated + * + */ +trait LogMessages extends FlinkActor { + abstract override def receive: Receive = { + val _receive = super.receive + + new Receive { + override def isDefinedAt(x: Any): Boolean = _receive.isDefinedAt(x) + + override def apply(x: Any): Unit = { + if (!log.isDebugEnabled) { + _receive(x) + } + else { + log.debug(s"Received message $x at ${context.self.path} from ${context.sender()}.") + + val start = System.nanoTime() + + _receive(x) + + val duration = (System.nanoTime() - start) / 1000000 + log.debug(s"Handled message $x in $duration ms from ${context.sender()}.") + } + } + } + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/flink-runtime/src/main/scala/org/apache/flink/runtime/MessageDecorator.scala ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/MessageDecorator.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/MessageDecorator.scala new file mode 100644 index 0000000..5b1700f --- /dev/null +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/MessageDecorator.scala @@ -0,0 +1,32 @@ +/* + * 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 + +/** Base trait for message decorators + * + */ +trait MessageDecorator { + + /** Decorates a message + * + * @param message Message to decorate + * @return Decorated message + */ + def decorate(message: Any): Any +} http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala index d38e503..b8cce41 100644 --- a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala @@ -62,8 +62,10 @@ object AkkaUtils { * parameter is None, then a local actor system will be created. * @return created actor system */ - def createActorSystem(configuration: Configuration, - listeningAddress: Option[(String, Int)]): ActorSystem = { + def createActorSystem( + configuration: Configuration, + listeningAddress: Option[(String, Int)]) + : ActorSystem = { val akkaConfig = getAkkaConfig(configuration, listeningAddress) createActorSystem(akkaConfig) } http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/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 4325e41..7bf4447 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,7 +21,7 @@ package org.apache.flink.runtime.jobmanager import java.io.{File, IOException} import java.lang.reflect.{InvocationTargetException, Constructor} import java.net.InetSocketAddress -import java.util.Collections +import java.util.{UUID, Collections} import akka.actor.Status.{Failure, Success} import akka.actor._ @@ -51,10 +51,11 @@ import org.apache.flink.runtime.taskmanager.TaskManager import org.apache.flink.runtime.util.ZooKeeperUtil import org.apache.flink.runtime.util.{SerializedValue, EnvironmentInformation} import org.apache.flink.runtime.webmonitor.WebMonitor -import org.apache.flink.runtime.{StreamingMode, ActorSynchronousLogging, ActorLogMessages} +import org.apache.flink.runtime.{FlinkActor, StreamingMode, LeaderSessionMessages} +import org.apache.flink.runtime.{LogMessages} 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.instance.{ActorGateway, AkkaActorGateway, InstanceManager} import org.apache.flink.runtime.jobgraph.{JobVertexID, JobGraph, JobStatus} import org.apache.flink.runtime.jobmanager.scheduler.{Scheduler => FlinkScheduler} import org.apache.flink.runtime.messages.JobManagerMessages._ @@ -62,6 +63,7 @@ import org.apache.flink.runtime.messages.RegistrationMessages._ import org.apache.flink.runtime.messages.TaskManagerMessages.{SendStackTrace, Heartbeat} import org.apache.flink.util.{ExceptionUtils, InstantiationUtil} +import _root_.akka.pattern.ask import scala.concurrent._ import scala.concurrent.duration._ import scala.concurrent.forkjoin.ForkJoinPool @@ -107,11 +109,17 @@ class JobManager( protected val delayBetweenRetries: Long, protected val timeout: FiniteDuration, protected val mode: StreamingMode) - extends Actor with ActorLogMessages with ActorSynchronousLogging { + extends FlinkActor + with LeaderSessionMessages // order of the mixin is important, we want filtering after logging + with LogMessages // order of the mixin is important, we want first logging + { + + override val log = Logger(getClass) /** List of current jobs running jobs */ protected val currentJobs = scala.collection.mutable.HashMap[JobID, (ExecutionGraph, JobInfo)]() + override val leaderSessionID = Some(UUID.randomUUID()) /** * Run when the job manager is started. Simply logs an informational message. @@ -125,10 +133,10 @@ class JobManager( // disconnect the registered task managers instanceManager.getAllRegisteredInstances.asScala.foreach { - _.getInstanceGateway().tell(Disconnect("JobManager is shutting down")) + _.getActorGateway().tell(Disconnect("JobManager is shutting down")) } - archive ! PoisonPill + archive ! decorateMessage(PoisonPill) for((e,_) <- currentJobs.values) { e.fail(new Exception("The JobManager is shutting down.")) @@ -151,26 +159,48 @@ class JobManager( * * @return */ - override def receiveWithLogMessages: Receive = { + override def handleMessage: Receive = { - case RegisterTaskManager(taskManager, connectionInfo, hardwareInformation, numberOfSlots) => + case RegisterTaskManager( + registrationSessionID, + taskManager, + connectionInfo, + hardwareInformation, + numberOfSlots) => if (instanceManager.isRegistered(taskManager)) { val instanceID = instanceManager.getRegisteredInstance(taskManager).getId // IMPORTANT: Send the response to the "sender", which is not the // TaskManager actor, but the ask future! - sender() ! AlreadyRegistered(self, instanceID, libraryCacheManager.getBlobServerPort) + sender() ! decorateMessage( + AlreadyRegistered( + registrationSessionID, + leaderSessionID.get, + self, + instanceID, + libraryCacheManager.getBlobServerPort) + ) } else { try { - val instanceID = instanceManager.registerTaskManager(taskManager, connectionInfo, - hardwareInformation, numberOfSlots) + val instanceID = instanceManager.registerTaskManager( + taskManager, + connectionInfo, + hardwareInformation, + numberOfSlots, + leaderSessionID) // IMPORTANT: Send the response to the "sender", which is not the // TaskManager actor, but the ask future! - sender() ! AcknowledgeRegistration(self, instanceID, - libraryCacheManager.getBlobServerPort) + sender() ! decorateMessage( + AcknowledgeRegistration( + registrationSessionID, + leaderSessionID.get, + self, + instanceID, + libraryCacheManager.getBlobServerPort) + ) // to be notified when the taskManager is no longer reachable context.watch(taskManager) @@ -183,15 +213,19 @@ class JobManager( // IMPORTANT: Send the response to the "sender", which is not the // TaskManager actor, but the ask future! - sender() ! RefuseRegistration(ExceptionUtils.stringifyException(e)) + sender() ! decorateMessage( + RefuseRegistration( + registrationSessionID, + ExceptionUtils.stringifyException(e)) + ) } } case RequestNumberRegisteredTaskManager => - sender ! instanceManager.getNumberOfRegisteredTaskManagers + sender ! decorateMessage(instanceManager.getNumberOfRegisteredTaskManagers) case RequestTotalNumberOfSlots => - sender ! instanceManager.getTotalNumberOfSlots + sender ! decorateMessage(instanceManager.getTotalNumberOfSlots) case SubmitJob(jobGraph, listen) => submitJob(jobGraph, listenToEvents = listen) @@ -206,16 +240,19 @@ class JobManager( executionGraph.cancel() }(context.dispatcher) - sender ! CancellationSuccess(jobID) + sender ! decorateMessage(CancellationSuccess(jobID)) case None => log.info(s"No job found with ID $jobID.") - sender ! CancellationFailure(jobID, new IllegalArgumentException("No job found with " + - s"ID $jobID.")) + sender ! decorateMessage( + CancellationFailure( + jobID, + new IllegalArgumentException(s"No job found with ID $jobID.")) + ) } case UpdateTaskExecutionState(taskExecutionState) => if (taskExecutionState == null) { - sender ! false + sender ! decorateMessage(false) } else { currentJobs.get(taskExecutionState.getJobID) match { case Some((executionGraph, _)) => @@ -223,13 +260,13 @@ class JobManager( Future { val result = executionGraph.updateState(taskExecutionState) - originalSender ! result + originalSender ! decorateMessage(result) }(context.dispatcher) case None => log.error("Cannot find execution graph for ID " + s"${taskExecutionState.getJobID} to change state to " + s"${taskExecutionState.getExecutionState}.") - sender ! false + sender ! decorateMessage(false) } } @@ -283,7 +320,7 @@ class JobManager( null } - sender ! NextInputSplit(serializedInputSplit) + sender ! decorateMessage(NextInputSplit(serializedInputSplit)) case checkpointMessage : AbstractCheckpointMessage => handleCheckpointMessage(checkpointMessage) @@ -310,20 +347,31 @@ class JobManager( } val result = new SerializedJobExecutionResult(jobID, jobInfo.duration, accumulatorResults) - jobInfo.client ! JobResultSuccess(result) + jobInfo.client ! decorateMessage(JobResultSuccess(result)) case JobStatus.CANCELED => - jobInfo.client ! Failure(new JobCancellationException(jobID, - "Job was cancelled.", error)) + jobInfo.client ! decorateMessage( + Failure( + new JobCancellationException( + jobID, + "Job was cancelled.", error) + ) + ) case JobStatus.FAILED => - jobInfo.client ! Failure(new JobExecutionException(jobID, - "Job execution failed.", error)) + jobInfo.client ! decorateMessage( + 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) + jobInfo.client ! decorateMessage(Failure(exception)) throw exception } @@ -337,13 +385,17 @@ class JobManager( case ScheduleOrUpdateConsumers(jobId, partitionId) => currentJobs.get(jobId) match { case Some((executionGraph, _)) => - sender ! Acknowledge + sender ! decorateMessage(Acknowledge) executionGraph.scheduleOrUpdateConsumers(partitionId) case None => log.error(s"Cannot find execution graph for job ID $jobId to schedule or update " + s"consumers.") - sender ! Failure(new IllegalStateException("Cannot find execution graph for job ID " + - s"$jobId to schedule or update consumers.")) + sender ! decorateMessage( + Failure( + new IllegalStateException("Cannot find execution graph for job ID " + + s"$jobId to schedule or update consumers.") + ) + ) } case RequestPartitionState(jobId, partitionId, taskExecutionId, taskResultId) => @@ -360,15 +412,21 @@ class JobManager( null } - sender ! PartitionState( - taskExecutionId, taskResultId, partitionId.getPartitionId, state) + sender ! decorateMessage( + PartitionState( + taskExecutionId, + taskResultId, + partitionId.getPartitionId, + state) + ) case RequestJobStatus(jobID) => currentJobs.get(jobID) match { - case Some((executionGraph,_)) => sender ! CurrentJobStatus(jobID, executionGraph.getState) + case Some((executionGraph,_)) => + sender ! decorateMessage(CurrentJobStatus(jobID, executionGraph.getState)) case None => // check the archive - archive forward RequestJobStatus(jobID) + archive forward decorateMessage(RequestJobStatus(jobID)) } case RequestRunningJobs => @@ -376,16 +434,21 @@ class JobManager( case (_, (eg, jobInfo)) => eg } - sender ! RunningJobs(executionGraphs) + sender ! decorateMessage(RunningJobs(executionGraphs)) case RequestRunningJobsStatus => try { val jobs = currentJobs map { - case (_, (eg, _)) => new JobStatusMessage(eg.getJobID, eg.getJobName, - eg.getState, eg.getStatusTimestamp(JobStatus.CREATED)) + case (_, (eg, _)) => + new JobStatusMessage( + eg.getJobID, + eg.getJobName, + eg.getState, + eg.getStatusTimestamp(JobStatus.CREATED) + ) } - sender ! RunningJobsStatus(jobs) + sender ! decorateMessage(RunningJobsStatus(jobs)) } catch { case t: Throwable => log.error("Exception while responding to RequestRunningJobsStatus", t) @@ -393,18 +456,22 @@ class JobManager( case RequestJob(jobID) => currentJobs.get(jobID) match { - case Some((eg, _)) => sender ! JobFound(jobID, eg) + case Some((eg, _)) => sender ! decorateMessage(JobFound(jobID, eg)) case None => // check the archive - archive forward RequestJob(jobID) + archive forward decorateMessage(RequestJob(jobID)) } case RequestBlobManagerPort => - sender ! libraryCacheManager.getBlobServerPort + sender ! decorateMessage(libraryCacheManager.getBlobServerPort) case RequestRegisteredTaskManagers => import scala.collection.JavaConverters._ - sender ! RegisteredTaskManagers(instanceManager.getAllRegisteredInstances.asScala) + sender ! decorateMessage( + RegisteredTaskManagers( + instanceManager.getAllRegisteredInstances.asScala + ) + ) case Heartbeat(instanceID, metricsReport, accumulators) => log.debug(s"Received hearbeat message from $instanceID.") @@ -420,8 +487,8 @@ class JobManager( case message: InfoMessage => handleInfoRequestMessage(message, sender()) case RequestStackTrace(instanceID) => - val gateway = instanceManager.getRegisteredInstanceById(instanceID).getInstanceGateway - gateway.forward(SendStackTrace, sender()) + val gateway = instanceManager.getRegisteredInstanceById(instanceID).getActorGateway + gateway.forward(SendStackTrace, new AkkaActorGateway(sender(), leaderSessionID)) case Terminated(taskManager) => if (instanceManager.isRegistered(taskManager)) { @@ -432,7 +499,7 @@ class JobManager( } case RequestJobManagerStatus => - sender() ! JobManagerStatusAlive + sender() ! decorateMessage(JobManagerStatusAlive) case Disconnect(msg) => val taskManager = sender() @@ -443,6 +510,9 @@ class JobManager( instanceManager.unregisterTaskManager(taskManager) context.unwatch(taskManager) } + + case RequestLeaderSessionID => + sender() ! ResponseLeaderSessionID(leaderSessionID) } /** @@ -456,7 +526,11 @@ class JobManager( */ private def submitJob(jobGraph: JobGraph, listenToEvents: Boolean): Unit = { if (jobGraph == null) { - sender ! Failure(new JobSubmissionException(null, "JobGraph must not be null.")) + sender ! decorateMessage( + Failure( + new JobSubmissionException(null, "JobGraph must not be null.") + ) + ) } else { val jobId = jobGraph.getJobID @@ -499,7 +573,8 @@ class JobManager( timeout, jobGraph.getUserJarBlobKeys, userCodeLoader), - JobInfo(sender(), System.currentTimeMillis())) + JobInfo(sender(), System.currentTimeMillis()) + ) )._1 // configure the execution graph @@ -577,23 +652,29 @@ class JobManager( val confirmVertices: java.util.List[ExecutionJobVertex] = snapshotSettings.getVerticesToConfirm.asScala.map(idToVertex).asJava - executionGraph.enableSnaphotCheckpointing( - snapshotSettings.getCheckpointInterval, snapshotSettings.getCheckpointTimeout, - triggerVertices, ackVertices, confirmVertices, - context.system) + executionGraph.enableSnapshotCheckpointing( + snapshotSettings.getCheckpointInterval, + snapshotSettings.getCheckpointTimeout, + triggerVertices, + ackVertices, + confirmVertices, + context.system, + leaderSessionID) } // get notified about job status changes - executionGraph.registerJobStatusListener(self) + executionGraph.registerJobStatusListener(new AkkaActorGateway(self, leaderSessionID)) if (listenToEvents) { // the sender wants to be notified about state changes - executionGraph.registerExecutionListener(sender()) - executionGraph.registerJobStatusListener(sender()) + val gateway = new AkkaActorGateway(sender(), leaderSessionID) + + executionGraph.registerExecutionListener(gateway) + executionGraph.registerJobStatusListener(gateway) } // done with submitting the job - sender ! Success(jobGraph.getJobID) + sender ! decorateMessage(Success(jobGraph.getJobID)) } catch { case t: Throwable => @@ -612,7 +693,7 @@ class JobManager( new JobExecutionException(jobId, s"Failed to submit job ${jobId} (${jobName})", t) } - sender ! Failure(rt) + sender ! decorateMessage(Failure(rt)) return } @@ -690,14 +771,14 @@ class JobManager( currentJobs.get(jobID) match { case Some((graph, jobInfo)) => val accumulatorValues = graph.getAccumulatorsSerialized() - sender() ! AccumulatorResultsFound(jobID, accumulatorValues) + sender() ! decorateMessage(AccumulatorResultsFound(jobID, accumulatorValues)) case None => archive.forward(message) } } catch { case e: Exception => log.error("Cannot serialize accumulator result.", e) - sender() ! AccumulatorResultsErroneous(jobID, e) + sender() ! decorateMessage(AccumulatorResultsErroneous(jobID, e)) } case RequestAccumulatorResultsStringified(jobId) => @@ -830,7 +911,7 @@ class JobManager( try { eg.prepareForArchiving() - archive ! ArchiveExecutionGraph(jobID, eg) + archive ! decorateMessage(ArchiveExecutionGraph(jobID, eg)) } catch { case t: Throwable => log.error(s"Could not prepare the execution graph $eg for " + "archiving.", t) @@ -967,11 +1048,13 @@ object JobManager { * @param listeningAddress The hostname where the JobManager should listen for messages. * @param listeningPort The port where the JobManager should listen for messages. */ - def runJobManager(configuration: Configuration, - executionMode: JobManagerMode, - streamingMode: StreamingMode, - listeningAddress: String, - listeningPort: Int) : Unit = { + def runJobManager( + configuration: Configuration, + executionMode: JobManagerMode, + streamingMode: StreamingMode, + listeningAddress: String, + listeningPort: Int) + : Unit = { LOG.info("Starting JobManager") @@ -979,8 +1062,10 @@ object JobManager { LOG.info(s"Starting JobManager actor system at $listeningAddress:$listeningPort.") val jobManagerSystem = try { - val akkaConfig = AkkaUtils.getAkkaConfig(configuration, - Some((listeningAddress, listeningPort))) + val akkaConfig = AkkaUtils.getAkkaConfig( + configuration, + Some((listeningAddress, listeningPort)) + ) if (LOG.isDebugEnabled) { LOG.debug("Using akka configuration\n " + akkaConfig) } @@ -1003,14 +1088,20 @@ object JobManager { try { // bring up the job manager actor LOG.info("Starting JobManager actor") - val (jobManager, archiver) = startJobManagerActors(configuration, - jobManagerSystem, streamingMode) + val (jobManager, archiver) = startJobManagerActors( + configuration, + jobManagerSystem, + streamingMode) // start a process reaper that watches the JobManager. If the JobManager actor dies, // the process reaper will kill the JVM process (to ensure easy failure detection) LOG.debug("Starting JobManager process reaper") jobManagerSystem.actorOf( - Props(classOf[ProcessReaper], jobManager, LOG.logger, RUNTIME_FAILURE_RETURN_CODE), + Props( + classOf[ProcessReaper], + jobManager, + LOG.logger, + RUNTIME_FAILURE_RETURN_CODE), "JobManager_Process_Reaper") // bring up a local task manager, if needed @@ -1018,16 +1109,22 @@ object JobManager { LOG.info("Starting embedded TaskManager for JobManager's LOCAL execution mode") val taskManagerActor = TaskManager.startTaskManagerComponentsAndActor( - configuration, jobManagerSystem, - listeningAddress, - Some(TaskManager.TASK_MANAGER_NAME), - Some(jobManager.path.toString), - true, streamingMode, - classOf[TaskManager]) + configuration, + jobManagerSystem, + listeningAddress, + Some(TaskManager.TASK_MANAGER_NAME), + Some(jobManager.path.toString), + true, + streamingMode, + classOf[TaskManager]) LOG.debug("Starting TaskManager process reaper") jobManagerSystem.actorOf( - Props(classOf[ProcessReaper], taskManagerActor, LOG.logger, RUNTIME_FAILURE_RETURN_CODE), + Props( + classOf[ProcessReaper], + taskManagerActor, + LOG.logger, + RUNTIME_FAILURE_RETURN_CODE), "TaskManager_Process_Reaper") } @@ -1257,13 +1354,18 @@ object JobManager { * @param actorSystem Teh actor system running the JobManager * @return A tuple of references (JobManager Ref, Archiver Ref) */ - def startJobManagerActors(configuration: Configuration, - actorSystem: ActorSystem, - streamingMode: StreamingMode): (ActorRef, ActorRef) = { + def startJobManagerActors( + configuration: Configuration, + actorSystem: ActorSystem, + streamingMode: StreamingMode) + : (ActorRef, ActorRef) = { - startJobManagerActors(configuration, actorSystem, - Some(JOB_MANAGER_NAME), Some(ARCHIVE_NAME), - streamingMode) + startJobManagerActors( + configuration, + actorSystem, + Some(JOB_MANAGER_NAME), + Some(ARCHIVE_NAME), + streamingMode) } /** * Starts the JobManager and job archiver based on the given configuration, in the @@ -1279,11 +1381,13 @@ object JobManager { * * @return A tuple of references (JobManager Ref, Archiver Ref) */ - def startJobManagerActors(configuration: Configuration, - actorSystem: ActorSystem, - jobMangerActorName: Option[String], - archiverActorName: Option[String], - streamingMode: StreamingMode): (ActorRef, ActorRef) = { + def startJobManagerActors( + configuration: Configuration, + actorSystem: ActorSystem, + jobMangerActorName: Option[String], + archiverActorName: Option[String], + streamingMode: StreamingMode) + : (ActorRef, ActorRef) = { val (executionContext, instanceManager, @@ -1352,9 +1456,11 @@ object JobManager { "akka://flink/user/" + JOB_MANAGER_NAME } - def getJobManagerRemoteReferenceFuture(address: InetSocketAddress, - system: ActorSystem, - timeout: FiniteDuration): Future[ActorRef] = { + def getJobManagerRemoteReferenceFuture( + address: InetSocketAddress, + system: ActorSystem, + timeout: FiniteDuration) + : Future[ActorRef] = { AkkaUtils.getReference(getRemoteJobManagerAkkaURL(address), system, timeout) } @@ -1369,9 +1475,12 @@ object JobManager { * @return The ActorRef to the JobManager */ @throws(classOf[IOException]) - def getJobManagerRemoteReference(jobManagerUrl: String, - system: ActorSystem, - timeout: FiniteDuration): ActorRef = { + def getJobManagerRemoteReference( + jobManagerUrl: String, + system: ActorSystem, + timeout: FiniteDuration) + : ActorRef = { + try { val future = AkkaUtils.getReference(jobManagerUrl, system, timeout) Await.result(future, timeout) @@ -1397,9 +1506,11 @@ object JobManager { * @return The ActorRef to the JobManager */ @throws(classOf[IOException]) - def getJobManagerRemoteReference(address: InetSocketAddress, - system: ActorSystem, - timeout: FiniteDuration): ActorRef = { + def getJobManagerRemoteReference( + address: InetSocketAddress, + system: ActorSystem, + timeout: FiniteDuration) + : ActorRef = { val jmAddress = getRemoteJobManagerAkkaURL(address) getJobManagerRemoteReference(jmAddress, system, timeout) @@ -1415,14 +1526,38 @@ object JobManager { * @return The ActorRef to the JobManager */ @throws(classOf[IOException]) - def getJobManagerRemoteReference(address: InetSocketAddress, - system: ActorSystem, - config: Configuration): ActorRef = { + def getJobManagerRemoteReference( + address: InetSocketAddress, + system: ActorSystem, + config: Configuration) + : ActorRef = { val timeout = AkkaUtils.getLookupTimeout(config) getJobManagerRemoteReference(address, system, timeout) } + /** Returns the [[ActorGateway]] for the provided JobManager. The function automatically + * retrieves the current leader session ID from the JobManager and instantiates the + * [[AkkaActorGateway]] with it. + * + * @param jobManager ActorRef to the [[JobManager]] + * @param timeout Timeout for the blocking leader session ID retrieval + * @throws java.lang.Exception + * @return Gateway to the specified JobManager + */ + @throws(classOf[Exception]) + def getJobManagerGateway( + jobManager: ActorRef, + timeout: FiniteDuration + ): ActorGateway = { + val futureLeaderSessionID = (jobManager ? RequestLeaderSessionID)(timeout) + .mapTo[ResponseLeaderSessionID] + + val leaderSessionID = Await.result(futureLeaderSessionID, timeout).leaderSessionID + + new AkkaActorGateway(jobManager, leaderSessionID) + } + // -------------------------------------------------------------------------- // Utilities // -------------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/MemoryArchivist.scala ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/MemoryArchivist.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/MemoryArchivist.scala index 7572e72..9f228ed 100644 --- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/MemoryArchivist.scala +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/MemoryArchivist.scala @@ -20,13 +20,12 @@ package org.apache.flink.runtime.jobmanager import java.util -import akka.actor.Actor - +import grizzled.slf4j.Logger import org.apache.flink.api.common.JobID import org.apache.flink.runtime.jobgraph.JobStatus import org.apache.flink.runtime.messages.accumulators._ +import org.apache.flink.runtime.{FlinkActor, LogMessages} import org.apache.flink.runtime.messages.webmonitor._ -import org.apache.flink.runtime.{ActorSynchronousLogging, ActorLogMessages} import org.apache.flink.runtime.executiongraph.ExecutionGraph import org.apache.flink.runtime.messages.ArchiveMessages._ import org.apache.flink.runtime.messages.JobManagerMessages._ @@ -56,9 +55,11 @@ import scala.collection.mutable * @param max_entries Maximum number of stored Flink jobs */ class MemoryArchivist(private val max_entries: Int) - extends Actor - with ActorLogMessages - with ActorSynchronousLogging { + extends FlinkActor + with LogMessages { + + override val log = Logger(getClass) + /* * Map of execution graphs belonging to recently started jobs with the time stamp of the last * received job event. The insert order is preserved through a LinkedHashMap. @@ -70,41 +71,45 @@ class MemoryArchivist(private val max_entries: Int) var canceledCnt: Int = 0 var failedCnt: Int = 0 - override def receiveWithLogMessages: Receive = { + override def handleMessage: Receive = { /* Receive Execution Graph to archive */ case ArchiveExecutionGraph(jobID, graph) => // wrap graph inside a soft reference graphs.update(jobID, graph) + // update job counters graph.getState match { case JobStatus.FINISHED => finishedCnt += 1 case JobStatus.CANCELED => canceledCnt += 1 case JobStatus.FAILED => failedCnt += 1 + // ignore transitional states, e.g. Cancelling, Running, Failing, etc. + case _ => } + trimHistory() case RequestArchivedJob(jobID: JobID) => val graph = graphs.get(jobID) - sender ! ArchivedJob(graph) + sender ! decorateMessage(ArchivedJob(graph)) case RequestArchivedJobs => - sender ! ArchivedJobs(graphs.values) + sender ! decorateMessage(ArchivedJobs(graphs.values)) case RequestJob(jobID) => graphs.get(jobID) match { - case Some(graph) => sender ! JobFound(jobID, graph) - case None => sender ! JobNotFound(jobID) + case Some(graph) => sender ! decorateMessage(JobFound(jobID, graph)) + case None => sender ! decorateMessage(JobNotFound(jobID)) } case RequestJobStatus(jobID) => graphs.get(jobID) match { - case Some(graph) => sender ! CurrentJobStatus(jobID, graph.getState) - case None => sender ! JobNotFound(jobID) + case Some(graph) => sender ! decorateMessage(CurrentJobStatus(jobID, graph.getState)) + case None => sender ! decorateMessage(JobNotFound(jobID)) } case RequestJobCounts => - sender ! (finishedCnt, canceledCnt, failedCnt) + sender ! decorateMessage((finishedCnt, canceledCnt, failedCnt)) case _ : RequestJobsOverview => try { http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/ExecutionGraphMessages.scala ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/ExecutionGraphMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/ExecutionGraphMessages.scala index 6bc59a2..0cb3b0d 100644 --- a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/ExecutionGraphMessages.scala +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/ExecutionGraphMessages.scala @@ -45,11 +45,18 @@ object ExecutionGraphMessages { * @param timestamp of the execution state change * @param optionalMessage */ - case class ExecutionStateChanged(jobID: JobID, vertexID: JobVertexID, - taskName: String, totalNumberOfSubTasks: Int, subtaskIndex: Int, - executionID: ExecutionAttemptID, - newExecutionState: ExecutionState, timestamp: Long, - optionalMessage: String){ + case class ExecutionStateChanged( + jobID: JobID, + vertexID: JobVertexID, + taskName: String, + totalNumberOfSubTasks: Int, + subtaskIndex: Int, + executionID: ExecutionAttemptID, + newExecutionState: ExecutionState, + timestamp: Long, + optionalMessage: String) + extends RequiresLeaderSessionID { + override def toString: String = { val oMsg = if (optionalMessage != null) { s"\n$optionalMessage" @@ -69,8 +76,12 @@ object ExecutionGraphMessages { * @param timestamp * @param error */ - case class JobStatusChanged(jobID: JobID, newJobStatus: JobStatus, timestamp: Long, - error: Throwable){ + case class JobStatusChanged( + jobID: JobID, + newJobStatus: JobStatus, + timestamp: Long, + error: Throwable) + extends RequiresLeaderSessionID { override def toString: String = { s"${timestampToString(timestamp)}\tJob execution switched to status $newJobStatus." } http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/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 c9a2878..e38986b 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 @@ -18,6 +18,8 @@ package org.apache.flink.runtime.messages +import java.util.UUID + import org.apache.flink.api.common.JobID import org.apache.flink.runtime.client.{SerializedJobExecutionResult, JobStatusMessage} import org.apache.flink.runtime.executiongraph.{ExecutionAttemptID, ExecutionGraph} @@ -32,6 +34,8 @@ import scala.collection.JavaConverters._ */ object JobManagerMessages { + case class LeaderSessionMessage(leaderSessionID: Option[UUID], message: Any) + /** * Submits a job to the job manager. If [[registerForEvents]] is true, * then the sender will be registered as listener for the state change messages. @@ -41,6 +45,7 @@ object JobManagerMessages { * @param registerForEvents if true, then register for state change events */ case class SubmitJob(jobGraph: JobGraph, registerForEvents: Boolean) + extends RequiresLeaderSessionID /** * Cancels a job with the given [[jobID]] at the JobManager. The result of the cancellation is @@ -48,19 +53,22 @@ object JobManagerMessages { * * @param jobID */ - case class CancelJob(jobID: JobID) + case class CancelJob(jobID: JobID) extends RequiresLeaderSessionID /** * 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. + * [[NextInputSplit]] message. * * @param jobID * @param vertexID */ - case class RequestNextInputSplit(jobID: JobID, vertexID: JobVertexID, executionAttempt: - ExecutionAttemptID) + case class RequestNextInputSplit( + jobID: JobID, + vertexID: JobVertexID, + executionAttempt: ExecutionAttemptID) + extends RequiresLeaderSessionID /** * Contains the next input split for a task. This message is a response to @@ -80,10 +88,12 @@ object JobManagerMessages { * @param taskExecutionId The execution attempt ID of the task requesting the partition state. * @param taskResultId The input gate ID of the task requesting the partition state. */ - case class RequestPartitionState(jobId: JobID, - partitionId: ResultPartitionID, - taskExecutionId: ExecutionAttemptID, - taskResultId: IntermediateDataSetID) + case class RequestPartitionState( + jobId: JobID, + partitionId: ResultPartitionID, + taskExecutionId: ExecutionAttemptID, + taskResultId: IntermediateDataSetID) + extends RequiresLeaderSessionID /** * Notifies the [[org.apache.flink.runtime.jobmanager.JobManager]] about available data for a @@ -101,8 +111,7 @@ object JobManagerMessages { * @see [[org.apache.flink.runtime.io.network.partition.ResultPartition]] */ case class ScheduleOrUpdateConsumers(jobId: JobID, partitionId: ResultPartitionID) - - case class ConsumerNotificationResult(success: Boolean, error: Option[Throwable] = None) + extends RequiresLeaderSessionID /** * Requests the current [[JobStatus]] of the job identified by [[jobID]]. This message triggers @@ -142,6 +151,17 @@ object JobManagerMessages { */ case object RequestBlobManagerPort + /** Requests the current leader session ID of the job manager. The result is sent back to the + * sender as an [[ResponseLeaderSessionID]] + */ + case object RequestLeaderSessionID + + /** Response to the [[RequestLeaderSessionID]] message. + * + * @param leaderSessionID + */ + case class ResponseLeaderSessionID(leaderSessionID: Option[UUID]) + /** * Denotes a successful job execution. */ @@ -300,4 +320,8 @@ object JobManagerMessages { def getJobManagerStatusAlive : AnyRef = { JobManagerStatusAlive } + + def getRequestLeaderSessionID: AnyRef = { + RequestLeaderSessionID + } } http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/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 e292a47..75036b3 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 @@ -39,7 +39,7 @@ object Messages { * * @param reason The reason for disconnecting, to be displayed in log and error messages. */ - case class Disconnect(reason: String) + case class Disconnect(reason: String) extends RequiresLeaderSessionID /** * Accessor for the case object instance, to simplify Java interoperability. http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/RegistrationMessages.scala ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/RegistrationMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/RegistrationMessages.scala index 3051d00..b435ebc 100644 --- a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/RegistrationMessages.scala +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/RegistrationMessages.scala @@ -18,6 +18,8 @@ package org.apache.flink.runtime.messages +import java.util.UUID + import akka.actor.ActorRef import org.apache.flink.runtime.instance.{InstanceConnectionInfo, InstanceID, HardwareDescription} @@ -32,7 +34,9 @@ object RegistrationMessages { /** * Marker trait for registration messages. */ - trait RegistrationMessage + trait RegistrationMessage { + def registrationSessionID: UUID + } /** * Triggers the TaskManager to attempt a registration at the JobManager. @@ -42,10 +46,12 @@ object RegistrationMessages { * @param deadline Optional deadline until when the registration must be completed. * @param attempt The attempt number, for logging. */ - case class TriggerTaskManagerRegistration(jobManagerAkkaURL: String, - timeout: FiniteDuration, - deadline: Option[Deadline], - attempt: Int) + case class TriggerTaskManagerRegistration( + registrationSessionID: UUID, + jobManagerAkkaURL: String, + timeout: FiniteDuration, + deadline: Option[Deadline], + attempt: Int) extends RegistrationMessage /** @@ -57,10 +63,12 @@ object RegistrationMessages { * @param resources The TaskManagers resources. * @param numberOfSlots The number of processing slots offered by the TaskManager. */ - case class RegisterTaskManager(taskManager: ActorRef, - connectionInfo: InstanceConnectionInfo, - resources: HardwareDescription, - numberOfSlots: Int) + case class RegisterTaskManager( + registrationSessionID: UUID, + taskManager: ActorRef, + connectionInfo: InstanceConnectionInfo, + resources: HardwareDescription, + numberOfSlots: Int) extends RegistrationMessage /** @@ -71,7 +79,12 @@ object RegistrationMessages { * JobManager. * @param blobPort The server port where the JobManager's BLOB service runs. */ - case class AcknowledgeRegistration(jobManager: ActorRef, instanceID: InstanceID, blobPort: Int) + case class AcknowledgeRegistration( + registrationSessionID: UUID, + leaderSessionID: UUID, + jobManager: ActorRef, + instanceID: InstanceID, + blobPort: Int) extends RegistrationMessage /** @@ -80,7 +93,12 @@ object RegistrationMessages { * @param instanceID The instance ID under which the TaskManager is registered. * @param blobPort The server port where the JobManager's BLOB service runs. */ - case class AlreadyRegistered(jobManager: ActorRef, instanceID: InstanceID, blobPort: Int) + case class AlreadyRegistered( + registrationSessionID: UUID, + leaderSessionID: UUID, + jobManager: ActorRef, + instanceID: InstanceID, + blobPort: Int) extends RegistrationMessage /** @@ -89,6 +107,6 @@ object RegistrationMessages { * * @param reason Reason why the task manager registration was refused */ - case class RefuseRegistration(reason: String) + case class RefuseRegistration(registrationSessionID: UUID, reason: String) extends RegistrationMessage } http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskMessages.scala ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskMessages.scala index 9373576..a80ca99 100644 --- a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskMessages.scala +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskMessages.scala @@ -47,7 +47,7 @@ object TaskMessages { * @param tasks Descriptor which contains the information to start the task. */ case class SubmitTask(tasks: TaskDeploymentDescriptor) - extends TaskMessage + extends TaskMessage with RequiresLeaderSessionID /** * Cancels the task associated with [[attemptID]]. The result is sent back to the sender as a @@ -56,7 +56,7 @@ object TaskMessages { * @param attemptID The task's execution attempt ID. */ case class CancelTask(attemptID: ExecutionAttemptID) - extends TaskMessage + extends TaskMessage with RequiresLeaderSessionID /** * Triggers a fail of specified task from the outside (as opposed to the task throwing @@ -86,15 +86,16 @@ object TaskMessages { * Answer to a [[RequestPartitionState]] with the state of the respective partition. */ case class PartitionState( - taskExecutionId: ExecutionAttemptID, - taskResultId: IntermediateDataSetID, - partitionId: IntermediateResultPartitionID, - state: ExecutionState) extends TaskMessage + taskExecutionId: ExecutionAttemptID, + taskResultId: IntermediateDataSetID, + partitionId: IntermediateResultPartitionID, + state: ExecutionState) + extends TaskMessage with RequiresLeaderSessionID /** * Base class for messages that update the information about location of input partitions */ - abstract sealed class UpdatePartitionInfo extends TaskMessage { + abstract sealed class UpdatePartitionInfo extends TaskMessage with RequiresLeaderSessionID { def executionID: ExecutionAttemptID } @@ -104,9 +105,10 @@ object TaskMessages { * @param resultId The input reader to update. * @param partitionInfo The partition info update. */ - case class UpdateTaskSinglePartitionInfo(executionID: ExecutionAttemptID, - resultId: IntermediateDataSetID, - partitionInfo: InputChannelDeploymentDescriptor) + case class UpdateTaskSinglePartitionInfo( + executionID: ExecutionAttemptID, + resultId: IntermediateDataSetID, + partitionInfo: InputChannelDeploymentDescriptor) extends UpdatePartitionInfo /** @@ -115,8 +117,8 @@ object TaskMessages { * @param partitionInfos List of input gates with channel descriptors to update. */ case class UpdateTaskMultiplePartitionInfos( - executionID: ExecutionAttemptID, - partitionInfos: Seq[(IntermediateDataSetID, InputChannelDeploymentDescriptor)]) + executionID: ExecutionAttemptID, + partitionInfos: Seq[(IntermediateDataSetID, InputChannelDeploymentDescriptor)]) extends UpdatePartitionInfo /** @@ -126,7 +128,7 @@ object TaskMessages { * @param executionID The task's execution attempt ID. */ case class FailIntermediateResultPartitions(executionID: ExecutionAttemptID) - extends TaskMessage + extends TaskMessage with RequiresLeaderSessionID // -------------------------------------------------------------------------- @@ -140,7 +142,7 @@ object TaskMessages { * @param taskExecutionState The changed task state */ case class UpdateTaskExecutionState(taskExecutionState: TaskExecutionState) - extends TaskMessage + extends TaskMessage with RequiresLeaderSessionID /** * Response message to updates in the task state. Send for example as a response to @@ -152,11 +154,11 @@ object TaskMessages { * @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 - { + case class TaskOperationResult( + executionID: ExecutionAttemptID, + success: Boolean, + description: String) + extends TaskMessage { def this(executionID: ExecutionAttemptID, success: Boolean) = this(executionID, success, "") } @@ -166,10 +168,10 @@ object TaskMessages { // -------------------------------------------------------------------------- def createUpdateTaskMultiplePartitionInfos( - executionID: ExecutionAttemptID, - resultIDs: java.util.List[IntermediateDataSetID], - partitionInfos: java.util.List[InputChannelDeploymentDescriptor]): - UpdateTaskMultiplePartitionInfos = { + executionID: ExecutionAttemptID, + resultIDs: java.util.List[IntermediateDataSetID], + partitionInfos: java.util.List[InputChannelDeploymentDescriptor]) + : UpdateTaskMultiplePartitionInfos = { require(resultIDs.size() == partitionInfos.size(), "ResultIDs must have the same length as partitionInfos.") http://git-wip-us.apache.org/repos/asf/flink/blob/22224542/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 49c701e..6f810fc 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 @@ -28,8 +28,11 @@ import org.apache.flink.api.common.JobSubmissionResult import org.apache.flink.configuration.{ConfigConstants, Configuration} import org.apache.flink.runtime.StreamingMode import org.apache.flink.runtime.akka.AkkaUtils -import org.apache.flink.runtime.client.{JobExecutionException, JobClient, SerializedJobExecutionResult} +import org.apache.flink.runtime.client.{JobExecutionException, JobClient, +SerializedJobExecutionResult} +import org.apache.flink.runtime.instance.ActorGateway import org.apache.flink.runtime.jobgraph.JobGraph +import org.apache.flink.runtime.jobmanager.JobManager import org.apache.flink.runtime.messages.TaskManagerMessages.NotifyWhenRegisteredAtJobManager import org.slf4j.LoggerFactory @@ -132,8 +135,9 @@ abstract class FlinkMiniCluster( AkkaUtils.createActorSystem(config) } - def getJobManager: ActorRef = { - jobManagerActor + def getJobManagerGateway(): ActorGateway = { + // create ActorGateway from the JobManager's ActorRef + JobManager.getJobManagerGateway(jobManagerActor, timeout) } def getTaskManagers = { @@ -206,12 +210,17 @@ abstract class FlinkMiniCluster( val clientActorSystem = if (singleActorSystem) jobManagerActorSystem else JobClient.startJobClientActorSystem(configuration) - JobClient.submitJobAndWait(clientActorSystem, jobManagerActor, jobGraph, timeout, printUpdates) + JobClient.submitJobAndWait( + clientActorSystem, + getJobManagerGateway(), + jobGraph, + timeout, + printUpdates) } @throws(classOf[JobExecutionException]) def submitJobDetached(jobGraph: JobGraph) : JobSubmissionResult = { - JobClient.submitJobDetached(jobManagerActor, jobGraph, timeout) + JobClient.submitJobDetached(getJobManagerGateway(), jobGraph, timeout) new JobSubmissionResult(jobGraph.getJobID) } }
