You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/04/29 11:50:30 UTC
[1/3] flink git commit: [FLINK-1925] [runtime] Splits the processing
of the SubmitTask message into two phases: 1. TDD reception with eager
acknowledgement and 2. TDD instantiation with a subsequent state update
message.
Repository: flink
Updated Branches:
refs/heads/master 9a18e5790 -> 4cb5b7254
[FLINK-1925] [runtime] Splits the processing of the SubmitTask message into two phases:
1. TDD reception with eager acknowledgement and
2. TDD instantiation with a subsequent state update message.
This closes #622
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4cb5b725
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4cb5b725
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4cb5b725
Branch: refs/heads/master
Commit: 4cb5b725447dad55dfbe66efbfc4580c51d8087d
Parents: 5a2ca81
Author: Till Rohrmann <tr...@apache.org>
Authored: Thu Apr 23 16:13:22 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Apr 29 10:40:39 2015 +0200
----------------------------------------------------------------------
.../flink/runtime/executiongraph/Execution.java | 24 +---
.../runtime/executiongraph/ExecutionGraph.java | 2 +
.../flink/runtime/messages/Messages.scala | 16 +--
.../flink/runtime/taskmanager/TaskManager.scala | 80 +++++++----
.../ExecutionGraphDeploymentTest.java | 2 +-
.../executiongraph/ExecutionGraphTestUtils.java | 9 +-
.../ExecutionVertexCancelTest.java | 12 +-
.../ExecutionVertexDeploymentTest.java | 37 ++---
.../ExecutionVertexSchedulingTest.java | 4 +-
.../runtime/taskmanager/TaskManagerTest.java | 136 ++++++++++++++-----
.../testingUtils/TestingTaskManager.scala | 42 +++++-
.../TestingTaskManagerMessages.scala | 2 +
12 files changed, 223 insertions(+), 143 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4cb5b725/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index baed947..93b4f2f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -43,6 +43,7 @@ import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
import org.apache.flink.runtime.jobmanager.scheduler.SlotAllocationFuture;
import org.apache.flink.runtime.jobmanager.scheduler.SlotAllocationFutureAction;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.runtime.messages.Messages;
import org.apache.flink.runtime.messages.TaskMessages.TaskOperationResult;
import org.apache.flink.runtime.state.StateHandle;
import org.apache.flink.util.ExceptionUtils;
@@ -347,26 +348,9 @@ public class Execution implements Serializable {
}
}
else {
- if (success == null) {
- markFailed(new Exception("Failed to deploy the task to slot " + slot + ": TaskOperationResult was null"));
- }
-
- if (success instanceof TaskOperationResult) {
- TaskOperationResult result = (TaskOperationResult) success;
-
- if (!result.executionID().equals(attemptId)) {
- markFailed(new Exception("Answer execution id does not match the request execution id."));
- } else if (result.success()) {
- switchToRunning();
- } else {
- // deployment failed :(
- markFailed(new Exception("Failed to deploy the task " +
- getVertexWithAttempt() + " to slot " + slot + ": " + result
- .description()));
- }
- } else {
+ if (!(success.equals(Messages.getAcknowledge()))) {
markFailed(new Exception("Failed to deploy the task to slot " + slot +
- ": Response was not of type TaskOperationResult"));
+ ": Response was not of type Acknowledge"));
}
}
}
@@ -759,7 +743,7 @@ public class Execution implements Serializable {
}
}
- private boolean switchToRunning() {
+ boolean switchToRunning() {
if (transitionState(DEPLOYING, RUNNING)) {
sendPartitionInfos();
http://git-wip-us.apache.org/repos/asf/flink/blob/4cb5b725/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
index 08e44f9..d38913e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
@@ -601,6 +601,8 @@ public class ExecutionGraph implements Serializable {
Execution attempt = this.currentExecutions.get(state.getID());
if (attempt != null) {
switch (state.getExecutionState()) {
+ case RUNNING:
+ return attempt.switchToRunning();
case FINISHED:
attempt.markFinished();
return true;
http://git-wip-us.apache.org/repos/asf/flink/blob/4cb5b725/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 ab8b8c2..e292a47 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,15 +26,7 @@ object Messages {
/**
* Message to signal the successful reception of another message
*/
- case object Acknowledge {
-
- /**
- * Accessor for the case object instance, to simplify Java interoperability.
- *
- * @return The Acknowledge case object instance.
- */
- def get(): Acknowledge.type = this
- }
+ case object Acknowledge
/**
* Signals that the receiver (JobManager/TaskManager) shall disconnect the sender.
@@ -49,4 +41,10 @@ object Messages {
*/
case class Disconnect(reason: String)
+ /**
+ * Accessor for the case object instance, to simplify Java interoperability.
+ *
+ * @return The Acknowledge case object instance.
+ */
+ def getAcknowledge(): Acknowledge.type = Acknowledge
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4cb5b725/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 511de6b..a6b9133 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
@@ -68,6 +68,8 @@ import org.apache.flink.runtime.security.SecurityUtils.FlinkSecuredRunner
import org.apache.flink.runtime.util.{MathUtils, EnvironmentInformation}
import org.apache.flink.util.ExceptionUtils
+import org.slf4j.LoggerFactory
+
import scala.concurrent._
import scala.concurrent.duration._
import scala.util.{Failure, Success}
@@ -133,7 +135,7 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
protected val resources = HardwareDescription.extractFromSystem(memoryManager.getMemorySize)
/** Registry of all tasks currently executed by this TaskManager */
- protected val runningTasks = scala.collection.mutable.HashMap[ExecutionAttemptID, Task]()
+ protected val runningTasks = new util.concurrent.ConcurrentHashMap[ExecutionAttemptID, Task]()
/** Handler for shared broadcast variables (shared between multiple Tasks) */
protected val bcVarManager = new BroadcastVariableManager()
@@ -380,7 +382,7 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
// marks a task as failed for an external reason
// external reasons are reasons other than the task code itself throwing an exception
case FailTask(executionID, cause) =>
- runningTasks.get(executionID) match {
+ Option(runningTasks.get(executionID)) match {
case Some(task) =>
// execute failing operation concurrently
@@ -395,7 +397,7 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
// cancels a task
case CancelTask(executionID) =>
- runningTasks.get(executionID) match {
+ Option(runningTasks.get(executionID)) match {
case Some(task) =>
// execute cancel operation concurrently
implicit val executor = context.dispatcher
@@ -422,11 +424,12 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
private def handleCheckpointingMessage(message: CheckpointingMessage): Unit = {
message match {
+
case BarrierReq(attemptID, checkpointID) =>
log.debug(s"[FT-TaskManager] Barrier $checkpointID request received " +
s"for attempt $attemptID.")
- runningTasks.get(attemptID) match {
+ Option(runningTasks.get(attemptID)) match {
case Some(i) =>
if (i.getExecutionState == ExecutionState.RUNNING) {
i.getEnvironment.getInvokable match {
@@ -773,31 +776,45 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
// --------------------------------------------------------------------------
/**
- * Receives a [[TaskDeploymentDescriptor]] describing the task to be executed. Sets up a
- * [[RuntimeEnvironment]] for the task and starts its execution in a separate thread.
+ * Receives a [[TaskDeploymentDescriptor]] describing the task to be executed. It eagerly
+ * acknowledges the task reception to the sender and asynchronously starts the initialization of
+ * the task.
*
* @param tdd TaskDeploymentDescriptor describing the task to be executed on this [[TaskManager]]
*/
private def submitTask(tdd: TaskDeploymentDescriptor): Unit = {
+ val slot = tdd.getTargetSlotNumber
+
+ if (!isConnected) {
+ sender ! Failure(
+ new IllegalStateException("TaskManager is not associated with a JobManager.")
+ )
+ } else if (slot < 0 || slot >= numberOfSlots) {
+ sender ! Failure(new Exception(s"Target slot $slot does not exist on TaskManager."))
+ } else {
+ sender ! Acknowledge
+
+ Future {
+ initializeTask(tdd)
+ }(context.dispatcher)
+ }
+ }
+
+ /** Sets up a [[org.apache.flink.runtime.execution.RuntimeEnvironment]] for the task and starts
+ * its execution in a separate thread.
+ *
+ * @param tdd TaskDeploymentDescriptor describing the task to be executed on this [[TaskManager]]
+ */
+ private def initializeTask(tdd: TaskDeploymentDescriptor): Unit ={
val jobID = tdd.getJobID
val vertexID = tdd.getVertexID
val executionID = tdd.getExecutionId
val taskIndex = tdd.getIndexInSubtaskGroup
val numSubtasks = tdd.getNumberOfSubtasks
- val slot = tdd.getTargetSlotNumber
var startRegisteringTask = 0L
var task: Task = null
- // all operations are in a try / catch block to make sure we send a result upon any failure
try {
- // check that we are already registered
- if (!isConnected) {
- throw new IllegalStateException("TaskManager is not associated with a JobManager")
- }
- if (slot < 0 || slot >= numberOfSlots) {
- throw new Exception(s"Target slot ${slot} does not exist on TaskManager.")
- }
-
val userCodeClassLoader = libraryCacheManager match {
case Some(manager) =>
if (log.isDebugEnabled) {
@@ -807,8 +824,10 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
// triggers the download of all missing jar files from the job manager
manager.registerTask(jobID, executionID, tdd.getRequiredJarFiles)
+ if (log.isDebugEnabled) {
log.debug(s"Register task $executionID at library cache manager " +
s"took ${(System.currentTimeMillis() - startRegisteringTask) / 1000.0}s")
+ }
manager.getClassLoader(jobID)
case None => throw new IllegalStateException("There is no valid library cache manager.")
@@ -820,8 +839,8 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
task = new Task(jobID, vertexID, taskIndex, numSubtasks, executionID,
tdd.getTaskName, self)
-
- runningTasks.put(executionID, task) match {
+
+ Option(runningTasks.put(executionID, task)) match {
case Some(_) => throw new RuntimeException(
s"TaskManager contains already a task with executionID $executionID.")
case None =>
@@ -848,7 +867,7 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
opStateCarrier.injectState(tdd.getOperatorStates)
}
}
-
+
// register the task with the network stack and profiles
log.info(s"Register task $task.")
network.registerTask(task)
@@ -865,15 +884,13 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
throw new RuntimeException("Cannot start task. Task was canceled or failed.")
}
- sender ! new TaskOperationResult(executionID, true)
- }
- catch {
+ self ! UpdateTaskExecutionState(
+ new TaskExecutionState(jobID, executionID, ExecutionState.RUNNING)
+ )
+ } catch {
case t: Throwable =>
- val message = if (t.isInstanceOf[CancelTaskException]) {
- "Task was canceled"
- } else {
+ if (!t.isInstanceOf[CancelTaskException]) {
log.error("Could not instantiate task with execution ID " + executionID, t)
- ExceptionUtils.stringifyException(t)
}
try {
@@ -887,7 +904,9 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
case t: Throwable => log.error("Error during cleanup of task deployment.", t)
}
- sender ! new TaskOperationResult(executionID, false, message)
+ self ! UpdateTaskExecutionState(
+ new TaskExecutionState(jobID, executionID, ExecutionState.FAILED, t)
+ )
}
}
@@ -901,7 +920,7 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
executionId: ExecutionAttemptID,
partitionInfos: Seq[(IntermediateDataSetID, InputChannelDeploymentDescriptor)]) : Unit = {
- runningTasks.get(executionId) match {
+ Option(runningTasks.get(executionId)) match {
case Some(task) =>
val errors: Seq[String] = partitionInfos.flatMap { info =>
@@ -958,7 +977,7 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
if (runningTasks.size > 0) {
log.info("Cancelling all computations and discarding all cached data.")
- for (t <- runningTasks.values) {
+ for (t <- runningTasks.values().asScala) {
t.failExternally(cause)
unregisterTaskAndNotifyFinalState(t.getExecutionId)
}
@@ -966,7 +985,8 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
}
private def unregisterTaskAndNotifyFinalState(executionID: ExecutionAttemptID): Unit = {
- runningTasks.remove(executionID) match {
+
+ Option(runningTasks.remove(executionID)) match {
case Some(task) =>
// mark the task as failed if it is not yet in a final state
http://git-wip-us.apache.org/repos/asf/flink/blob/4cb5b725/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 8c4879f..42c3f84 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
@@ -126,7 +126,7 @@ public class ExecutionGraphDeploymentTest {
vertex.deployToSlot(slot);
- assertEquals(ExecutionState.RUNNING, vertex.getExecutionState());
+ assertEquals(ExecutionState.DEPLOYING, vertex.getExecutionState());
TaskDeploymentDescriptor descr = tm.lastTDD;
assertNotNull(descr);
http://git-wip-us.apache.org/repos/asf/flink/blob/4cb5b725/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 7380b36..a1ee79b 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
@@ -27,6 +27,7 @@ import java.net.InetAddress;
import java.util.LinkedList;
import akka.actor.ActorRef;
+import akka.actor.Status;
import akka.actor.UntypedActor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.JobException;
@@ -43,6 +44,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.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;
@@ -119,7 +121,7 @@ public class ExecutionGraphTestUtils {
SubmitTask submitTask = (SubmitTask) msg;
lastTDD = submitTask.tasks();
- getSender().tell(new TaskOperationResult(submitTask.tasks().getExecutionId(), true), getSelf());
+ getSender().tell(Messages.getAcknowledge(), getSelf());
} else if (msg instanceof CancelTask) {
CancelTask cancelTask = (CancelTask) msg;
getSender().tell(new TaskOperationResult(cancelTask.attemptID(), true), getSelf());
@@ -136,10 +138,7 @@ public class ExecutionGraphTestUtils {
@Override
public void onReceive(Object msg) throws Exception {
if (msg instanceof SubmitTask) {
- SubmitTask submitTask = (SubmitTask) msg;
-
- getSender().tell(new TaskOperationResult(submitTask.tasks().getExecutionId(),
- false, ERROR_MESSAGE), getSelf());
+ getSender().tell(new Status.Failure(new Exception(ERROR_MESSAGE)), getSelf());
} else if (msg instanceof CancelTask) {
CancelTask cancelTask = (CancelTask) msg;
getSender().tell(new TaskOperationResult(cancelTask.attemptID(), true), getSelf());
http://git-wip-us.apache.org/repos/asf/flink/blob/4cb5b725/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 226b256..757976d 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
@@ -35,13 +35,13 @@ import akka.japi.Creator;
import akka.testkit.JavaTestKit;
import org.apache.flink.runtime.akka.AkkaUtils;
-import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.instance.SimpleSlot;
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.messages.TaskMessages;
import org.apache.flink.runtime.messages.TaskMessages.TaskOperationResult;
import org.apache.flink.runtime.testingUtils.TestingUtils;
@@ -223,7 +223,7 @@ public class ExecutionVertexCancelTest {
TaskOperationResult(execId, false), new TaskOperationResult(execId, true))));
Instance instance = getInstance(taskManager);
- SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
vertex.deployToSlot(slot);
@@ -252,11 +252,6 @@ public class ExecutionVertexCancelTest {
// the call did not yet execute, so it is still in canceling
assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
- // trigger the correcting cancel call, should properly set state to cancelled
- actions.triggerNextAction();
- // process onComplete callback
- actions.triggerNextAction();
-
vertex.getCurrentExecutionAttempt().cancelingComplete();
assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
@@ -645,8 +640,7 @@ public class ExecutionVertexCancelTest {
@Override
public void onReceive(Object message) throws Exception {
if(message instanceof TaskMessages.SubmitTask){
- TaskDeploymentDescriptor tdd = ((TaskMessages.SubmitTask) message).tasks();
- getSender().tell(new TaskOperationResult(tdd.getExecutionId(), true), getSelf());
+ getSender().tell(Messages.getAcknowledge(), getSelf());
}else if(message instanceof TaskMessages.CancelTask){
index++;
if(index >= results.length){
http://git-wip-us.apache.org/repos/asf/flink/blob/4cb5b725/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexDeploymentTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexDeploymentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexDeploymentTest.java
index 7b72669..c08ae01 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexDeploymentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexDeploymentTest.java
@@ -22,6 +22,7 @@ import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.*;
import static org.junit.Assert.*;
+import akka.actor.Actor;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
@@ -73,7 +74,7 @@ public class ExecutionVertexDeploymentTest {
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
vertex.deployToSlot(slot);
- assertEquals(ExecutionState.RUNNING, vertex.getExecutionState());
+ assertEquals(ExecutionState.DEPLOYING, vertex.getExecutionState());
// no repeated scheduling
try {
@@ -102,7 +103,7 @@ public class ExecutionVertexDeploymentTest {
final JobVertexID jid = new JobVertexID();
- final TestActorRef<?> simpleTaskManager = TestActorRef.create(system,
+ final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
Props.create(SimpleAcknowledgingTaskManager.class));
final Instance instance = getInstance(simpleTaskManager);
@@ -118,7 +119,7 @@ public class ExecutionVertexDeploymentTest {
vertex.deployToSlot(slot);
- assertEquals(ExecutionState.RUNNING, vertex.getExecutionState());
+ assertEquals(ExecutionState.DEPLOYING, vertex.getExecutionState());
// no repeated scheduling
try {
@@ -131,7 +132,7 @@ public class ExecutionVertexDeploymentTest {
assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.DEPLOYING) > 0);
- assertTrue(vertex.getStateTimestamp(ExecutionState.RUNNING) > 0);
+ assertTrue(vertex.getStateTimestamp(ExecutionState.RUNNING) == 0);
}
catch (Exception e) {
e.printStackTrace();
@@ -146,7 +147,7 @@ public class ExecutionVertexDeploymentTest {
try {
final JobVertexID jid = new JobVertexID();
- final TestActorRef<?> simpleTaskManager = TestActorRef.create(system,
+ final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
Props.create(SimpleAcknowledgingTaskManager.class));
final Instance instance = getInstance(simpleTaskManager);
@@ -168,15 +169,8 @@ public class ExecutionVertexDeploymentTest {
fail("Scheduled from wrong state");
}
catch (IllegalStateException e) {}
-
- // wait until the state transition must be done
- for (int i = 0; i < 100; i++) {
- if (vertex.getExecutionState() != ExecutionState.RUNNING) {
- Thread.sleep(10);
- }
- }
-
- assertEquals(ExecutionState.RUNNING, vertex.getExecutionState());
+
+ assertEquals(ExecutionState.DEPLOYING, vertex.getExecutionState());
// no repeated scheduling
try {
@@ -187,7 +181,7 @@ public class ExecutionVertexDeploymentTest {
assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.DEPLOYING) > 0);
- assertTrue(vertex.getStateTimestamp(ExecutionState.RUNNING) > 0);
+ assertTrue(vertex.getStateTimestamp(ExecutionState.RUNNING) == 0);
}
catch (Exception e) {
e.printStackTrace();
@@ -202,7 +196,7 @@ public class ExecutionVertexDeploymentTest {
final JobVertexID jid = new JobVertexID();
- final TestActorRef<?> simpleTaskManager = TestActorRef.create(system,
+ final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
Props.create(SimpleFailingTaskManager.class));
final Instance instance = getInstance(simpleTaskManager);
@@ -238,7 +232,7 @@ public class ExecutionVertexDeploymentTest {
try {
final JobVertexID jid = new JobVertexID();
- final TestActorRef<?> simpleTaskManager = TestActorRef.create(system,
+ final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
Props.create(SimpleFailingTaskManager.class));
final Instance instance = getInstance(simpleTaskManager);
@@ -287,7 +281,7 @@ public class ExecutionVertexDeploymentTest {
TestingUtils.setExecutionContext(ec);
- final TestActorRef<?> simpleTaskManager = TestActorRef.create(system,
+ final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
Props.create(SimpleAcknowledgingTaskManager.class));
final Instance instance = getInstance(simpleTaskManager);
final SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
@@ -337,7 +331,7 @@ public class ExecutionVertexDeploymentTest {
AkkaUtils.getDefaultTimeout());
final ExecutionAttemptID eid = vertex.getCurrentExecutionAttempt().getAttemptId();
- final TestActorRef<?> simpleTaskManager = TestActorRef.create(system, Props.create(new
+ final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system, Props.create(new
ExecutionVertexCancelTest.CancelSequenceTaskManagerCreator(new
TaskOperationResult(eid, false), new TaskOperationResult(eid, true))));
@@ -366,11 +360,6 @@ public class ExecutionVertexDeploymentTest {
assertEquals(ExecutionState.FAILED, vertex.getExecutionState());
- // should have sent another cancel call
- queue.triggerNextAction();
- // execute onComplete callback
- queue.triggerNextAction();
-
assertEquals(testError, vertex.getFailureCause());
assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
http://git-wip-us.apache.org/repos/asf/flink/blob/4cb5b725/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexSchedulingTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexSchedulingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexSchedulingTest.java
index e24a2b4..06f0e9d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexSchedulingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexSchedulingTest.java
@@ -129,7 +129,7 @@ public class ExecutionVertexSchedulingTest {
}
@Test
- public void testScheduleToRunning() {
+ public void testScheduleToDeploying() {
try {
TestingUtils.setCallingThreadDispatcher(system);
ActorRef tm = TestActorRef.create(system, Props.create(ExecutionGraphTestUtils
@@ -149,7 +149,7 @@ public class ExecutionVertexSchedulingTest {
// try to deploy to the slot
vertex.scheduleForExecution(scheduler, false);
- assertEquals(ExecutionState.RUNNING, vertex.getExecutionState());
+ assertEquals(ExecutionState.DEPLOYING, vertex.getExecutionState());
}
catch (Exception e) {
e.printStackTrace();
http://git-wip-us.apache.org/repos/asf/flink/blob/4cb5b725/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
index 8b7915f..056ed37 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
@@ -47,6 +47,7 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.jobmanager.Tasks;
import org.apache.flink.runtime.messages.JobManagerMessages;
+import org.apache.flink.runtime.messages.Messages;
import org.apache.flink.runtime.messages.RegistrationMessages;
import org.apache.flink.runtime.messages.TaskManagerMessages;
import org.apache.flink.runtime.messages.TaskMessages;
@@ -67,8 +68,10 @@ import scala.concurrent.duration.FiniteDuration;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals;
@@ -83,6 +86,8 @@ public class TaskManagerTest {
private static Timeout timeout = new Timeout(1, TimeUnit.MINUTES);
+ private final FiniteDuration d = new FiniteDuration(20, TimeUnit.SECONDS);
+
@BeforeClass
public static void setup() {
system = ActorSystem.create("TestActorSystem", TestingUtils.testConfig());
@@ -115,12 +120,12 @@ public class TaskManagerTest {
new ArrayList<BlobKey>(), 0);
final ActorRef tmClosure = taskManager;
- new Within(duration("10 seconds")) {
+ new Within(d) {
@Override
protected void run() {
tmClosure.tell(new SubmitTask(tdd), getRef());
- expectMsgEquals(new TaskOperationResult(eid, true));
+ expectMsgEquals(Messages.getAcknowledge());
}
};
}
@@ -172,18 +177,29 @@ public class TaskManagerTest {
new ArrayList<BlobKey>(), 0);
final ActorRef tm = taskManager;
- final FiniteDuration d = duration("10 second");
new Within(d) {
@Override
protected void run() {
try {
+ Future<Object> t1Running = Patterns.ask(
+ tm,
+ new TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid1),
+ timeout);
+ Future<Object> t2Running = Patterns.ask(
+ tm,
+ new TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid2),
+ timeout);
+
tm.tell(new SubmitTask(tdd1), getRef());
tm.tell(new SubmitTask(tdd2), getRef());
- expectMsgEquals(new TaskOperationResult(eid1, true));
- expectMsgEquals(new TaskOperationResult(eid2, true));
+ expectMsgEquals(Messages.getAcknowledge());
+ expectMsgEquals(Messages.getAcknowledge());
+
+ Await.ready(t1Running, d);
+ Await.ready(t2Running, d);
tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
@@ -288,34 +304,34 @@ public class TaskManagerTest {
Collections.<InputGateDeploymentDescriptor>emptyList(),
new ArrayList<BlobKey>(), 0);
- new Within(duration("10 second")){
+ new Within(d){
@Override
protected void run() {
- tm.tell(new SubmitTask(tdd1), getRef());
- tm.tell(new SubmitTask(tdd2), getRef());
-
- TaskOperationResult result = expectMsgClass(TaskOperationResult.class);
- assertFalse(result.success());
- assertEquals(eid1, result.executionID());
+ try {
+ tm.tell(new SubmitTask(tdd1), getRef());
+ tm.tell(new SubmitTask(tdd2), getRef());
- result = expectMsgClass(TaskOperationResult.class);
- assertFalse(result.success());
- assertEquals(eid2, result.executionID());
+ expectMsgEquals(Messages.getAcknowledge());
+ expectMsgEquals(Messages.getAcknowledge());
- tm.tell(new TestingTaskManagerMessages.NotifyWhenTaskRemoved(eid1),
- getRef());
- tm.tell(new TestingTaskManagerMessages.NotifyWhenTaskRemoved(eid2),
- getRef());
+ tm.tell(new TestingTaskManagerMessages.NotifyWhenTaskRemoved(eid1),
+ getRef());
+ tm.tell(new TestingTaskManagerMessages.NotifyWhenTaskRemoved(eid2),
+ getRef());
- expectMsgEquals(true);
- expectMsgEquals(true);
+ expectMsgEquals(true);
+ expectMsgEquals(true);
- tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
- Map<ExecutionAttemptID, Task> tasks = expectMsgClass(TestingTaskManagerMessages
- .ResponseRunningTasks.class).asJava();
+ tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
+ Map<ExecutionAttemptID, Task> tasks = expectMsgClass(TestingTaskManagerMessages
+ .ResponseRunningTasks.class).asJava();
- assertEquals(0, tasks.size());
+ assertEquals(0, tasks.size());
+ } catch (Exception e){
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
}
};
}
@@ -378,17 +394,27 @@ public class TaskManagerTest {
Collections.singletonList(ircdd),
new ArrayList<BlobKey>(), 0);
- final FiniteDuration d = duration("10 second");
-
new Within(d) {
@Override
protected void run() {
try {
+ Future<Object> t1Running = Patterns.ask(
+ tm,
+ new TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid1),
+ timeout);
+ Future<Object> t2Running = Patterns.ask(
+ tm,
+ new TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid2),
+ timeout);
+
tm.tell(new SubmitTask(tdd1), getRef());
- expectMsgEquals(new TaskOperationResult(eid1, true));
+ expectMsgEquals(Messages.getAcknowledge());
tm.tell(new SubmitTask(tdd2), getRef());
- expectMsgEquals(new TaskOperationResult(eid2, true));
+ expectMsgEquals(Messages.getAcknowledge());
+
+ Await.ready(t1Running, d);
+ Await.ready(t2Running, d);
tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
Map<ExecutionAttemptID, Task> tasks = expectMsgClass(TestingTaskManagerMessages.ResponseRunningTasks
@@ -460,7 +486,11 @@ public class TaskManagerTest {
final ExecutionAttemptID eid1 = new ExecutionAttemptID();
final ExecutionAttemptID eid2 = new ExecutionAttemptID();
- jobManager = system.actorOf(Props.create(new SimpleLookupFailingUpdateJobManagerCreator()));
+ jobManager = system.actorOf(
+ Props.create(
+ new SimpleLookupFailingUpdateJobManagerCreator(eid2)
+ )
+ );
taskManager = createTaskManager(jobManager);
final ActorRef tm = taskManager;
@@ -488,19 +518,28 @@ public class TaskManagerTest {
Collections.singletonList(ircdd),
new ArrayList<BlobKey>(), 0);
- final FiniteDuration d = duration("10 second");
-
new Within(d){
@Override
protected void run() {
try {
- // deploy sender before receiver, so the target is online when the sender requests the connection info
+ Future<Object> t1Running = Patterns.ask(
+ tm,
+ new TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid1),
+ timeout);
+ Future<Object> t2Running = Patterns.ask(
+ tm,
+ new TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid2),
+ timeout);
+
tm.tell(new SubmitTask(tdd2), getRef());
tm.tell(new SubmitTask(tdd1), getRef());
- expectMsgEquals(new TaskOperationResult(eid2, true));
- expectMsgEquals(new TaskOperationResult(eid1, true));
+ expectMsgEquals(Messages.getAcknowledge());
+ expectMsgEquals(Messages.getAcknowledge());
+
+ Await.ready(t1Running, d);
+ Await.ready(t2Running, d);
tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
Map<ExecutionAttemptID, Task> tasks = expectMsgClass(TestingTaskManagerMessages
@@ -588,10 +627,23 @@ public class TaskManagerTest {
public static class SimpleLookupFailingUpdateJobManager extends SimpleLookupJobManager{
+ private final Set<ExecutionAttemptID> validIDs;
+
+ public SimpleLookupFailingUpdateJobManager(Set<ExecutionAttemptID> ids) {
+ this.validIDs = new HashSet<ExecutionAttemptID>(ids);
+ }
+
@Override
public void onReceive(Object message) throws Exception{
if (message instanceof TaskMessages.UpdateTaskExecutionState) {
- getSender().tell(false, getSelf());
+ TaskMessages.UpdateTaskExecutionState updateMsg =
+ (TaskMessages.UpdateTaskExecutionState) message;
+
+ if(validIDs.contains(updateMsg.taskExecutionState().getID())) {
+ getSender().tell(true, getSelf());
+ } else {
+ getSender().tell(false, getSelf());
+ }
} else {
super.onReceive(message);
}
@@ -608,9 +660,19 @@ public class TaskManagerTest {
public static class SimpleLookupFailingUpdateJobManagerCreator implements Creator<SimpleLookupFailingUpdateJobManager>{
+ private final Set<ExecutionAttemptID> validIDs;
+
+ public SimpleLookupFailingUpdateJobManagerCreator(ExecutionAttemptID ... ids) {
+ validIDs = new HashSet<ExecutionAttemptID>();
+
+ for(ExecutionAttemptID id : ids) {
+ this.validIDs.add(id);
+ }
+ }
+
@Override
public SimpleLookupFailingUpdateJobManager create() throws Exception {
- return new SimpleLookupFailingUpdateJobManager();
+ return new SimpleLookupFailingUpdateJobManager(validIDs);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4cb5b725/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
index 702e34c..bb0c1f9 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
@@ -20,13 +20,14 @@ package org.apache.flink.runtime.testingUtils
import akka.actor.{Terminated, ActorRef}
import org.apache.flink.api.common.JobID
+import org.apache.flink.runtime.execution.ExecutionState
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID
import org.apache.flink.runtime.instance.InstanceConnectionInfo
import org.apache.flink.runtime.io.disk.iomanager.IOManager
import org.apache.flink.runtime.io.network.NetworkEnvironment
import org.apache.flink.runtime.memorymanager.DefaultMemoryManager
import org.apache.flink.runtime.messages.Messages.Disconnect
-import org.apache.flink.runtime.messages.TaskMessages.UnregisterTask
+import org.apache.flink.runtime.messages.TaskMessages.{UpdateTaskExecutionState, UnregisterTask}
import org.apache.flink.runtime.taskmanager.{TaskManagerConfiguration, TaskManager}
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages.NotifyWhenJobRemoved
import org.apache.flink.runtime.testingUtils.TestingMessages.DisableDisconnect
@@ -48,10 +49,14 @@ class TestingTaskManager(config: TaskManagerConfiguration,
extends TaskManager(config, connectionInfo, jobManagerAkkaURL,
memoryManager, ioManager, network, numberOfSlots) {
+ import scala.collection.JavaConverters._
+
val waitForRemoval = scala.collection.mutable.HashMap[ExecutionAttemptID, Set[ActorRef]]()
val waitForJobRemoval = scala.collection.mutable.HashMap[JobID, Set[ActorRef]]()
val waitForJobManagerToBeTerminated = scala.collection.mutable.HashMap[String, Set[ActorRef]]()
+ val waitForRunning = scala.collection.mutable.HashMap[ExecutionAttemptID, Set[ActorRef]]()
+ val unregisteredTasks = scala.collection.mutable.HashSet[ExecutionAttemptID]()
var disconnectDisabled = false
@@ -64,16 +69,30 @@ class TestingTaskManager(config: TaskManagerConfiguration,
* Handler for testing related messages
*/
def receiveTestMessages: Receive = {
+ case NotifyWhenTaskIsRunning(executionID) => {
+ Option(runningTasks.get(executionID)) match {
+ case Some(_) => sender ! true
+ case None =>
+ val listeners = waitForRunning.getOrElse(executionID, Set())
+ waitForRunning += (executionID -> (listeners + sender))
+ }
+ }
case RequestRunningTasks =>
- sender ! ResponseRunningTasks(runningTasks.toMap)
+ sender ! ResponseRunningTasks(runningTasks.asScala.toMap)
case NotifyWhenTaskRemoved(executionID) =>
- runningTasks.get(executionID) match {
+ Option(runningTasks.get(executionID)) match {
case Some(_) =>
val set = waitForRemoval.getOrElse(executionID, Set())
waitForRemoval += (executionID -> (set + sender))
- case None => sender ! true
+ case None =>
+ if(unregisteredTasks.contains(executionID)) {
+ sender ! true
+ } else {
+ val set = waitForRemoval.getOrElse(executionID, Set())
+ waitForRemoval += (executionID -> (set + sender))
+ }
}
case UnregisterTask(executionID) =>
@@ -82,6 +101,8 @@ class TestingTaskManager(config: TaskManagerConfiguration,
case Some(actors) => for(actor <- actors) actor ! true
case None =>
}
+
+ unregisteredTasks += executionID
case RequestBroadcastVariablesWithReferences =>
sender ! ResponseBroadcastVariablesWithReferences(
@@ -96,7 +117,7 @@ class TestingTaskManager(config: TaskManagerConfiguration,
sender ! ResponseNumActiveConnections(numActive)
case NotifyWhenJobRemoved(jobID) =>
- if(runningTasks.values.exists(_.getJobID == jobID)){
+ if(runningTasks.values.asScala.exists(_.getJobID == jobID)){
val set = waitForJobRemoval.getOrElse(jobID, Set())
waitForJobRemoval += (jobID -> (set + sender))
import context.dispatcher
@@ -109,7 +130,7 @@ class TestingTaskManager(config: TaskManagerConfiguration,
}
case CheckIfJobRemoved(jobID) =>
- if(runningTasks.values.forall(_.getJobID != jobID)){
+ if(runningTasks.values.asScala.forall(_.getJobID != jobID)){
waitForJobRemoval.remove(jobID) match {
case Some(listeners) => listeners foreach (_ ! true)
case None =>
@@ -147,5 +168,14 @@ class TestingTaskManager(config: TaskManagerConfiguration,
case DisableDisconnect =>
disconnectDisabled = true
+
+ case msg @ UpdateTaskExecutionState(taskExecutionState) =>
+ super.receiveWithLogMessages(msg)
+
+ if(taskExecutionState.getExecutionState == ExecutionState.RUNNING) {
+ waitForRunning.get(taskExecutionState.getID) foreach {
+ _ foreach (_ ! true)
+ }
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4cb5b725/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManagerMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManagerMessages.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManagerMessages.scala
index 2051ef5..c9a2f73 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManagerMessages.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManagerMessages.scala
@@ -29,6 +29,8 @@ import org.apache.flink.runtime.taskmanager.Task
object TestingTaskManagerMessages {
case class NotifyWhenTaskRemoved(executionID: ExecutionAttemptID)
+
+ case class NotifyWhenTaskIsRunning(executionID: ExecutionAttemptID)
case class ResponseRunningTasks(tasks: Map[ExecutionAttemptID, Task]){
import collection.JavaConverters._
[3/3] flink git commit: [FLINK-1922] [runtime] Fixes NPE when TM
receives a null input split
Posted by se...@apache.org.
[FLINK-1922] [runtime] Fixes NPE when TM receives a null input split
This closes #631
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ccd574a4
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ccd574a4
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ccd574a4
Branch: refs/heads/master
Commit: ccd574a46e6fce44a9c1d0bf0ec72424c8252c98
Parents: 9a18e57
Author: Till Rohrmann <tr...@apache.org>
Authored: Mon Apr 27 14:30:16 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Apr 29 10:40:39 2015 +0200
----------------------------------------------------------------------
.../flink/runtime/executiongraph/Execution.java | 6 +-
.../taskmanager/TaskInputSplitProvider.java | 15 ++--
.../taskmanager/TaskInputSplitProviderTest.java | 93 ++++++++++++++++++++
3 files changed, 105 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ccd574a4/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index 3ba378c..baed947 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -328,7 +328,7 @@ public class Execution implements Serializable {
// register this execution at the execution graph, to receive call backs
vertex.getExecutionGraph().registerExecution(this);
- Instance instance = slot.getInstance();
+ final Instance instance = slot.getInstance();
Future<Object> deployAction = Patterns.ask(instance.getTaskManager(),
new SubmitTask(deployment), new Timeout(timeout));
@@ -338,7 +338,9 @@ public class Execution implements Serializable {
public void onComplete(Throwable failure, Object success) throws Throwable {
if (failure != null) {
if (failure instanceof TimeoutException) {
- markFailed(new Exception("Cannot deploy task - TaskManager not responding.", failure));
+ markFailed(new Exception(
+ "Cannot deploy task - TaskManager " + instance + " not responding.",
+ failure));
}
else {
markFailed(failure);
http://git-wip-us.apache.org/repos/asf/flink/blob/ccd574a4/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 1bdc346..5a69850 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
@@ -68,10 +68,6 @@ public class TaskInputSplitProvider implements InputSplitProvider {
final Object result = Await.result(response, timeout.duration());
- if (result == null) {
- return null;
- }
-
if(!(result instanceof JobManagerMessages.NextInputSplit)){
throw new RuntimeException("RequestNextInputSplit requires a response of type " +
"NextInputSplit. Instead response is of type " + result.getClass() + ".");
@@ -80,9 +76,14 @@ public class TaskInputSplitProvider implements InputSplitProvider {
(JobManagerMessages.NextInputSplit) result;
byte[] serializedData = nextInputSplit.splitData();
- Object deserialized = InstantiationUtil.deserializeObject(serializedData,
- usercodeClassLoader);
- return (InputSplit) deserialized;
+
+ if(serializedData == null) {
+ return null;
+ } else {
+ Object deserialized = InstantiationUtil.deserializeObject(serializedData,
+ usercodeClassLoader);
+ return (InputSplit) deserialized;
+ }
}
} catch (Exception e) {
throw new RuntimeException("Requesting the next InputSplit failed.", e);
http://git-wip-us.apache.org/repos/asf/flink/blob/ccd574a4/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskInputSplitProviderTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskInputSplitProviderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskInputSplitProviderTest.java
new file mode 100644
index 0000000..f068978
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskInputSplitProviderTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.taskmanager;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import akka.actor.Status;
+import akka.actor.UntypedActor;
+import akka.testkit.JavaTestKit;
+import akka.util.Timeout;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.io.InputSplit;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.messages.JobManagerMessages;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+import java.util.concurrent.TimeUnit;
+
+public class TaskInputSplitProviderTest {
+
+ private static ActorSystem system;
+
+ @BeforeClass
+ public static void setup() throws Exception {
+ system = ActorSystem.create("TestActorSystem", TestingUtils.testConfig());
+ }
+
+ @AfterClass
+ public static void teardown() throws Exception {
+ JavaTestKit.shutdownActorSystem(system);
+ system = null;
+ }
+
+ @Test
+ public void testRequestNextInputSplitWithInvalidExecutionID() {
+
+ final JobID jobID = new JobID();
+ final JobVertexID vertexID = new JobVertexID();
+ final ExecutionAttemptID executionID = new ExecutionAttemptID();
+ final Timeout timeout = new Timeout(10, TimeUnit.SECONDS);
+
+ final ActorRef jobManagerRef = system.actorOf(Props.create(NullInputSplitJobManager.class));
+
+ final TaskInputSplitProvider provider = new TaskInputSplitProvider(
+ jobManagerRef,
+ jobID,
+ vertexID,
+ executionID,
+ getClass().getClassLoader(),
+ timeout
+ );
+
+ // The jobManager will return a
+ InputSplit nextInputSplit = provider.getNextInputSplit();
+
+ assertTrue(nextInputSplit == null);
+ }
+
+ public static class NullInputSplitJobManager extends UntypedActor {
+
+ @Override
+ public void onReceive(Object message) throws Exception {
+ if(message instanceof JobManagerMessages.RequestNextInputSplit) {
+ sender().tell(new JobManagerMessages.NextInputSplit(null), getSelf());
+ } else {
+ sender().tell(new Status.Failure(new Exception("Invalid message type")), getSelf());
+ }
+ }
+ }
+}
[2/3] flink git commit: [FLINK-1923] [runtime] Replaces asynchronous
logging with synchronous logging using grizzled-slf4j wrapper for Scala.
Posted by se...@apache.org.
[FLINK-1923] [runtime] Replaces asynchronous logging with synchronous logging using grizzled-slf4j wrapper for Scala.
This closes #628
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5a2ca819
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5a2ca819
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5a2ca819
Branch: refs/heads/master
Commit: 5a2ca81912193552e74cd6a33637b7254e5a7174
Parents: ccd574a
Author: Till Rohrmann <tr...@apache.org>
Authored: Fri Apr 24 16:33:34 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Apr 29 10:40:39 2015 +0200
----------------------------------------------------------------------
flink-dist/src/main/flink-bin/LICENSE | 1 +
flink-runtime/pom.xml | 5 +
.../apache/flink/runtime/taskmanager/Task.java | 11 --
.../apache/flink/runtime/ActorLogMessages.scala | 5 +-
.../flink/runtime/ActorSynchronousLogging.scala | 31 ++++
.../flink/runtime/jobmanager/JobManager.scala | 91 +++++-----
.../runtime/jobmanager/JobManagerProfiler.scala | 11 +-
.../runtime/jobmanager/MemoryArchivist.scala | 10 +-
.../StreamCheckpointCoordinator.scala | 14 +-
.../flink/runtime/taskmanager/TaskManager.scala | 167 +++++++++----------
.../taskmanager/TaskManagerProfiler.scala | 8 +-
.../apache/flink/yarn/ApplicationClient.scala | 16 +-
.../apache/flink/yarn/ApplicationMaster.scala | 9 +-
.../flink/yarn/ApplicationMasterActor.scala | 79 +++++----
pom.xml | 6 +
15 files changed, 243 insertions(+), 221 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5a2ca819/flink-dist/src/main/flink-bin/LICENSE
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/flink-bin/LICENSE b/flink-dist/src/main/flink-bin/LICENSE
index d66922c..1236381 100644
--- a/flink-dist/src/main/flink-bin/LICENSE
+++ b/flink-dist/src/main/flink-bin/LICENSE
@@ -308,6 +308,7 @@ BSD-style licenses:
- Scala Compiler Reflect (http://www.scala-lang.org/) - Copyright (c) 2002-2014 EPFL, Copyright (c) 2011-2014 Typesafe, Inc.
- Scala Quasiquotes (http://scalamacros.org/) - Copyright (c) 2002-2014 EPFL, Copyright (c) 2011-2014 Typesafe, Inc.
- ASM (http://asm.ow2.org/) - Copyright (c) 2000-2011 INRIA, France Telecom
+ - Grizzled SLF4J (http://software.clapper.org/grizzled-slf4j/) - Copyright (c) 2010 Brian M. Clapper
(Below is the 3-clause BSD license)
http://git-wip-us.apache.org/repos/asf/flink/blob/5a2ca819/flink-runtime/pom.xml
----------------------------------------------------------------------
diff --git a/flink-runtime/pom.xml b/flink-runtime/pom.xml
index f490ed9..56ed0d2 100644
--- a/flink-runtime/pom.xml
+++ b/flink-runtime/pom.xml
@@ -148,6 +148,11 @@ under the License.
</dependency>
<dependency>
+ <groupId>org.clapper</groupId>
+ <artifactId>grizzled-slf4j_${scala.binary.version}</artifactId>
+ </dependency>
+
+ <dependency>
<groupId>org.scalatest</groupId>
<artifactId>scalatest_${scala.binary.version}</artifactId>
</dependency>
http://git-wip-us.apache.org/repos/asf/flink/blob/5a2ca819/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 f6eb907..e6eee5b 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
@@ -30,7 +30,6 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.memorymanager.MemoryManager;
import org.apache.flink.runtime.messages.ExecutionGraphMessages;
-import org.apache.flink.runtime.messages.TaskMessages;
import org.apache.flink.runtime.messages.TaskMessages.UnregisterTask;
import org.apache.flink.runtime.profiling.TaskManagerProfiler;
import org.apache.flink.util.ExceptionUtils;
@@ -357,16 +356,6 @@ public class Task {
taskManager.tell(new UnregisterTask(executionId), ActorRef.noSender());
}
- protected void notifyExecutionStateChange(ExecutionState executionState,
- Throwable optionalError) {
- LOG.info("Update execution state of {} ({}) to {}.", this.getTaskName(),
- this.getExecutionId(), executionState);
- taskManager.tell(new TaskMessages.UpdateTaskExecutionState(
- new TaskExecutionState(jobId, executionId, executionState, optionalError)),
- ActorRef.noSender());
-
- }
-
// -----------------------------------------------------------------------------------------------------------------
// Task Profiling
// -----------------------------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5a2ca819/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
index acd4346..c74c339 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorLogMessages.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorLogMessages.scala
@@ -19,13 +19,12 @@
package org.apache.flink.runtime
import _root_.akka.actor.Actor
-import _root_.akka.event.LoggingAdapter
/**
* Mixin to add debug message logging
*/
trait ActorLogMessages {
- that: Actor =>
+ that: Actor with ActorSynchronousLogging =>
override def receive: Receive = new Actor.Receive {
private val _receiveWithLogMessages = receiveWithLogMessages
@@ -50,6 +49,4 @@ trait ActorLogMessages {
}
def receiveWithLogMessages: Receive
-
- protected def log: LoggingAdapter
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5a2ca819/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
new file mode 100644
index 0000000..4d3a988
--- /dev/null
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorSynchronousLogging.scala
@@ -0,0 +1,31 @@
+/*
+ * 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/5a2ca819/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 6870d95..83f9e35 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
@@ -23,6 +23,7 @@ import java.net.InetSocketAddress
import java.util.Collections
import akka.actor.Status.{Success, Failure}
+import grizzled.slf4j.Logger
import org.apache.flink.api.common.{JobID, ExecutionConfig}
import org.apache.flink.configuration.{ConfigConstants, GlobalConfiguration, Configuration}
import org.apache.flink.core.io.InputSplitAssigner
@@ -42,7 +43,7 @@ import org.apache.flink.runtime.security.SecurityUtils
import org.apache.flink.runtime.security.SecurityUtils.FlinkSecuredRunner
import org.apache.flink.runtime.taskmanager.TaskManager
import org.apache.flink.runtime.util.{SerializedValue, EnvironmentInformation}
-import org.apache.flink.runtime.ActorLogMessages
+import org.apache.flink.runtime.{ActorSynchronousLogging, ActorLogMessages}
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager
import org.apache.flink.runtime.instance.InstanceManager
@@ -101,10 +102,7 @@ class JobManager(val flinkConfiguration: Configuration,
val defaultExecutionRetries: Int,
val delayBetweenRetries: Long,
val timeout: FiniteDuration)
- extends Actor with ActorLogMessages with ActorLogging {
-
- /** Reference to the log, for debugging */
- val LOG = JobManager.LOG
+ extends Actor with ActorLogMessages with ActorSynchronousLogging {
/** List of current jobs running jobs */
val currentJobs = scala.collection.mutable.HashMap[JobID, (ExecutionGraph, JobInfo)]()
@@ -114,7 +112,7 @@ class JobManager(val flinkConfiguration: Configuration,
* Run when the job manager is started. Simply logs an informational message.
*/
override def preStart(): Unit = {
- LOG.info(s"Starting JobManager at ${self.path.toSerializationFormat}.")
+ log.info(s"Starting JobManager at ${self.path.toSerializationFormat}.")
}
override def postStop(): Unit = {
@@ -138,12 +136,11 @@ class JobManager(val flinkConfiguration: Configuration,
try {
libraryCacheManager.shutdown()
} catch {
- case e: IOException => log.error(e, "Could not properly shutdown the library cache manager.")
+ case e: IOException => log.error("Could not properly shutdown the library cache manager.", e)
}
- if (log.isDebugEnabled) {
- log.debug("Job manager {} is completely stopped.", self.path)
- }
+ log.debug(s"Job manager ${self.path} is completely stopped.")
+
}
/**
@@ -179,7 +176,7 @@ class JobManager(val flinkConfiguration: Configuration,
// registerTaskManager throws an IllegalStateException if it is already shut down
// let the actor crash and restart itself in this case
case e: Exception =>
- log.error(e, "Failed to register TaskManager at instance manager")
+ log.error("Failed to register TaskManager at instance manager", e)
// IMPORTANT: Send the response to the "sender", which is not the
// TaskManager actor, but the ask future!
@@ -197,7 +194,7 @@ class JobManager(val flinkConfiguration: Configuration,
submitJob(jobGraph, listenToEvents = listen)
case CancelJob(jobID) =>
- log.info("Trying to cancel job with ID {}.", jobID)
+ log.info(s"Trying to cancel job with ID $jobID.")
currentJobs.get(jobID) match {
case Some((executionGraph, _)) =>
@@ -208,7 +205,7 @@ class JobManager(val flinkConfiguration: Configuration,
sender ! CancellationSuccess(jobID)
case None =>
- log.info("No job found with ID {}.", jobID)
+ log.info(s"No job found with ID $jobID.")
sender ! CancellationFailure(jobID, new IllegalArgumentException("No job found with " +
s"ID $jobID."))
}
@@ -227,8 +224,9 @@ class JobManager(val flinkConfiguration: Configuration,
}(context.dispatcher)
sender ! true
- case None => log.error("Cannot find execution graph for ID {} to change state to {}.",
- taskExecutionState.getJobID, taskExecutionState.getExecutionState)
+ case None => log.error("Cannot find execution graph for ID " +
+ s"${taskExecutionState.getJobID} to change state to " +
+ s"${taskExecutionState.getExecutionState}.")
sender ! false
}
}
@@ -239,7 +237,7 @@ class JobManager(val flinkConfiguration: Configuration,
val execution = executionGraph.getRegisteredExecutions.get(executionAttempt)
if (execution == null) {
- log.error("Can not find Execution for attempt {}.", executionAttempt)
+ log.error(s"Can not find Execution for attempt $executionAttempt.")
null
} else {
val slot = execution.getAssignedResource
@@ -256,32 +254,30 @@ class JobManager(val flinkConfiguration: Configuration,
case splitAssigner: InputSplitAssigner =>
val nextInputSplit = splitAssigner.getNextInputSplit(host, taskId)
- if (log.isDebugEnabled) {
- log.debug("Send next input split {}.", nextInputSplit)
- }
+ log.debug(s"Send next input split $nextInputSplit.")
try {
InstantiationUtil.serializeObject(nextInputSplit)
} catch {
case ex: Exception =>
- log.error(ex, "Could not serialize the next input split of class {}.",
- nextInputSplit.getClass)
+ log.error(s"Could not serialize the next input split of " +
+ s"class ${nextInputSplit.getClass}.", ex)
vertex.fail(new RuntimeException("Could not serialize the next input split " +
"of class " + nextInputSplit.getClass + ".", ex))
null
}
case _ =>
- log.error("No InputSplitAssigner for vertex ID {}.", vertexID)
+ log.error(s"No InputSplitAssigner for vertex ID $vertexID.")
null
}
case _ =>
- log.error("Cannot find execution vertex for vertex ID {}.", vertexID)
+ log.error(s"Cannot find execution vertex for vertex ID $vertexID.")
null
}
}
case None =>
- log.error("Cannot find execution graph for job ID {}.", jobID)
+ log.error(s"Cannot find execution graph for job ID $jobID.")
null
}
@@ -290,9 +286,8 @@ class JobManager(val flinkConfiguration: Configuration,
case JobStatusChanged(jobID, newJobStatus, timeStamp, error) =>
currentJobs.get(jobID) match {
case Some((executionGraph, jobInfo)) => executionGraph.getJobName
- log.info("Status of job {} ({}) changed to {} {}.",
- jobID, executionGraph.getJobName, newJobStatus,
- if (error == null) "" else error.getMessage)
+ log.info(s"Status of job $jobID (${executionGraph.getJobName}) changed to $newJobStatus" +
+ s" ${if (error == null) "" else error.getMessage}.")
if (newJobStatus.isTerminalState) {
jobInfo.end = timeStamp
@@ -304,7 +299,7 @@ class JobManager(val flinkConfiguration: Configuration,
accumulatorManager.getJobAccumulatorResultsSerialized(jobID)
} catch {
case e: Exception =>
- log.error(e, "Cannot fetch serialized accumulators for job {}", jobID)
+ log.error(s"Cannot fetch serialized accumulators for job $jobID", e)
Collections.emptyMap()
}
val result = new SerializedJobExecutionResult(jobID, jobInfo.duration,
@@ -352,8 +347,8 @@ class JobManager(val flinkConfiguration: Configuration,
sender ! Acknowledge
executionGraph.scheduleOrUpdateConsumers(partitionId)
case None =>
- log.error("Cannot find execution graph for job ID {} to schedule or update consumers",
- jobId)
+ 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."))
}
@@ -383,7 +378,7 @@ class JobManager(val flinkConfiguration: Configuration,
sender ! RunningJobsStatus(jobs)
}
catch {
- case t: Throwable => LOG.error("Exception while responding to RequestRunningJobsStatus", t)
+ case t: Throwable => log.error("Exception while responding to RequestRunningJobsStatus", t)
}
case RequestJob(jobID) =>
@@ -403,10 +398,10 @@ class JobManager(val flinkConfiguration: Configuration,
case Heartbeat(instanceID, metricsReport) =>
try {
- log.debug("Received hearbeat message from {}", instanceID)
+ log.debug(s"Received hearbeat message from $instanceID.")
instanceManager.reportHeartBeat(instanceID, metricsReport)
} catch {
- case t: Throwable => log.error(t, "Could not report heart beat from {}.", sender().path)
+ case t: Throwable => log.error(s"Could not report heart beat from ${sender().path}.", t)
}
case message: AccumulatorMessage => handleAccumulatorMessage(message)
@@ -417,7 +412,7 @@ class JobManager(val flinkConfiguration: Configuration,
case Terminated(taskManager) =>
if (instanceManager.isRegistered(taskManager)) {
- log.info("Task manager {} terminated.", taskManager.path)
+ log.info(s"Task manager ${taskManager.path} terminated.")
instanceManager.unregisterTaskManager(taskManager)
context.unwatch(taskManager)
@@ -430,7 +425,7 @@ class JobManager(val flinkConfiguration: Configuration,
val taskManager = sender()
if (instanceManager.isRegistered(taskManager)) {
- log.info("Task manager {} wants to disconnect, because {}.", taskManager.path, msg)
+ log.info(s"Task manager ${taskManager.path} wants to disconnect, because $msg.")
instanceManager.unregisterTaskManager(taskManager)
context.unwatch(taskManager)
@@ -590,7 +585,7 @@ class JobManager(val flinkConfiguration: Configuration,
}
catch {
case tt: Throwable => {
- log.error(tt, "Error while marking ExecutionGraph as failed.")
+ log.error("Error while marking ExecutionGraph as failed.", tt)
}
}
}
@@ -618,7 +613,7 @@ class JobManager(val flinkConfiguration: Configuration,
libraryCacheManager.getClassLoader(jobId)
} catch {
case e: Exception =>
- log.error(e, "Dropping accumulators. No class loader available for job " + jobId)
+ log.error("Dropping accumulators. No class loader available for job " + jobId, e)
return
}
@@ -628,7 +623,7 @@ class JobManager(val flinkConfiguration: Configuration,
accumulatorManager.processIncomingAccumulators(jobId, accumulators)
}
catch {
- case e: Exception => log.error(e, "Cannot update accumulators for job " + jobId)
+ case e: Exception => log.error("Cannot update accumulators for job " + jobId, e)
}
} else {
log.error("Dropping accumulators. No class loader available for job " + jobId)
@@ -643,7 +638,7 @@ class JobManager(val flinkConfiguration: Configuration,
}
catch {
case e: Exception =>
- log.error(e, "Cannot serialize accumulator result")
+ log.error("Cannot serialize accumulator result", e)
sender() ! AccumulatorResultsErroneous(jobID, e)
}
@@ -656,7 +651,7 @@ class JobManager(val flinkConfiguration: Configuration,
}
catch {
case e: Exception =>
- log.error(e, "Cannot fetch accumulator result")
+ log.error("Cannot fetch accumulator result", e)
sender() ! AccumulatorResultsErroneous(jobId, e)
}
@@ -676,8 +671,8 @@ class JobManager(val flinkConfiguration: Configuration,
archive ! ArchiveExecutionGraph(jobID, eg)
} catch {
- case t: Throwable => log.error(t, "Could not prepare the execution graph {} for " +
- "archiving.", eg)
+ case t: Throwable => log.error(s"Could not prepare the execution graph $eg for " +
+ "archiving.", t)
}
case None =>
@@ -687,7 +682,7 @@ class JobManager(val flinkConfiguration: Configuration,
libraryCacheManager.unregisterJob(jobID)
} catch {
case t: Throwable =>
- log.error(t, "Could not properly unregister job {} form the library cache.", jobID)
+ log.error(s"Could not properly unregister job $jobID form the library cache.", t)
}
}
}
@@ -699,7 +694,7 @@ class JobManager(val flinkConfiguration: Configuration,
*/
object JobManager {
- val LOG = LoggerFactory.getLogger(classOf[JobManager])
+ val LOG = Logger(classOf[JobManager])
val STARTUP_FAILURE_RETURN_CODE = 1
val RUNTIME_FAILURE_RETURN_CODE = 2
@@ -717,7 +712,7 @@ object JobManager {
*/
def main(args: Array[String]): Unit = {
// startup checks and logging
- EnvironmentInformation.logEnvironmentInfo(LOG, "JobManager", args)
+ EnvironmentInformation.logEnvironmentInfo(LOG.logger, "JobManager", args)
EnvironmentInformation.checkJavaVersion()
// parsing the command line arguments
@@ -798,7 +793,7 @@ object JobManager {
LOG.info("Starting JobManager")
// Bring up the job manager actor system first, bind it to the given address.
- LOG.info("Starting JobManager actor system at {}:{}", listeningAddress, listeningPort)
+ LOG.info(s"Starting JobManager actor system at $listeningAddress:$listeningPort.")
val jobManagerSystem = try {
val akkaConfig = AkkaUtils.getAkkaConfig(configuration,
@@ -831,7 +826,7 @@ object JobManager {
// 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, 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
@@ -847,7 +842,7 @@ object JobManager {
LOG.debug("Starting TaskManager process reaper")
jobManagerSystem.actorOf(
- Props(classOf[ProcessReaper], taskManagerActor, LOG, RUNTIME_FAILURE_RETURN_CODE),
+ Props(classOf[ProcessReaper], taskManagerActor, LOG.logger, RUNTIME_FAILURE_RETURN_CODE),
"TaskManager_Process_Reaper")
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5a2ca819/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManagerProfiler.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManagerProfiler.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManagerProfiler.scala
index eb8f913..dd3a1b7 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManagerProfiler.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManagerProfiler.scala
@@ -18,18 +18,21 @@
package org.apache.flink.runtime.jobmanager
-import akka.actor.{ActorLogging, Actor}
-import org.apache.flink.runtime.ActorLogMessages
+import akka.actor.Actor
+import org.apache.flink.runtime.{ActorSynchronousLogging, ActorLogMessages}
import org.apache.flink.runtime.messages.JobManagerProfilerMessages.ReportProfilingData
import org.apache.flink.runtime.profiling.impl.types.{InternalInstanceProfilingData, InternalExecutionVertexThreadProfilingData}
-import org.apache.flink.runtime.profiling.types.ThreadProfilingEvent
import scala.collection.convert.WrapAsScala
/**
* Basic skeleton for the JobManager profiler. Currently, it simply logs the received messages.
*/
-class JobManagerProfiler extends Actor with ActorLogMessages with ActorLogging with WrapAsScala {
+class JobManagerProfiler
+ extends Actor
+ with ActorLogMessages
+ with ActorSynchronousLogging
+ with WrapAsScala {
override def receiveWithLogMessages: Receive = {
case ReportProfilingData(profilingContainer) =>
profilingContainer.getIterator foreach {
http://git-wip-us.apache.org/repos/asf/flink/blob/5a2ca819/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 cb7bfec..9e71ebb 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
@@ -18,9 +18,9 @@
package org.apache.flink.runtime.jobmanager
-import akka.actor.{ActorLogging, Actor}
+import akka.actor.Actor
import org.apache.flink.api.common.JobID
-import org.apache.flink.runtime.ActorLogMessages
+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._
@@ -48,8 +48,10 @@ import scala.ref.SoftReference
*
* @param max_entries Maximum number of stored Flink jobs
*/
-class MemoryArchivist(private val max_entries: Int) extends Actor with ActorLogMessages with
-ActorLogging {
+class MemoryArchivist(private val max_entries: Int)
+ extends Actor
+ with ActorLogMessages
+ with ActorSynchronousLogging {
/*
* 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.
http://git-wip-us.apache.org/repos/asf/flink/blob/5a2ca819/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/StreamCheckpointCoordinator.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/StreamCheckpointCoordinator.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/StreamCheckpointCoordinator.scala
index 48266e2..8bb1274 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/StreamCheckpointCoordinator.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/StreamCheckpointCoordinator.scala
@@ -21,7 +21,7 @@ package org.apache.flink.runtime.jobmanager
import java.lang.{Long => JLong}
import akka.actor._
-import org.apache.flink.runtime.ActorLogMessages
+import org.apache.flink.runtime.{ActorSynchronousLogging, ActorLogMessages}
import org.apache.flink.runtime.execution.ExecutionState
import org.apache.flink.runtime.executiongraph.{ExecutionGraph, ExecutionVertex}
import org.apache.flink.runtime.jobgraph.JobStatus._
@@ -69,7 +69,7 @@ class StreamCheckpointCoordinator(val executionGraph: ExecutionGraph,
val interval: FiniteDuration,
var curId: JLong,
var ackId: JLong)
-extends Actor with ActorLogMessages with ActorLogging {
+extends Actor with ActorLogMessages with ActorSynchronousLogging {
implicit private val executor = context.dispatcher
@@ -78,13 +78,13 @@ extends Actor with ActorLogMessages with ActorLogging {
case InitBarrierScheduler =>
context.system.scheduler.schedule(interval,interval,self,BarrierTimeout)
context.system.scheduler.schedule(2 * interval,2 * interval,self,CompactAndUpdate)
- log.info("Started Stream State Monitor for job {}{}",
- executionGraph.getJobID,executionGraph.getJobName)
+ log.info("Started Stream State Monitor for job " +
+ s"${executionGraph.getJobID}${executionGraph.getJobName}")
case BarrierTimeout =>
executionGraph.getState match {
case FAILED | CANCELED | FINISHED =>
- log.info("Stopping monitor for terminated job {}", executionGraph.getJobID)
+ log.info(s"Stopping monitor for terminated job ${executionGraph.getJobID}.")
self ! PoisonPill
case RUNNING =>
curId += 1
@@ -94,8 +94,8 @@ extends Actor with ActorLogMessages with ActorLogging {
=> vertex.getCurrentAssignedResource.getInstance.getTaskManager
! BarrierReq(vertex.getCurrentExecutionAttempt.getAttemptId,curId))
case _ =>
- log.debug("Omitting sending barrier since graph is in {} state for job {}",
- executionGraph.getState, executionGraph.getJobID)
+ log.debug("Omitting sending barrier since graph is in " +
+ s"${executionGraph.getState} state for job ${executionGraph.getJobID}.")
}
case StateBarrierAck(jobID, jobVertexID, instanceID, checkpointID, opState) =>
http://git-wip-us.apache.org/repos/asf/flink/blob/5a2ca819/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 d6b91ec..511de6b 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
@@ -33,11 +33,12 @@ import com.codahale.metrics.json.MetricsModule
import com.codahale.metrics.jvm.{MemoryUsageGaugeSet, GarbageCollectorMetricSet}
import com.fasterxml.jackson.databind.ObjectMapper
+import grizzled.slf4j.Logger
import org.apache.flink.api.common.cache.DistributedCache
import org.apache.flink.configuration._
import org.apache.flink.core.fs.Path
-import org.apache.flink.runtime.ActorLogMessages
+import org.apache.flink.runtime.{ActorSynchronousLogging, ActorLogMessages}
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.blob.{BlobService, BlobCache}
import org.apache.flink.runtime.broadcast.BroadcastVariableManager
@@ -67,8 +68,6 @@ import org.apache.flink.runtime.security.SecurityUtils.FlinkSecuredRunner
import org.apache.flink.runtime.util.{MathUtils, EnvironmentInformation}
import org.apache.flink.util.ExceptionUtils
-import org.slf4j.LoggerFactory
-
import scala.concurrent._
import scala.concurrent.duration._
import scala.util.{Failure, Success}
@@ -125,10 +124,7 @@ class TaskManager(protected val config: TaskManagerConfiguration,
protected val network: NetworkEnvironment,
protected val numberOfSlots: Int)
-extends Actor with ActorLogMessages with ActorLogging {
-
- /** The log for all synchronous logging calls */
- private val LOG = TaskManager.LOG
+extends Actor with ActorLogMessages with ActorSynchronousLogging {
/** The timeout for all actor ask futures */
protected val askTimeout = new Timeout(config.timeout)
@@ -178,13 +174,13 @@ extends Actor with ActorLogMessages with ActorLogging {
* JobManager.
*/
override def preStart(): Unit = {
- LOG.info("Starting TaskManager actor at {}.", self.path.toSerializationFormat)
- LOG.info("TaskManager data connection information: {}", connectionInfo)
- LOG.info("TaskManager has {} task slot(s).", numberOfSlots)
+ log.info(s"Starting TaskManager actor at ${self.path.toSerializationFormat}.")
+ log.info(s"TaskManager data connection information: $connectionInfo")
+ log.info(s"TaskManager has $numberOfSlots task slot(s).")
// log the initial memory utilization
- if (LOG.isInfoEnabled) {
- LOG.info(TaskManager.getMemoryUsageStatsAsString(ManagementFactory.getMemoryMXBean))
+ if (log.isInfoEnabled) {
+ log.info(TaskManager.getMemoryUsageStatsAsString(ManagementFactory.getMemoryMXBean))
}
// kick off the registration
@@ -200,7 +196,7 @@ extends Actor with ActorLogMessages with ActorLogging {
* (like network stack, library cache, memory manager, ...) are properly shut down.
*/
override def postStop(): Unit = {
- LOG.info("Stopping TaskManager {}.", self.path.toSerializationFormat)
+ log.info(s"Stopping TaskManager ${self.path.toSerializationFormat}.")
cancelAndClearEverything(new Exception("TaskManager is shutting down."))
@@ -208,35 +204,35 @@ extends Actor with ActorLogMessages with ActorLogging {
try {
disassociateFromJobManager()
} catch {
- case t: Exception => LOG.error("Could not cleanly disassociate from JobManager", t)
+ case t: Exception => log.error("Could not cleanly disassociate from JobManager", t)
}
}
try {
ioManager.shutdown()
} catch {
- case t: Exception => LOG.error("I/O manager did not shutdown properly.", t)
+ case t: Exception => log.error("I/O manager did not shutdown properly.", t)
}
try {
memoryManager.shutdown()
} catch {
- case t: Exception => LOG.error("Memory manager did not shutdown properly.", t)
+ case t: Exception => log.error("Memory manager did not shutdown properly.", t)
}
try {
network.shutdown()
} catch {
- case t: Exception => LOG.error("Network environment did not shutdown properly.", t)
+ case t: Exception => log.error("Network environment did not shutdown properly.", t)
}
try {
fileCache.shutdown()
} catch {
- case t: Exception => LOG.error("FileCache did not shutdown properly.", t)
+ case t: Exception => log.error("FileCache did not shutdown properly.", t)
}
- LOG.info("Task manager {} is completely shut down.", self.path)
+ log.info(s"Task manager ${self.path} is completely shut down.")
}
/**
@@ -277,8 +273,8 @@ extends Actor with ActorLogMessages with ActorLogging {
handleJobManagerDisconnect(sender(), "JobManager is no longer reachable")
}
else {
- LOG.warn("Received unrecognized disconnect message from {}",
- if (actor == null) null else actor.path)
+ log.warn(s"Received unrecognized disconnect message " +
+ s"from ${if (actor == null) null else actor.path}.")
}
case Disconnect(msg) =>
@@ -291,7 +287,7 @@ extends Actor with ActorLogMessages with ActorLogging {
override def unhandled(message: Any): Unit = {
val errorMessage = "Received unknown message " + message
val error = new RuntimeException(errorMessage)
- LOG.error(errorMessage)
+ log.error(errorMessage)
// terminate all we are currently running (with a dedicated message)
// before the actor is stopped
@@ -311,8 +307,8 @@ extends Actor with ActorLogMessages with ActorLogging {
// at very first, check that we are actually currently associated with a JobManager
if (!isConnected) {
- LOG.debug("Dropping message {} because the TaskManager is currently " +
- "not connected to a JobManager", message)
+ log.debug(s"Dropping message $message because the TaskManager is currently " +
+ "not connected to a JobManager.")
}
// we order the messages by frequency, to make sure the code paths for matching
@@ -329,7 +325,7 @@ extends Actor with ActorLogMessages with ActorLogging {
// discards intermediate result partitions of a task execution on this TaskManager
case FailIntermediateResultPartitions(executionID) =>
- LOG.info("Discarding the results produced by task execution " + executionID)
+ log.info("Discarding the results produced by task execution " + executionID)
if (network.isAssociated) {
try {
network.getPartitionManager.releasePartitionsProducedBy(executionID)
@@ -392,7 +388,7 @@ extends Actor with ActorLogMessages with ActorLogging {
Future {
task.failExternally(cause)
}.onFailure{
- case t: Throwable => LOG.error(s"Could not fail task ${task} externally.", t)
+ case t: Throwable => log.error(s"Could not fail task ${task} externally.", t)
}
case None =>
}
@@ -406,7 +402,7 @@ extends Actor with ActorLogMessages with ActorLogging {
Future {
task.cancelExecution()
}.onFailure{
- case t: Throwable => LOG.error("Could not cancel task " + task, t)
+ case t: Throwable => log.error("Could not cancel task " + task, t)
}
sender ! new TaskOperationResult(executionID, true)
@@ -426,10 +422,9 @@ extends Actor with ActorLogMessages with ActorLogging {
private def handleCheckpointingMessage(message: CheckpointingMessage): Unit = {
message match {
-
case BarrierReq(attemptID, checkpointID) =>
- LOG.debug("[FT-TaskManager] Barrier {} request received for attempt {}",
- checkpointID, attemptID)
+ log.debug(s"[FT-TaskManager] Barrier $checkpointID request received " +
+ s"for attempt $attemptID.")
runningTasks.get(attemptID) match {
case Some(i) =>
@@ -441,16 +436,14 @@ extends Actor with ActorLogMessages with ActorLogging {
barrierTransceiver.broadcastBarrierFromSource(checkpointID)
}).start()
- case _ => LOG.error(
- "Taskmanager received a checkpoint request for non-checkpointing task {}",
- attemptID)
+ case _ => log.error("Taskmanager received a checkpoint request for " +
+ s"non-checkpointing task $attemptID.")
}
}
case None =>
// may always happen in case of canceled/finished tasks
- LOG.debug("Taskmanager received a checkpoint request for unknown task {}",
- attemptID)
+ log.debug(s"Taskmanager received a checkpoint request for unknown task $attemptID.")
}
// unknown checkpoint message
@@ -474,19 +467,19 @@ extends Actor with ActorLogMessages with ActorLogging {
if (isConnected) {
// this may be the case, if we queue another attempt and
// in the meantime, the registration is acknowledged
- LOG.debug(
+ log.debug(
"TaskManager was triggered to register at JobManager, but is already registered")
}
else if (deadline.exists(_.isOverdue())) {
// we failed to register in time. that means we should quit
- LOG.error("Failed to register at the JobManager withing the defined maximum " +
+ log.error("Failed to register at the JobManager withing the defined maximum " +
"connect time. Shutting down ...")
// terminate ourselves (hasta la vista)
self ! PoisonPill
}
else {
- LOG.info(s"Trying to register at JobManager ${jobManagerURL} " +
+ log.info(s"Trying to register at JobManager ${jobManagerURL} " +
s"(attempt ${attempt}, timeout: ${timeout})")
val jobManager = context.actorSelection(jobManagerAkkaURL)
@@ -510,9 +503,9 @@ extends Actor with ActorLogMessages with ActorLogging {
case AcknowledgeRegistration(jobManager, id, blobPort) =>
if (isConnected) {
if (jobManager == currentJobManager.orNull) {
- LOG.debug("Ignoring duplicate registration acknowledgement.")
+ log.debug("Ignoring duplicate registration acknowledgement.")
} else {
- LOG.warn(s"Ignoring 'AcknowledgeRegistration' message from ${jobManager.path} , " +
+ log.warn(s"Ignoring 'AcknowledgeRegistration' message from ${jobManager.path} , " +
s"because the TaskManager is already registered at ${currentJobManager.orNull}")
}
}
@@ -531,15 +524,15 @@ extends Actor with ActorLogMessages with ActorLogging {
case AlreadyRegistered(jobManager, id, blobPort) =>
if (isConnected) {
if (jobManager == currentJobManager.orNull) {
- LOG.debug("Ignoring duplicate registration acknowledgement.")
+ log.debug("Ignoring duplicate registration acknowledgement.")
} else {
- LOG.warn(s"Received 'AlreadyRegistered' message from JobManager ${jobManager.path}, " +
+ log.warn(s"Received 'AlreadyRegistered' message from JobManager ${jobManager.path}, " +
s"even through TaskManager is currently registered at ${currentJobManager.orNull}")
}
}
else {
// not connected, yet, to let's associate
- LOG.info("Received 'AlreadyRegistered' message before 'AcknowledgeRegistration'")
+ log.info("Received 'AlreadyRegistered' message before 'AcknowledgeRegistration'")
try {
associateWithJobManager(jobManager, id, blobPort)
@@ -552,7 +545,7 @@ extends Actor with ActorLogMessages with ActorLogging {
case RefuseRegistration(reason) =>
if (currentJobManager.isEmpty) {
- LOG.error(s"The registration at JobManager ${jobManagerAkkaURL} was refused, " +
+ log.error(s"The registration at JobManager ${jobManagerAkkaURL} was refused, " +
s"because: ${reason}. Retrying later...")
// try the registration again after some time
@@ -570,11 +563,11 @@ extends Actor with ActorLogMessages with ActorLogging {
else {
// ignore RefuseRegistration messages which arrived after AcknowledgeRegistration
if (sender() == currentJobManager.orNull) {
- LOG.warn(s"Received 'RefuseRegistration' from the JobManager (${sender().path})" +
+ log.warn(s"Received 'RefuseRegistration' from the JobManager (${sender().path})" +
s" even though this TaskManager is already registered there.")
}
else {
- LOG.warn(s"Ignoring 'RefuseRegistration' from unrelated JobManager (${sender().path})")
+ log.warn(s"Ignoring 'RefuseRegistration' from unrelated JobManager (${sender().path})")
}
}
@@ -621,7 +614,7 @@ extends Actor with ActorLogMessages with ActorLogging {
// sanity check that we are not currently registered with a different JobManager
if (isConnected) {
if (currentJobManager.get == jobManager) {
- LOG.warn("Received call to finish registration with JobManager " +
+ log.warn("Received call to finish registration with JobManager " +
jobManager.path + " even though TaskManager is already registered.")
return
}
@@ -633,8 +626,8 @@ extends Actor with ActorLogMessages with ActorLogging {
}
// not yet associated, so associate
- LOG.info("Successful registration at JobManager ({}), " +
- "starting network stack and library cache.", jobManager.path)
+ log.info(s"Successful registration at JobManager (${jobManager.path}), " +
+ "starting network stack and library cache.")
// sanity check that the JobManager dependent components are not set up currently
if (network.isAssociated || blobService.isDefined) {
@@ -648,7 +641,7 @@ extends Actor with ActorLogMessages with ActorLogging {
catch {
case e: Exception =>
val message = "Could not start network environment."
- LOG.error(message, e)
+ log.error(message, e)
throw new RuntimeException(message, e)
}
@@ -657,7 +650,7 @@ extends Actor with ActorLogMessages with ActorLogging {
val jmHost = jobManager.path.address.host.getOrElse("localhost")
val address = new InetSocketAddress(jmHost, blobPort)
- LOG.info("Determined BLOB server address to be {}. Starting BLOB cache.", address)
+ log.info(s"Determined BLOB server address to be $address. Starting BLOB cache.")
try {
val blobcache = new BlobCache(address, config.configuration)
@@ -667,7 +660,7 @@ extends Actor with ActorLogMessages with ActorLogging {
catch {
case e: Exception =>
val message = "Could not create BLOB cache or library cache."
- LOG.error(message, e)
+ log.error(message, e)
throw new RuntimeException(message, e)
}
}
@@ -700,12 +693,12 @@ extends Actor with ActorLogMessages with ActorLogging {
*/
private def disassociateFromJobManager(): Unit = {
if (!isConnected) {
- LOG.warn("TaskManager received message to disassociate from JobManager, even though " +
+ log.warn("TaskManager received message to disassociate from JobManager, even though " +
"it is not currently associated with a JobManager")
return
}
- LOG.info("Disassociating from JobManager")
+ log.info("Disassociating from JobManager")
// stop the periodic heartbeats
heartbeatScheduler foreach {
@@ -748,7 +741,7 @@ extends Actor with ActorLogMessages with ActorLogging {
if (jobManager == currentJobManager.orNull) {
try {
val message = "Disconnecting from JobManager: " + msg
- LOG.info(message)
+ log.info(message)
// cancel all our tasks with a proper error message
cancelAndClearEverything(new Exception(message))
@@ -769,7 +762,7 @@ extends Actor with ActorLogMessages with ActorLogging {
}
}
else {
- LOG.warn("Received erroneous JobManager disconnect message for {}", jobManager.path)
+ log.warn(s"Received erroneous JobManager disconnect message for ${jobManager.path}.")
}
}
}
@@ -807,17 +800,15 @@ extends Actor with ActorLogMessages with ActorLogging {
val userCodeClassLoader = libraryCacheManager match {
case Some(manager) =>
- if (LOG.isDebugEnabled) {
+ if (log.isDebugEnabled) {
startRegisteringTask = System.currentTimeMillis()
}
// triggers the download of all missing jar files from the job manager
manager.registerTask(jobID, executionID, tdd.getRequiredJarFiles)
- if (LOG.isDebugEnabled) {
- LOG.debug("Register task {} at library cache manager took {}s", executionID,
- (System.currentTimeMillis() - startRegisteringTask) / 1000.0)
- }
+ log.debug(s"Register task $executionID at library cache manager " +
+ s"took ${(System.currentTimeMillis() - startRegisteringTask) / 1000.0}s")
manager.getClassLoader(jobID)
case None => throw new IllegalStateException("There is no valid library cache manager.")
@@ -859,7 +850,7 @@ extends Actor with ActorLogMessages with ActorLogging {
}
// register the task with the network stack and profiles
- LOG.info("Register task {}", task)
+ log.info(s"Register task $task.")
network.registerTask(task)
val cpTasks = new util.HashMap[String, FutureTask[Path]]()
@@ -881,7 +872,7 @@ extends Actor with ActorLogMessages with ActorLogging {
val message = if (t.isInstanceOf[CancelTaskException]) {
"Task was canceled"
} else {
- LOG.error("Could not instantiate task with execution ID " + executionID, t)
+ log.error("Could not instantiate task with execution ID " + executionID, t)
ExceptionUtils.stringifyException(t)
}
@@ -893,7 +884,7 @@ extends Actor with ActorLogMessages with ActorLogging {
libraryCacheManager foreach { _.unregisterTask(jobID, executionID) }
} catch {
- case t: Throwable => LOG.error("Error during cleanup of task deployment.", t)
+ case t: Throwable => log.error("Error during cleanup of task deployment.", t)
}
sender ! new TaskOperationResult(executionID, false, message)
@@ -924,7 +915,7 @@ extends Actor with ActorLogMessages with ActorLogging {
reader.updateInputChannel(partitionInfo)
} catch {
case t: Throwable =>
- LOG.error(s"Could not update input data location for task " +
+ log.error(s"Could not update input data location for task " +
s"${task.getTaskName}. Trying to fail task.", t)
try {
@@ -932,7 +923,7 @@ extends Actor with ActorLogMessages with ActorLogging {
}
catch {
case t: Throwable =>
- LOG.error("Failed canceling task with execution ID " + executionId +
+ log.error("Failed canceling task with execution ID " + executionId +
" after task update failure.", t)
}
}
@@ -951,8 +942,8 @@ extends Actor with ActorLogMessages with ActorLogging {
}
case None =>
- LOG.debug("Discard update for input partitions of task {} : task is no longer running.",
- executionId)
+ log.debug(s"Discard update for input partitions of task $executionId : " +
+ s"task is no longer running.")
sender ! Acknowledge
}
}
@@ -965,7 +956,7 @@ extends Actor with ActorLogMessages with ActorLogging {
*/
private def cancelAndClearEverything(cause: Throwable) {
if (runningTasks.size > 0) {
- LOG.info("Cancelling all computations and discarding all cached data.")
+ log.info("Cancelling all computations and discarding all cached data.")
for (t <- runningTasks.values) {
t.failExternally(cause)
@@ -983,22 +974,22 @@ extends Actor with ActorLogMessages with ActorLogging {
try {
task.failExternally(new Exception("Task is being removed from TaskManager"))
} catch {
- case e: Exception => LOG.error("Could not properly fail task", e)
+ case e: Exception => log.error("Could not properly fail task", e)
}
}
- LOG.info("Unregister task with execution ID {}.", executionID)
+ log.info(s"Unregister task with execution ID $executionID.")
removeAllTaskResources(task)
libraryCacheManager foreach { _.unregisterTask(task.getJobID, executionID) }
- LOG.info("Updating FINAL execution state of {} ({}) to {}.",
- task.getTaskName, task.getExecutionId, task.getExecutionState)
+ log.info(s"Updating FINAL execution state of ${task.getTaskName} " +
+ s"(${task.getExecutionId}) to ${task.getExecutionState}.")
self ! UpdateTaskExecutionState(new TaskExecutionState(
task.getJobID, task.getExecutionId, task.getExecutionState, task.getFailureCause))
case None =>
- LOG.debug("Cannot find task with ID {} to unregister.", executionID)
+ log.debug(s"Cannot find task with ID $executionID to unregister.")
}
}
@@ -1044,7 +1035,7 @@ extends Actor with ActorLogMessages with ActorLogging {
}
} catch {
// this is pretty unpleasant, but not a reason to give up immediately
- case e: Exception => LOG.error(
+ case e: Exception => log.error(
"Error cleaning up local temp files from the distributed cache.", e)
}
}
@@ -1060,14 +1051,14 @@ extends Actor with ActorLogMessages with ActorLogging {
*/
private def sendHeartbeatToJobManager(): Unit = {
try {
- LOG.debug("Sending heartbeat to JobManager")
+ log.debug("Sending heartbeat to JobManager")
val report: Array[Byte] = metricRegistryMapper.writeValueAsBytes(metricRegistry)
currentJobManager foreach {
jm => jm ! Heartbeat(instanceID, report)
}
}
catch {
- case e: Exception => LOG.warn("Error sending the metric heartbeat to the JobManager", e)
+ case e: Exception => log.warn("Error sending the metric heartbeat to the JobManager", e)
}
}
@@ -1093,7 +1084,7 @@ extends Actor with ActorLogMessages with ActorLogging {
recipient ! StackTrace(instanceID, stackTraceStr)
}
catch {
- case e: Exception => LOG.error("Failed to send stack trace to " + recipient.path, e)
+ case e: Exception => log.error("Failed to send stack trace to " + recipient.path, e)
}
}
@@ -1103,7 +1094,7 @@ extends Actor with ActorLogMessages with ActorLogging {
* @param cause The exception that caused the fatal problem.
*/
private def killTaskManagerFatal(message: String, cause: Throwable): Unit = {
- LOG.error("\n" +
+ log.error("\n" +
"==============================================================\n" +
"====================== FATAL =======================\n" +
"==============================================================\n" +
@@ -1121,7 +1112,7 @@ extends Actor with ActorLogMessages with ActorLogging {
object TaskManager {
/** TaskManager logger for synchronous logging (not through the logging actor) */
- val LOG = LoggerFactory.getLogger(classOf[TaskManager])
+ val LOG = Logger(classOf[TaskManager])
/** Return code for unsuccessful TaskManager startup */
val STARTUP_FAILURE_RETURN_CODE = 1
@@ -1159,7 +1150,7 @@ object TaskManager {
*/
def main(args: Array[String]): Unit = {
// startup checks and logging
- EnvironmentInformation.logEnvironmentInfo(LOG, "TaskManager", args)
+ EnvironmentInformation.logEnvironmentInfo(LOG.logger, "TaskManager", args)
EnvironmentInformation.checkJavaVersion()
// try to parse the command line arguments
@@ -1283,8 +1274,8 @@ object TaskManager {
LOG.info("Trying to select the network interface and address to use " +
"by connecting to the configured JobManager.")
- LOG.info("TaskManager will try to connect for {} seconds before falling back to heuristics",
- MAX_STARTUP_CONNECT_TIME)
+ LOG.info(s"TaskManager will try to connect for $MAX_STARTUP_CONNECT_TIME seconds before " +
+ "falling back to heuristics")
val jobManagerAddress = new InetSocketAddress(jobManagerHostname, jobManagerPort)
val taskManagerAddress = try {
@@ -1361,7 +1352,7 @@ object TaskManager {
// Bring up the TaskManager actor system first, bind it to the given address.
- LOG.info("Starting TaskManager actor system at {}:{}", taskManagerHostname, actorSystemPort)
+ LOG.info(s"Starting TaskManager actor system at $taskManagerHostname:$actorSystemPort")
val taskManagerSystem = try {
val akkaConfig = AkkaUtils.getAkkaConfig(configuration,
@@ -1400,7 +1391,7 @@ object TaskManager {
// the process reaper will kill the JVM process (to ensure easy failure detection)
LOG.debug("Starting TaskManager process reaper")
taskManagerSystem.actorOf(
- Props(classOf[ProcessReaper], taskManager, LOG, RUNTIME_FAILURE_RETURN_CODE),
+ Props(classOf[ProcessReaper], taskManager, LOG.logger, RUNTIME_FAILURE_RETURN_CODE),
"TaskManager_Process_Reaper")
// if desired, start the logging daemon that periodically logs the
@@ -1521,7 +1512,7 @@ object TaskManager {
"pick a fraction of the available memory.")
val memorySize = if (configuredMemory > 0) {
- LOG.info("Using {} MB for Flink managed memory.", configuredMemory)
+ LOG.info(s"Using $configuredMemory MB for Flink managed memory.")
configuredMemory << 20 // megabytes to bytes
}
else {
@@ -1534,8 +1525,8 @@ object TaskManager {
val relativeMemSize = (EnvironmentInformation.getSizeOfFreeHeapMemoryWithDefrag() *
fraction).toLong
- LOG.info("Using {} of the currently free heap space for Flink managed memory ({} MB).",
- fraction, relativeMemSize >> 20)
+ LOG.info(s"Using $fraction of the currently free heap space for Flink managed " +
+ s"memory (${relativeMemSize >> 20} MB).")
relativeMemSize
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5a2ca819/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManagerProfiler.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManagerProfiler.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManagerProfiler.scala
index 51e99f9..f0079f8 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManagerProfiler.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManagerProfiler.scala
@@ -22,7 +22,7 @@ import java.lang.management.ManagementFactory
import java.util.concurrent.TimeUnit
import akka.actor.{Cancellable, ActorRef, Actor, ActorLogging}
-import org.apache.flink.runtime.ActorLogMessages
+import org.apache.flink.runtime.{ActorSynchronousLogging, ActorLogMessages}
import org.apache.flink.runtime.execution.{RuntimeEnvironment, ExecutionState}
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID
import org.apache.flink.runtime.jobgraph.JobVertexID
@@ -44,7 +44,7 @@ import scala.concurrent.duration.FiniteDuration
* @param reportInterval Interval of profiling action
*/
class TaskManagerProfiler(val instancePath: String, val reportInterval: Int)
- extends Actor with ActorLogMessages with ActorLogging {
+ extends Actor with ActorLogMessages with ActorSynchronousLogging {
import context.dispatcher
@@ -101,7 +101,7 @@ class TaskManagerProfiler(val instancePath: String, val reportInterval: Int)
Some(instanceProfiler.generateProfilingData(timestamp))
} catch {
case e: ProfilingException =>
- log.error(e, "Error while retrieving instance profiling data.")
+ log.error("Error while retrieving instance profiling data.", e)
None
}
@@ -133,7 +133,7 @@ class TaskManagerProfiler(val instancePath: String, val reportInterval: Int)
case _ =>
}
case None =>
- log.warning(s"Could not find environment for execution id $executionID.")
+ log.warn(s"Could not find environment for execution id $executionID.")
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5a2ca819/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala
index b400bb9..d6760ec 100644
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala
+++ b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala
@@ -23,7 +23,7 @@ import java.net.InetSocketAddress
import akka.actor._
import akka.pattern.ask
import org.apache.flink.configuration.Configuration
-import org.apache.flink.runtime.ActorLogMessages
+import org.apache.flink.runtime.{ActorSynchronousLogging, ActorLogMessages}
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.jobmanager.JobManager
import org.apache.flink.runtime.yarn.FlinkYarnClusterStatus
@@ -34,8 +34,10 @@ import scala.concurrent.duration._
import scala.language.postfixOps
import scala.util.{Failure, Success}
-class ApplicationClient(flinkConfig: Configuration) extends Actor
- with ActorLogMessages with ActorLogging {
+class ApplicationClient(flinkConfig: Configuration)
+ extends Actor
+ with ActorLogMessages
+ with ActorSynchronousLogging {
import context._
val INITIAL_POLLING_DELAY = 0 seconds
@@ -78,8 +80,8 @@ class ApplicationClient(flinkConfig: Configuration) extends Actor
jobManagerFuture.onComplete {
case Success(jm) => self ! JobManagerActorRef(jm)
case Failure(t) =>
- log.error(t, "Registration at JobManager/ApplicationMaster failed. Shutting " +
- "ApplicationClient down.")
+ log.error("Registration at JobManager/ApplicationMaster failed. Shutting " +
+ "ApplicationClient down.", t)
// we could not connect to the job manager --> poison ourselves
self ! PoisonPill
@@ -93,7 +95,7 @@ class ApplicationClient(flinkConfig: Configuration) extends Actor
// sender as the Application Client (this class).
(jm ? RegisterClient(self))(timeout).onFailure{
case t: Throwable =>
- log.error(t, "Could not register at the job manager.")
+ log.error("Could not register at the job manager.", t)
self ! PoisonPill
}
@@ -144,7 +146,7 @@ class ApplicationClient(flinkConfig: Configuration) extends Actor
// ----------------- handle messages from the cluster -------------------
// receive remote messages
case msg: YarnMessage =>
- log.debug("Received new YarnMessage {}. Now {} messages in queue", msg, messagesQueue.size)
+ log.debug(s"Received new YarnMessage $msg. Now ${messagesQueue.size} messages in queue")
messagesQueue.enqueue(msg)
// locally forward messages
http://git-wip-us.apache.org/repos/asf/flink/blob/5a2ca819/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMaster.scala
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMaster.scala b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMaster.scala
index 0fdef47..06e16dc 100644
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMaster.scala
+++ b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMaster.scala
@@ -21,6 +21,7 @@ import java.io.{PrintWriter, FileWriter, BufferedWriter}
import java.security.PrivilegedAction
import akka.actor._
+import grizzled.slf4j.Logger
import org.apache.flink.client.CliFrontend
import org.apache.flink.configuration.{GlobalConfiguration, Configuration, ConfigConstants}
import org.apache.flink.runtime.akka.AkkaUtils
@@ -38,7 +39,7 @@ import scala.io.Source
object ApplicationMaster {
import scala.collection.JavaConversions._
- val LOG = LoggerFactory.getLogger(this.getClass)
+ val LOG = Logger(getClass)
val CONF_FILE = "flink-conf.yaml"
val MODIFIED_CONF_FILE = "flink-conf-modified.yaml"
@@ -50,9 +51,9 @@ object ApplicationMaster {
LOG.info(s"YARN daemon runs as ${UserGroupInformation.getCurrentUser.getShortUserName} " +
s"setting user to execute Flink ApplicationMaster/JobManager to ${yarnClientUsername}")
- EnvironmentInformation.logEnvironmentInfo(LOG, "YARN ApplicationMaster/JobManager", args)
+ EnvironmentInformation.logEnvironmentInfo(LOG.logger, "YARN ApplicationMaster/JobManager", args)
EnvironmentInformation.checkJavaVersion()
- org.apache.flink.runtime.util.SignalHandler.register(LOG)
+ org.apache.flink.runtime.util.SignalHandler.register(LOG.logger)
val ugi = UserGroupInformation.createRemoteUser(yarnClientUsername)
@@ -206,7 +207,7 @@ object ApplicationMaster {
(Configuration, ActorSystem, ActorRef, ActorRef) = {
LOG.info("Starting JobManager for YARN")
- LOG.info("Loading config from: {}", currDir)
+ LOG.info(s"Loading config from: $currDir.")
GlobalConfiguration.loadConfiguration(currDir)
val configuration = GlobalConfiguration.getConfiguration()
http://git-wip-us.apache.org/repos/asf/flink/blob/5a2ca819/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMasterActor.scala
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMasterActor.scala b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMasterActor.scala
index f384130..26d1f69 100644
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMasterActor.scala
+++ b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMasterActor.scala
@@ -98,7 +98,7 @@ trait ApplicationMasterActor extends ActorLogMessages {
def receiveYarnMessages: Receive = {
case StopYarnSession(status, diag) =>
- log.info("Stopping YARN JobManager with status {} and diagnostic {}.", status, diag)
+ log.info(s"Stopping YARN JobManager with status $status and diagnostic $diag.")
instanceManager.getAllRegisteredInstances.asScala foreach {
instance =>
@@ -108,11 +108,11 @@ trait ApplicationMasterActor extends ActorLogMessages {
rmClientOption foreach {
rmClient =>
Try(rmClient.unregisterApplicationMaster(status, diag, "")).recover{
- case t: Throwable => log.error(t, "Could not unregister the application master.")
+ case t: Throwable => log.error("Could not unregister the application master.", t)
}
Try(rmClient.close()).recover{
- case t:Throwable => log.error(t, "Could not close the AMRMClient.")
+ case t:Throwable => log.error("Could not close the AMRMClient.", t)
}
}
@@ -121,7 +121,7 @@ trait ApplicationMasterActor extends ActorLogMessages {
nmClientOption foreach {
nmClient =>
Try(nmClient.close()).recover{
- case t: Throwable => log.error(t, "Could not close the NMClient.")
+ case t: Throwable => log.error("Could not close the NMClient.", t)
}
}
@@ -133,7 +133,7 @@ trait ApplicationMasterActor extends ActorLogMessages {
context.system.shutdown()
case RegisterClient(client) =>
- log.info("Register {} as client.", client.path)
+ log.info(s"Register ${client.path} as client.")
messageListener = Some(client)
sender ! Acknowledge
@@ -142,7 +142,7 @@ trait ApplicationMasterActor extends ActorLogMessages {
case msg: StopAMAfterJob =>
val jobId = msg.jobId
- log.info("ApplicatonMaster will shut down YARN session when job {} has finished", jobId)
+ log.info(s"ApplicatonMaster will shut down YARN session when job $jobId has finished.")
stopWhenJobFinished = jobId
sender() ! Acknowledge
@@ -155,26 +155,26 @@ trait ApplicationMasterActor extends ActorLogMessages {
startYarnSession(conf, actorSystemPort, webServerPort)
case jnf: JobNotFound =>
- LOG.warn("Job with ID {} not found in JobManager", jnf.jobID)
+ log.warn(s"Job with ID ${jnf.jobID} not found in JobManager")
if(stopWhenJobFinished == null) {
- LOG.warn("The ApplicationMaster didn't expect to receive this message")
+ log.warn("The ApplicationMaster didn't expect to receive this message")
}
case jobStatus: CurrentJobStatus =>
if(stopWhenJobFinished == null) {
- LOG.warn("Received job status {} which wasn't requested", jobStatus)
+ log.warn(s"Received job status $jobStatus which wasn't requested.")
} else {
if(stopWhenJobFinished != jobStatus.jobID) {
- LOG.warn(s"Received job status for job ${jobStatus.jobID} but expected status for " +
+ log.warn(s"Received job status for job ${jobStatus.jobID} but expected status for " +
s"job $stopWhenJobFinished")
} else {
if(jobStatus.status.isTerminalState) {
- LOG.info(s"Job with ID ${jobStatus.jobID} is in terminal state ${jobStatus.status}. " +
+ log.info(s"Job with ID ${jobStatus.jobID} is in terminal state ${jobStatus.status}. " +
s"Shutting down YARN session")
self ! StopYarnSession(FinalApplicationStatus.SUCCEEDED,
s"The monitored job with ID ${jobStatus.jobID} has finished.")
} else {
- LOG.debug(s"Monitored job with ID ${jobStatus.jobID} is in state ${jobStatus.status}")
+ log.debug(s"Monitored job with ID ${jobStatus.jobID} is in state ${jobStatus.status}")
}
}
}
@@ -229,14 +229,14 @@ trait ApplicationMasterActor extends ActorLogMessages {
})
}
// return containers if the RM wants them and we haven't allocated them yet.
- val preemtionMessage = response.getPreemptionMessage
- if(preemtionMessage != null) {
- log.info("Received preemtion message from YARN {}", preemtionMessage)
- val contract = preemtionMessage.getContract
+ val preemptionMessage = response.getPreemptionMessage
+ if(preemptionMessage != null) {
+ log.info(s"Received preemtion message from YARN $preemptionMessage.")
+ val contract = preemptionMessage.getContract
if(contract != null) {
tryToReturnContainers(contract.getContainers.asScala)
}
- val strictContract = preemtionMessage.getStrictContract
+ val strictContract = preemptionMessage.getStrictContract
if(strictContract != null) {
tryToReturnContainers(strictContract.getContainers.asScala)
}
@@ -247,13 +247,13 @@ trait ApplicationMasterActor extends ActorLogMessages {
// check if we want to start some of our allocated containers.
if(runningContainers < numTaskManager) {
var missingContainers = numTaskManager - runningContainers
- log.info("The user requested {} containers, {} running. {} containers missing",
- numTaskManager, runningContainers, missingContainers)
+ log.info(s"The user requested $numTaskManager containers, $runningContainers " +
+ s"running. $missingContainers containers missing")
// not enough containers running
if(allocatedContainersList.size > 0) {
- log.info("{} containers already allocated by YARN. Starting...",
- allocatedContainersList.size)
+ log.info(s"${allocatedContainersList.size} containers already allocated by YARN. " +
+ "Starting...")
// we have some containers allocated to us --> start them
allocatedContainersList = allocatedContainersList.dropWhile(container => {
if (missingContainers <= 0) {
@@ -279,7 +279,7 @@ trait ApplicationMasterActor extends ActorLogMessages {
}
} catch {
case e: YarnException =>
- log.error(e, "Exception while starting YARN container")
+ log.error("Exception while starting YARN container", e)
}
}
case None =>
@@ -305,24 +305,24 @@ trait ApplicationMasterActor extends ActorLogMessages {
log.info(s"There are $missingContainers containers missing." +
s" $numPendingRequests are already requested. " +
s"Requesting $toAllocateFromYarn additional container(s) from YARN. " +
- s"Reallocation of failed containers is enabled=$reallocate ('{}')",
- ConfigConstants.YARN_REALLOCATE_FAILED_CONTAINERS)
+ s"Reallocation of failed containers is enabled=$reallocate " +
+ s"('${ConfigConstants.YARN_REALLOCATE_FAILED_CONTAINERS}')")
// there are still containers missing. Request them from YARN
if(reallocate) {
for(i <- 1 to toAllocateFromYarn) {
val containerRequest = getContainerRequest(memoryPerTaskManager)
rmClient.addContainerRequest(containerRequest)
numPendingRequests += 1
- log.info("Requested additional container from YARN. Pending requests {}",
- numPendingRequests)
+ log.info("Requested additional container from YARN. Pending requests " +
+ s"$numPendingRequests.")
}
}
}
}
if(runningContainers >= numTaskManager && allocatedContainersList.size > 0) {
- log.info("Flink has {} allocated containers which are not needed right now. " +
- "Returning them", allocatedContainersList.size)
+ log.info(s"Flink has ${allocatedContainersList.size} allocated containers which " +
+ s"are not needed right now. Returning them")
for(container <- allocatedContainersList) {
rmClient.releaseAssignedContainer(container.getId)
}
@@ -353,9 +353,9 @@ trait ApplicationMasterActor extends ActorLogMessages {
self ! StopYarnSession(FinalApplicationStatus.FAILED, "Fatal error in AM: AMRMClient " +
"was not set")
}
- log.debug("Processed Heartbeat with RMClient. Running containers {}," +
- "failed containers {}, allocated containers {}", runningContainers, failedContainers,
- allocatedContainersList.size)
+ log.debug(s"Processed Heartbeat with RMClient. Running containers $runningContainers, " +
+ s"failed containers $failedContainers, " +
+ s"allocated containers ${allocatedContainersList.size}.")
}
private def runningContainerIds(): mutable.MutableList[ContainerId] = {
@@ -381,17 +381,16 @@ trait ApplicationMasterActor extends ActorLogMessages {
YarnConfiguration.DEFAULT_RM_AM_EXPIRY_INTERVAL_MS), MILLISECONDS)
if(YARN_HEARTBEAT_DELAY.gteq(yarnExpiryInterval)) {
- log.warning("The heartbeat interval of the Flink Application master ({}) is greater than " +
- "YARN's expiry interval ({}). The application is likely to be killed by YARN.",
- YARN_HEARTBEAT_DELAY, yarnExpiryInterval)
+ log.warn(s"The heartbeat interval of the Flink Application master " +
+ s"($YARN_HEARTBEAT_DELAY) is greater than YARN's expiry interval " +
+ s"($yarnExpiryInterval). The application is likely to be killed by YARN.")
}
numTaskManager = env.get(FlinkYarnClient.ENV_TM_COUNT).toInt
maxFailedContainers = flinkConfiguration.
getInteger(ConfigConstants.YARN_MAX_FAILED_CONTAINERS, numTaskManager)
- log.info("Requesting {} TaskManagers. Tolerating {} failed TaskManagers",
- numTaskManager, maxFailedContainers)
-
+ log.info(s"Requesting $numTaskManager TaskManagers. Tolerating $maxFailedContainers failed " +
+ "TaskManagers")
val remoteFlinkJarPath = env.get(FlinkYarnClient.FLINK_JAR_PATH)
val fs = FileSystem.get(conf)
@@ -468,7 +467,7 @@ trait ApplicationMasterActor extends ActorLogMessages {
context.system.scheduler.scheduleOnce(FAST_YARN_HEARTBEAT_DELAY, self, HeartbeatWithYarn)
} recover {
case t: Throwable =>
- log.error(t, "Could not start yarn session.")
+ log.error("Could not start yarn session.", t)
self ! StopYarnSession(FinalApplicationStatus.FAILED,
s"ApplicationMaster failed while starting. Exception Message: ${t.getMessage}")
}
@@ -479,7 +478,7 @@ trait ApplicationMasterActor extends ActorLogMessages {
allocatedContainersList = allocatedContainersList.dropWhile( container => {
val result = requestedBackContainers.getId.equals(container.getId)
if(result) {
- log.info("Returning container {} back to ResourceManager.", container)
+ log.info(s"Returning container $container back to ResourceManager.")
}
result
})
@@ -548,7 +547,7 @@ trait ApplicationMasterActor extends ActorLogMessages {
ctx.setTokens(securityTokens)
} catch {
case t: Throwable =>
- log.error(t, "Getting current user info failed when trying to launch the container")
+ log.error("Getting current user info failed when trying to launch the container", t)
}
ctx
http://git-wip-us.apache.org/repos/asf/flink/blob/5a2ca819/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 524e45f..1278386 100644
--- a/pom.xml
+++ b/pom.xml
@@ -275,6 +275,12 @@ under the License.
</dependency>
<dependency>
+ <groupId>org.clapper</groupId>
+ <artifactId>grizzled-slf4j_${scala.binary.version}</artifactId>
+ <version>1.0.2</version>
+ </dependency>
+
+ <dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-actor_${scala.binary.version}</artifactId>
<version>${akka.version}</version>