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/07/13 18:54:04 UTC
[1/7] flink git commit: [FLINK-2008] [FLINK-2296] Fix checkpoint
committing & KafkaITCase
Repository: flink
Updated Branches:
refs/heads/master 8d1efa045 -> 84b39dcb5
[FLINK-2008] [FLINK-2296] Fix checkpoint committing & KafkaITCase
This closes #895
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/aa5e5b30
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/aa5e5b30
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/aa5e5b30
Branch: refs/heads/master
Commit: aa5e5b3087a68f2aac792c0b0fc64b4f9c707e9b
Parents: 8d1efa0
Author: Robert Metzger <rm...@apache.org>
Authored: Mon Jun 29 16:52:38 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Jul 13 17:54:30 2015 +0200
----------------------------------------------------------------------
docs/apis/streaming_guide.md | 3 +
.../checkpoint/CheckpointCoordinator.java | 11 +-
.../checkpoint/SuccessfulCheckpoint.java | 22 +--
.../runtime/executiongraph/ExecutionVertex.java | 4 +
.../tasks/CheckpointCommittingOperator.java | 27 ----
.../tasks/CheckpointNotificationOperator.java | 25 ++++
.../messages/checkpoint/ConfirmCheckpoint.java | 88 ------------
.../checkpoint/NotifyCheckpointComplete.java | 73 ++++++++++
.../apache/flink/runtime/taskmanager/Task.java | 20 +--
.../flink/runtime/taskmanager/TaskManager.scala | 7 +-
.../checkpoint/CheckpointCoordinatorTest.java | 18 +--
.../messages/CheckpointMessagesTest.java | 4 +-
.../runtime/taskmanager/TaskAsyncCallTest.java | 12 +-
.../flink/streaming/connectors/kafka/Utils.java | 30 +++-
.../api/persistent/PersistentKafkaSource.java | 108 ++++++++++----
.../streaming/connectors/kafka/KafkaITCase.java | 144 +++++++++++++++----
.../connectors/kafka/util/UtilsTest.java | 75 ++++++++++
.../api/checkpoint/CheckpointCommitter.java | 44 ------
.../api/checkpoint/CheckpointNotifier.java | 37 +++++
.../api/graph/StreamingJobGraphGenerator.java | 5 +-
.../operators/AbstractUdfStreamOperator.java | 9 +-
.../api/operators/StatefulStreamOperator.java | 2 +-
.../streaming/runtime/tasks/StreamTask.java | 80 ++++-------
.../src/test/resources/log4j-test.properties | 2 +-
.../StreamCheckpointingITCase.java | 14 +-
.../org/apache/flink/yarn/YarnTestBase.java | 6 +-
26 files changed, 525 insertions(+), 345 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/docs/apis/streaming_guide.md
----------------------------------------------------------------------
diff --git a/docs/apis/streaming_guide.md b/docs/apis/streaming_guide.md
index e337ea8..02da8cb 100644
--- a/docs/apis/streaming_guide.md
+++ b/docs/apis/streaming_guide.md
@@ -1319,6 +1319,9 @@ Another way of exposing user defined operator state for the Flink runtime for ch
When the user defined function implements the `Checkpointed` interface, the `snapshotState(…)` and `restoreState(…)` methods will be executed to draw and restore function state.
+In addition to that, user functions can also implement the `CheckpointNotifier` interface to receive notifications on completed checkpoints via the `notifyCheckpointComplete(long checkpointId)` method.
+Note that there is no guarantee for the user function to receive a notification if a failure happens between checkpoint completion and notification. The notifications should hence be treated in a way that notifications from later checkpoints can subsume missing notifications.
+
For example the same counting, reduce function shown for `OperatorState`s by using the `Checkpointed` interface instead:
{% highlight java %}
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
index 91fd424..2b2bf6b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
@@ -30,10 +30,8 @@ import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
import org.apache.flink.runtime.executiongraph.ExecutionVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint;
-import org.apache.flink.runtime.messages.checkpoint.ConfirmCheckpoint;
+import org.apache.flink.runtime.messages.checkpoint.NotifyCheckpointComplete;
import org.apache.flink.runtime.messages.checkpoint.TriggerCheckpoint;
-import org.apache.flink.runtime.state.StateHandle;
-import org.apache.flink.runtime.util.SerializedValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -375,11 +373,8 @@ public class CheckpointCoordinator {
Execution ee = ev.getCurrentExecutionAttempt();
if (ee != null) {
ExecutionAttemptID attemptId = ee.getAttemptId();
- StateForTask stateForTask = completed.getState(ev.getJobvertexId());
- SerializedValue<StateHandle<?>> taskState = (stateForTask != null) ? stateForTask.getState() : null;
- ConfirmCheckpoint confirmMessage = new ConfirmCheckpoint(job, attemptId, checkpointId,
- timestamp, taskState);
- ev.sendMessageToCurrentExecution(confirmMessage, ee.getAttemptId());
+ NotifyCheckpointComplete notifyMessage = new NotifyCheckpointComplete(job, attemptId, checkpointId, timestamp);
+ ev.sendMessageToCurrentExecution(notifyMessage, ee.getAttemptId());
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SuccessfulCheckpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SuccessfulCheckpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SuccessfulCheckpoint.java
index 5432d33..be0b301 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SuccessfulCheckpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SuccessfulCheckpoint.java
@@ -18,14 +18,11 @@
package org.apache.flink.runtime.checkpoint;
-import com.google.common.collect.Maps;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
-import java.util.Map;
/**
* A successful checkpoint describes a checkpoint after all required tasks acknowledged it (with their state)
@@ -41,9 +38,7 @@ public class SuccessfulCheckpoint {
private final long timestamp;
- private final Map<JobVertexID, StateForTask> vertexToState;
-
- private final List<StateForTask> states;
+ private final List<StateForTask> states;
public SuccessfulCheckpoint(JobID job, long checkpointID, long timestamp, List<StateForTask> states) {
@@ -51,10 +46,6 @@ public class SuccessfulCheckpoint {
this.checkpointID = checkpointID;
this.timestamp = timestamp;
this.states = states;
- vertexToState = Maps.newHashMap();
- for(StateForTask state : states){
- vertexToState.put(state.getOperatorId(), state);
- }
}
public JobID getJobId() {
@@ -73,17 +64,6 @@ public class SuccessfulCheckpoint {
return states;
}
- /**
- * Returns the task state included in the checkpoint for a given JobVertexID if it exists or
- * null if no state is included for that id.
- *
- * @param jobVertexID
- * @return
- */
- public StateForTask getState(JobVertexID jobVertexID) {
- return vertexToState.get(jobVertexID);
- }
-
// --------------------------------------------------------------------------------------------
public void discard(ClassLoader userClassLoader) {
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
index d44cb6a..a70fa7d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
@@ -168,6 +168,10 @@ public class ExecutionVertex implements Serializable {
getTotalNumberOfParallelSubtasks());
}
+ public int getSubTaskIndex() {
+ return subTaskIndex;
+ }
+
public int getTotalNumberOfParallelSubtasks() {
return this.jobVertex.getParallelism();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointCommittingOperator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointCommittingOperator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointCommittingOperator.java
deleted file mode 100644
index a6f9851..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointCommittingOperator.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.jobgraph.tasks;
-
-import org.apache.flink.runtime.state.StateHandle;
-import org.apache.flink.runtime.util.SerializedValue;
-
-public interface CheckpointCommittingOperator {
-
- void confirmCheckpoint(long checkpointId, SerializedValue<StateHandle<?>> state) throws Exception;
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointNotificationOperator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointNotificationOperator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointNotificationOperator.java
new file mode 100644
index 0000000..90c82b7
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointNotificationOperator.java
@@ -0,0 +1,25 @@
+/*
+ * 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.jobgraph.tasks;
+
+
+public interface CheckpointNotificationOperator {
+
+ void notifyCheckpointComplete(long checkpointId) throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/ConfirmCheckpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/ConfirmCheckpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/ConfirmCheckpoint.java
deleted file mode 100644
index 328f692..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/ConfirmCheckpoint.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.messages.checkpoint;
-
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
-import org.apache.flink.runtime.state.StateHandle;
-import org.apache.flink.runtime.util.SerializedValue;
-
-/**
- * This message is sent from the {@link org.apache.flink.runtime.jobmanager.JobManager} to the
- * {@link org.apache.flink.runtime.taskmanager.TaskManager} to tell a task that the checkpoint
- * has been confirmed and that the task can commit the checkpoint to the outside world.
- */
-public class ConfirmCheckpoint extends AbstractCheckpointMessage implements java.io.Serializable {
-
- private static final long serialVersionUID = 2094094662279578953L;
-
- /** The timestamp associated with the checkpoint */
- private final long timestamp;
-
- /** The stateHandle associated with the checkpoint confirmation message*/
- private final SerializedValue<StateHandle<?>> state;
-
- public ConfirmCheckpoint(JobID job, ExecutionAttemptID taskExecutionId, long checkpointId, long timestamp,
- SerializedValue<StateHandle<?>> state) {
- super(job, taskExecutionId, checkpointId);
- this.timestamp = timestamp;
- this.state = state;
- }
-
- // --------------------------------------------------------------------------------------------
-
- public long getTimestamp() {
- return timestamp;
- }
-
- // --------------------------------------------------------------------------------------------
-
- /**
- * Returns the stateHandle that was included in the confirmed checkpoint for a given task or null
- * if no state was commited in that checkpoint.
- */
- public SerializedValue<StateHandle<?>> getState() {
- return state;
- }
-
- @Override
- public int hashCode() {
- return super.hashCode() + (int) (timestamp ^ (timestamp >>> 32));
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- else if (o instanceof ConfirmCheckpoint) {
- ConfirmCheckpoint that = (ConfirmCheckpoint) o;
- return this.timestamp == that.timestamp && super.equals(o);
- }
- else {
- return false;
- }
- }
-
- @Override
- public String toString() {
- return String.format("ConfirmCheckpoint %d for (%s/%s)",
- getCheckpointId(), getJob(), getTaskExecutionId());
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/NotifyCheckpointComplete.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/NotifyCheckpointComplete.java b/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/NotifyCheckpointComplete.java
new file mode 100644
index 0000000..c64c77a
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/NotifyCheckpointComplete.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.messages.checkpoint;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+
+/**
+ * This message is sent from the {@link org.apache.flink.runtime.jobmanager.JobManager} to the
+ * {@link org.apache.flink.runtime.taskmanager.TaskManager} to tell a task that the checkpoint
+ * has been confirmed and that the task can commit the checkpoint to the outside world.
+ */
+public class NotifyCheckpointComplete extends AbstractCheckpointMessage implements java.io.Serializable {
+
+ private static final long serialVersionUID = 2094094662279578953L;
+
+ /** The timestamp associated with the checkpoint */
+ private final long timestamp;
+
+ public NotifyCheckpointComplete(JobID job, ExecutionAttemptID taskExecutionId, long checkpointId, long timestamp) {
+ super(job, taskExecutionId, checkpointId);
+ this.timestamp = timestamp;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public long getTimestamp() {
+ return timestamp;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public int hashCode() {
+ return super.hashCode() + (int) (timestamp ^ (timestamp >>> 32));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ else if (o instanceof NotifyCheckpointComplete) {
+ NotifyCheckpointComplete that = (NotifyCheckpointComplete) o;
+ return this.timestamp == that.timestamp && super.equals(o);
+ }
+ else {
+ return false;
+ }
+ }
+
+ @Override
+ public String toString() {
+ return String.format("ConfirmCheckpoint %d for (%s/%s)",
+ getCheckpointId(), getJob(), getTaskExecutionId());
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/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 1b2fb08..616998c 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
@@ -46,7 +46,7 @@ import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-import org.apache.flink.runtime.jobgraph.tasks.CheckpointCommittingOperator;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointNotificationOperator;
import org.apache.flink.runtime.jobgraph.tasks.CheckpointedOperator;
import org.apache.flink.runtime.jobgraph.tasks.OperatorStateCarrier;
import org.apache.flink.runtime.memorymanager.MemoryManager;
@@ -883,11 +883,11 @@ public class Task implements Runnable {
checkpointer.triggerCheckpoint(checkpointID, checkpointTimestamp);
}
catch (Throwable t) {
- logger.error("Error while triggering checkpoint for " + taskName, t);
+ failExternally(new RuntimeException("Error while triggering checkpoint for " + taskName, t));
}
}
};
- executeAsyncCallRunnable(runnable, "Checkpoint Trigger");
+ executeAsyncCallRunnable(runnable, "Checkpoint Trigger for " + taskName);
}
else {
LOG.error("Task received a checkpoint request, but is not a checkpointing task - "
@@ -899,15 +899,14 @@ public class Task implements Runnable {
}
}
- public void confirmCheckpoint(final long checkpointID,
- final SerializedValue<StateHandle<?>> state) {
+ public void notifyCheckpointComplete(final long checkpointID) {
AbstractInvokable invokable = this.invokable;
if (executionState == ExecutionState.RUNNING && invokable != null) {
- if (invokable instanceof CheckpointCommittingOperator) {
+ if (invokable instanceof CheckpointNotificationOperator) {
// build a local closure
- final CheckpointCommittingOperator checkpointer = (CheckpointCommittingOperator) invokable;
+ final CheckpointNotificationOperator checkpointer = (CheckpointNotificationOperator) invokable;
final Logger logger = LOG;
final String taskName = taskNameWithSubtask;
@@ -915,14 +914,15 @@ public class Task implements Runnable {
@Override
public void run() {
try {
- checkpointer.confirmCheckpoint(checkpointID, state);
+ checkpointer.notifyCheckpointComplete(checkpointID);
}
catch (Throwable t) {
- logger.error("Error while confirming checkpoint for " + taskName, t);
+ // fail task if checkpoint confirmation failed.
+ failExternally(new RuntimeException("Error while confirming checkpoint", t));
}
}
};
- executeAsyncCallRunnable(runnable, "Checkpoint Confirmation");
+ executeAsyncCallRunnable(runnable, "Checkpoint Confirmation for " + taskName);
}
else {
LOG.error("Task received a checkpoint commit notification, but is not a checkpoint committing task - "
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/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 8c816a1..520decd 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
@@ -36,7 +36,7 @@ import com.fasterxml.jackson.databind.ObjectMapper
import grizzled.slf4j.Logger
import org.apache.flink.configuration.{Configuration, ConfigConstants, GlobalConfiguration, IllegalConfigurationException}
-import org.apache.flink.runtime.messages.checkpoint.{ConfirmCheckpoint, TriggerCheckpoint, AbstractCheckpointMessage}
+import org.apache.flink.runtime.messages.checkpoint.{NotifyCheckpointComplete, TriggerCheckpoint, AbstractCheckpointMessage}
import org.apache.flink.runtime.{StreamingMode, ActorSynchronousLogging, ActorLogMessages}
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.blob.{BlobService, BlobCache}
@@ -425,17 +425,16 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
log.debug(s"Taskmanager received a checkpoint request for unknown task $taskExecutionId.")
}
- case message: ConfirmCheckpoint =>
+ case message: NotifyCheckpointComplete =>
val taskExecutionId = message.getTaskExecutionId
val checkpointId = message.getCheckpointId
val timestamp = message.getTimestamp
- val state = message.getState
log.debug(s"Receiver ConfirmCheckpoint ${checkpointId}@${timestamp} for $taskExecutionId.")
val task = runningTasks.get(taskExecutionId)
if (task != null) {
- task.confirmCheckpoint(checkpointId, state)
+ task.notifyCheckpointComplete(checkpointId)
} else {
log.debug(
s"Taskmanager received a checkpoint confirmation for unknown task $taskExecutionId.")
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
index d56704d..c02d301 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
@@ -26,7 +26,7 @@ import org.apache.flink.runtime.executiongraph.Execution;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.executiongraph.ExecutionVertex;
import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint;
-import org.apache.flink.runtime.messages.checkpoint.ConfirmCheckpoint;
+import org.apache.flink.runtime.messages.checkpoint.NotifyCheckpointComplete;
import org.apache.flink.runtime.messages.checkpoint.TriggerCheckpoint;
import org.junit.Test;
@@ -199,8 +199,8 @@ public class CheckpointCoordinatorTest {
// validate that the relevant tasks got a confirmation message
{
- ConfirmCheckpoint confirmMessage1 = new ConfirmCheckpoint(jid, attemptID1, checkpointId, timestamp, null);
- ConfirmCheckpoint confirmMessage2 = new ConfirmCheckpoint(jid, attemptID2, checkpointId, timestamp, null);
+ NotifyCheckpointComplete confirmMessage1 = new NotifyCheckpointComplete(jid, attemptID1, checkpointId, timestamp);
+ NotifyCheckpointComplete confirmMessage2 = new NotifyCheckpointComplete(jid, attemptID2, checkpointId, timestamp);
verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(confirmMessage1), eq(attemptID1));
verify(vertex2, times(1)).sendMessageToCurrentExecution(eq(confirmMessage2), eq(attemptID2));
}
@@ -237,8 +237,8 @@ public class CheckpointCoordinatorTest {
verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(expectedMessage1), eq(attemptID1));
verify(vertex2, times(1)).sendMessageToCurrentExecution(eq(expectedMessage2), eq(attemptID2));
- ConfirmCheckpoint confirmMessage1 = new ConfirmCheckpoint(jid, attemptID1, checkpointIdNew, timestampNew, null);
- ConfirmCheckpoint confirmMessage2 = new ConfirmCheckpoint(jid, attemptID2, checkpointIdNew, timestampNew, null);
+ NotifyCheckpointComplete confirmMessage1 = new NotifyCheckpointComplete(jid, attemptID1, checkpointIdNew, timestampNew);
+ NotifyCheckpointComplete confirmMessage2 = new NotifyCheckpointComplete(jid, attemptID2, checkpointIdNew, timestampNew);
verify(vertex1, times(1)).sendMessageToCurrentExecution(eq(confirmMessage1), eq(attemptID1));
verify(vertex2, times(1)).sendMessageToCurrentExecution(eq(confirmMessage2), eq(attemptID2));
}
@@ -343,7 +343,7 @@ public class CheckpointCoordinatorTest {
// the first confirm message should be out
verify(commitVertex, times(1)).sendMessageToCurrentExecution(
- new ConfirmCheckpoint(jid, commitAttemptID, checkpointId1, timestamp1, null), commitAttemptID);
+ new NotifyCheckpointComplete(jid, commitAttemptID, checkpointId1, timestamp1), commitAttemptID);
// send the last remaining ack for the second checkpoint
coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, ackAttemptID3, checkpointId2));
@@ -355,7 +355,7 @@ public class CheckpointCoordinatorTest {
// the second commit message should be out
verify(commitVertex, times(1)).sendMessageToCurrentExecution(
- new ConfirmCheckpoint(jid, commitAttemptID, checkpointId2, timestamp2, null), commitAttemptID);
+ new NotifyCheckpointComplete(jid, commitAttemptID, checkpointId2, timestamp2), commitAttemptID);
// validate the committed checkpoints
List<SuccessfulCheckpoint> scs = coord.getSuccessfulCheckpoints();
@@ -482,7 +482,7 @@ public class CheckpointCoordinatorTest {
// the first confirm message should be out
verify(commitVertex, times(1)).sendMessageToCurrentExecution(
- new ConfirmCheckpoint(jid, commitAttemptID, checkpointId2, timestamp2, null), commitAttemptID);
+ new NotifyCheckpointComplete(jid, commitAttemptID, checkpointId2, timestamp2), commitAttemptID);
// send the last remaining ack for the first checkpoint. This should not do anything
coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, ackAttemptID3, checkpointId1));
@@ -551,7 +551,7 @@ public class CheckpointCoordinatorTest {
// no confirm message must have been sent
verify(commitVertex, times(0))
- .sendMessageToCurrentExecution(any(ConfirmCheckpoint.class), any(ExecutionAttemptID.class));
+ .sendMessageToCurrentExecution(any(NotifyCheckpointComplete.class), any(ExecutionAttemptID.class));
coord.shutdown();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-runtime/src/test/java/org/apache/flink/runtime/messages/CheckpointMessagesTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/messages/CheckpointMessagesTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/messages/CheckpointMessagesTest.java
index 05255ac..211d5e3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/messages/CheckpointMessagesTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/messages/CheckpointMessagesTest.java
@@ -22,7 +22,7 @@ import static org.junit.Assert.*;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
-import org.apache.flink.runtime.messages.checkpoint.ConfirmCheckpoint;
+import org.apache.flink.runtime.messages.checkpoint.NotifyCheckpointComplete;
import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint;
import org.apache.flink.runtime.messages.checkpoint.TriggerCheckpoint;
import org.apache.flink.runtime.state.StateHandle;
@@ -38,7 +38,7 @@ public class CheckpointMessagesTest {
@Test
public void testTriggerAndConfirmCheckpoint() {
try {
- ConfirmCheckpoint cc = new ConfirmCheckpoint(new JobID(), new ExecutionAttemptID(), 45287698767345L, 467L, null);
+ NotifyCheckpointComplete cc = new NotifyCheckpointComplete(new JobID(), new ExecutionAttemptID(), 45287698767345L, 467L);
testSerializabilityEqualsHashCode(cc);
TriggerCheckpoint tc = new TriggerCheckpoint(new JobID(), new ExecutionAttemptID(), 347652734L, 7576752L);
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
index 618c01f..b18bf2a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
@@ -40,12 +40,10 @@ import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNo
import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-import org.apache.flink.runtime.jobgraph.tasks.CheckpointCommittingOperator;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointNotificationOperator;
import org.apache.flink.runtime.jobgraph.tasks.CheckpointedOperator;
import org.apache.flink.runtime.memorymanager.MemoryManager;
-import org.apache.flink.runtime.state.StateHandle;
-import org.apache.flink.runtime.util.SerializedValue;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
@@ -134,7 +132,7 @@ public class TaskAsyncCallTest {
for (int i = 1; i <= NUM_CALLS; i++) {
task.triggerCheckpointBarrier(i, 156865867234L);
- task.confirmCheckpoint(i, null);
+ task.notifyCheckpointComplete(i);
}
triggerLatch.await();
@@ -186,7 +184,7 @@ public class TaskAsyncCallTest {
}
public static class CheckpointsInOrderInvokable extends AbstractInvokable
- implements CheckpointedOperator, CheckpointCommittingOperator {
+ implements CheckpointedOperator, CheckpointNotificationOperator {
private volatile long lastCheckpointId = 0;
@@ -213,7 +211,7 @@ public class TaskAsyncCallTest {
}
@Override
- public void triggerCheckpoint(long checkpointId, long timestamp) throws Exception {
+ public void triggerCheckpoint(long checkpointId, long timestamp) {
lastCheckpointId++;
if (checkpointId == lastCheckpointId) {
if (lastCheckpointId == NUM_CALLS) {
@@ -229,7 +227,7 @@ public class TaskAsyncCallTest {
}
@Override
- public void confirmCheckpoint(long checkpointId, SerializedValue<StateHandle<?>> state) throws Exception {
+ public void notifyCheckpointComplete(long checkpointId) {
if (checkpointId != lastCheckpointId && this.error == null) {
this.error = new Exception("calls out of order");
synchronized (this) {
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/Utils.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/Utils.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/Utils.java
index 4286196..abe06b8 100644
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/Utils.java
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/Utils.java
@@ -27,16 +27,25 @@ import org.apache.flink.streaming.util.serialization.SerializationSchema;
import java.io.IOException;
+/**
+ * Utilities for the Kafka connector
+ */
public class Utils {
- public static class TypeInformationSerializationSchema<T>
- implements DeserializationSchema<T>, SerializationSchema<T, byte[]> {
+
+ /**
+ * Utility serialization schema, created from Flink's TypeInformation system.
+ * @param <T>
+ */
+ public static class TypeInformationSerializationSchema<T> implements DeserializationSchema<T>, SerializationSchema<T, byte[]> {
private final TypeSerializer<T> serializer;
private final TypeInformation<T> ti;
+ private transient DataOutputSerializer dos;
- public TypeInformationSerializationSchema(Object type, ExecutionConfig ec) {
- this.ti = (TypeInformation<T>) TypeExtractor.getForObject(type);
+ public TypeInformationSerializationSchema(T type, ExecutionConfig ec) {
+ this.ti = TypeExtractor.getForObject(type);
this.serializer = ti.createSerializer(ec);
}
+
@Override
public T deserialize(byte[] message) {
try {
@@ -53,13 +62,22 @@ public class Utils {
@Override
public byte[] serialize(T element) {
- DataOutputSerializer dos = new DataOutputSerializer(16);
+ if(dos == null) {
+ dos = new DataOutputSerializer(16);
+ }
try {
serializer.serialize(element, dos);
} catch (IOException e) {
throw new RuntimeException("Unable to serialize record", e);
}
- return dos.getByteArray();
+ byte[] ret = dos.getByteArray();
+ if(ret.length != dos.length()) {
+ byte[] n = new byte[dos.length()];
+ System.arraycopy(ret, 0, n, 0, dos.length());
+ ret = n;
+ }
+ dos.clear();
+ return ret;
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/api/persistent/PersistentKafkaSource.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/api/persistent/PersistentKafkaSource.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/api/persistent/PersistentKafkaSource.java
index 6758f2c..b4a0b8b 100644
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/api/persistent/PersistentKafkaSource.java
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/api/persistent/PersistentKafkaSource.java
@@ -20,7 +20,6 @@ package org.apache.flink.streaming.connectors.kafka.api.persistent;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
-import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.util.Arrays;
import java.util.Collections;
@@ -41,12 +40,12 @@ import kafka.utils.ZkUtils;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkMarshallingError;
import org.I0Itec.zkclient.serialize.ZkSerializer;
-import org.apache.flink.api.common.state.OperatorState;
+import org.apache.commons.collections.map.LinkedMap;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.state.StateHandle;
-import org.apache.flink.streaming.api.checkpoint.CheckpointCommitter;
+import org.apache.flink.streaming.api.checkpoint.CheckpointNotifier;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedAsynchronously;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.util.serialization.DeserializationSchema;
import org.apache.zookeeper.data.Stat;
@@ -67,13 +66,14 @@ import com.google.common.base.Preconditions;
*/
public class PersistentKafkaSource<OUT> extends RichParallelSourceFunction<OUT> implements
ResultTypeQueryable<OUT>,
- CheckpointCommitter {
+ CheckpointNotifier, CheckpointedAsynchronously<long[]> {
private static final long serialVersionUID = 287845877188312621L;
private static final Logger LOG = LoggerFactory.getLogger(PersistentKafkaSource.class);
-
+ private final LinkedMap pendingCheckpoints = new LinkedMap();
+
private final String topicName;
private final DeserializationSchema<OUT> deserializationSchema;
@@ -82,9 +82,11 @@ public class PersistentKafkaSource<OUT> extends RichParallelSourceFunction<OUT>
private transient ConsumerConnector consumer;
private transient ZkClient zkClient;
- private transient OperatorState<long[]> lastOffsets;
- private transient long[] commitedOffsets; // maintain committed offsets, to avoid committing the same over and over again.
-
+
+ private transient long[] lastOffsets; // Current offset (backuped state)
+ protected transient long[] commitedOffsets; // maintain committed offsets, to avoid committing the same over and over again.
+ private transient long[] restoreState; // set by the restore() method, used by open() to valdiate the restored state.
+
private volatile boolean running;
/**
@@ -145,23 +147,26 @@ public class PersistentKafkaSource<OUT> extends RichParallelSourceFunction<OUT>
// most likely the number of offsets we're going to store here will be lower than the number of partitions.
int numPartitions = getNumberOfPartitions();
LOG.debug("The topic {} has {} partitions", topicName, numPartitions);
- this.lastOffsets = getRuntimeContext().getOperatorState("offset", new long[numPartitions], false);
+ this.lastOffsets = new long[numPartitions];
this.commitedOffsets = new long[numPartitions];
+
// check if there are offsets to restore
- if (!Arrays.equals(lastOffsets.value(), new long[numPartitions])) {
- if (lastOffsets.value().length != numPartitions) {
- throw new IllegalStateException("There are "+lastOffsets.value().length+" offsets to restore for topic "+topicName+" but " +
+ if (restoreState != null) {
+ if (restoreState.length != numPartitions) {
+ throw new IllegalStateException("There are "+restoreState.length+" offsets to restore for topic "+topicName+" but " +
"there are only "+numPartitions+" in the topic");
}
- LOG.info("Setting restored offsets {} in ZooKeeper", Arrays.toString(lastOffsets.value()));
- setOffsetsInZooKeeper(lastOffsets.value());
+ LOG.info("Setting restored offsets {} in ZooKeeper", Arrays.toString(restoreState));
+ setOffsetsInZooKeeper(restoreState);
+ this.lastOffsets = restoreState;
} else {
// initialize empty offsets
- Arrays.fill(this.lastOffsets.value(), -1);
+ Arrays.fill(this.lastOffsets, -1);
}
Arrays.fill(this.commitedOffsets, 0); // just to make it clear
-
+
+ pendingCheckpoints.clear();
running = true;
}
@@ -175,7 +180,7 @@ public class PersistentKafkaSource<OUT> extends RichParallelSourceFunction<OUT>
while (running && iteratorToRead.hasNext()) {
MessageAndMetadata<byte[], byte[]> message = iteratorToRead.next();
- if(lastOffsets.value()[message.partition()] >= message.offset()) {
+ if(lastOffsets[message.partition()] >= message.offset()) {
LOG.info("Skipping message with offset {} from partition {}", message.offset(), message.partition());
continue;
}
@@ -188,7 +193,7 @@ public class PersistentKafkaSource<OUT> extends RichParallelSourceFunction<OUT>
// make the state update and the element emission atomic
synchronized (checkpointLock) {
- lastOffsets.value()[message.partition()] = message.offset();
+ lastOffsets[message.partition()] = message.offset();
ctx.collect(next);
}
@@ -210,19 +215,63 @@ public class PersistentKafkaSource<OUT> extends RichParallelSourceFunction<OUT>
zkClient.close();
}
+ // ----------------- State Checkpointing -----------------
+
+ @Override
+ public long[] snapshotState(long checkpointId, long checkpointTimestamp) throws Exception {
+ if (lastOffsets == null) {
+ LOG.warn("State snapshot requested on not yet opened source. Returning null");
+ return null;
+ }
+
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Snapshotting state. Offsets: {}, checkpoint id {}, timestamp {}",
+ Arrays.toString(lastOffsets), checkpointId, checkpointTimestamp);
+ }
+
+ long[] currentOffsets = Arrays.copyOf(lastOffsets, lastOffsets.length);
+
+ // the map may be asynchronously updates when committing to Kafka, so we synchronize
+ synchronized (pendingCheckpoints) {
+ pendingCheckpoints.put(checkpointId, currentOffsets);
+ }
+
+ return currentOffsets;
+ }
+
+ @Override
+ public void restoreState(long[] state) {
+ LOG.info("The state will be restored to {} in the open() method", Arrays.toString(state));
+ this.restoreState = Arrays.copyOf(state, state.length);
+ }
+
/**
* Notification on completed checkpoints
* @param checkpointId The ID of the checkpoint that has been completed.
* @throws Exception
*/
- @Override
- public void commitCheckpoint(long checkpointId, String stateName, StateHandle<Serializable> state) throws Exception {
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
LOG.info("Commit checkpoint {}", checkpointId);
long[] checkpointOffsets;
- checkpointOffsets = (long[]) state.getState();
+ // the map may be asynchronously updates when snapshotting state, so we synchronize
+ synchronized (pendingCheckpoints) {
+ final int posInMap = pendingCheckpoints.indexOf(checkpointId);
+ if (posInMap == -1) {
+ LOG.warn("Unable to find pending checkpoint for id {}", checkpointId);
+ return;
+ }
+
+ checkpointOffsets = (long[]) pendingCheckpoints.remove(posInMap);
+ // remove older checkpoints in map:
+ if (!pendingCheckpoints.isEmpty()) {
+ for(int i = 0; i < posInMap; i++) {
+ pendingCheckpoints.remove(0);
+ }
+ }
+ }
if (LOG.isInfoEnabled()) {
LOG.info("Committing offsets {} to ZooKeeper", Arrays.toString(checkpointOffsets));
@@ -254,11 +303,15 @@ public class PersistentKafkaSource<OUT> extends RichParallelSourceFunction<OUT>
}
protected void setOffset(int partition, long offset) {
- if(commitedOffsets[partition] < offset) {
- setOffset(zkClient, consumerConfig.groupId(), topicName, partition, offset);
- commitedOffsets[partition] = offset;
- } else {
- LOG.debug("Ignoring offset {} for partition {} because it is already committed", offset, partition);
+ // synchronize because notifyCheckpointComplete is called using asynchronous worker threads (= multiple checkpoints might be confirmed concurrently)
+ synchronized (commitedOffsets) {
+ if(commitedOffsets[partition] < offset) {
+ LOG.info("Committed offsets {}, partition={}, offset={}, locking on {}", Arrays.toString(commitedOffsets), partition, offset, commitedOffsets.hashCode());
+ setOffset(zkClient, consumerConfig.groupId(), topicName, partition, offset);
+ commitedOffsets[partition] = offset;
+ } else {
+ LOG.debug("Ignoring offset {} for partition {} because it is already committed", offset, partition);
+ }
}
}
@@ -305,7 +358,6 @@ public class PersistentKafkaSource<OUT> extends RichParallelSourceFunction<OUT>
return deserializationSchema.getProducedType();
}
-
// ---------------------- Zookeeper Serializer copied from Kafka (because it has private access there) -----------------
public static class KafkaZKStringSerializer implements ZkSerializer {
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
index 4b763b2..e039592 100644
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
@@ -21,6 +21,7 @@ import static org.junit.Assert.assertEquals;
import java.io.File;
import java.io.IOException;
+import java.lang.reflect.Field;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
@@ -46,6 +47,7 @@ import kafka.server.KafkaConfig;
import kafka.server.KafkaServer;
import org.I0Itec.zkclient.ZkClient;
+import org.apache.commons.collections.map.LinkedMap;
import org.apache.curator.test.TestingServer;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.MapFunction;
@@ -182,6 +184,63 @@ public class KafkaITCase {
zkClient.close();
}
+ // -------------------------- test checkpointing ------------------------
+ @Test
+ public void testCheckpointing() throws Exception {
+ createTestTopic("testCheckpointing", 1, 1);
+
+ Properties props = new Properties();
+ props.setProperty("zookeeper.connect", zookeeperConnectionString);
+ props.setProperty("group.id", "testCheckpointing");
+ props.setProperty("auto.commit.enable", "false");
+ ConsumerConfig cc = new ConsumerConfig(props);
+ PersistentKafkaSource<String> source = new PersistentKafkaSource<String>("testCheckpointing", new FakeDeserializationSchema(), cc);
+
+
+ Field pendingCheckpointsField = PersistentKafkaSource.class.getDeclaredField("pendingCheckpoints");
+ pendingCheckpointsField.setAccessible(true);
+ LinkedMap pendingCheckpoints = (LinkedMap) pendingCheckpointsField.get(source);
+
+
+ Assert.assertEquals(0, pendingCheckpoints.size());
+ // first restore
+ source.restoreState(new long[]{1337});
+ // then open
+ source.open(new Configuration());
+ long[] state1 = source.snapshotState(1, 15);
+ Assert.assertArrayEquals(new long[]{1337}, state1);
+ long[] state2 = source.snapshotState(2, 30);
+ Assert.assertArrayEquals(new long[]{1337}, state2);
+ Assert.assertEquals(2, pendingCheckpoints.size());
+
+ source.notifyCheckpointComplete(1);
+ Assert.assertEquals(1, pendingCheckpoints.size());
+
+ source.notifyCheckpointComplete(2);
+ Assert.assertEquals(0, pendingCheckpoints.size());
+
+ source.notifyCheckpointComplete(666); // invalid checkpoint
+ Assert.assertEquals(0, pendingCheckpoints.size());
+
+ // create 500 snapshots
+ for(int i = 0; i < 500; i++) {
+ source.snapshotState(i, 15 * i);
+ }
+ Assert.assertEquals(500, pendingCheckpoints.size());
+
+ // commit only the second last
+ source.notifyCheckpointComplete(498);
+ Assert.assertEquals(1, pendingCheckpoints.size());
+
+ // access invalid checkpoint
+ source.notifyCheckpointComplete(490);
+
+ // and the last
+ source.notifyCheckpointComplete(499);
+ Assert.assertEquals(0, pendingCheckpoints.size());
+ }
+
+
private static class FakeDeserializationSchema implements DeserializationSchema<String> {
@Override
@@ -220,6 +279,34 @@ public class KafkaITCase {
zk.close();
}
+
+ public static class TestPersistentKafkaSource<OUT> extends PersistentKafkaSource<OUT> {
+ private static Object sync = new Object();
+ public static long[] finalOffset;
+ public TestPersistentKafkaSource(String topicName, DeserializationSchema<OUT> deserializationSchema, ConsumerConfig consumerConfig) {
+ super(topicName, deserializationSchema, consumerConfig);
+ }
+
+ @Override
+ public void close() {
+ super.close();
+ LOG.info("Starting close " +Arrays.toString(commitedOffsets));
+ synchronized (sync) {
+ if (finalOffset == null) {
+ finalOffset = new long[commitedOffsets.length];
+ }
+ for(int i = 0; i < commitedOffsets.length; i++) {
+ if(commitedOffsets[i] > 0) {
+ if(finalOffset[i] > 0) {
+ throw new RuntimeException("This is unexpected on i = "+i);
+ }
+ finalOffset[i] = commitedOffsets[i];
+ }
+ }
+ }
+ LOG.info("Finished closing. Final "+Arrays.toString(finalOffset));
+ }
+ }
/**
* We want to use the High level java consumer API but manage the offset in Zookeeper manually.
*
@@ -246,27 +333,35 @@ public class KafkaITCase {
// write a sequence from 0 to 99 to each of the three partitions.
writeSequence(env, topicName, 0, 99);
-
readSequence(env, standardCC, topicName, 0, 100, 300);
+ LOG.info("State in persistent kafka sources {}", TestPersistentKafkaSource.finalOffset);
+
// check offsets to be set at least higher than 50.
// correctly, we would expect them to be set to 99, but right now there is no way of stopping a topology once all pending
// checkpoints have been committed.
// To work around that limitation, the persistent kafka consumer is throtteled with a thread.sleep().
- long o1 = PersistentKafkaSource.getOffset(zk, standardCC.groupId(), topicName, 0);
- long o2 = PersistentKafkaSource.getOffset(zk, standardCC.groupId(), topicName, 1);
- long o3 = PersistentKafkaSource.getOffset(zk, standardCC.groupId(), topicName, 2);
- Assert.assertTrue("The offset seems incorrect, got " + o1, o1 > 50L);
- Assert.assertTrue("The offset seems incorrect, got " + o2, o2 > 50L);
- Assert.assertTrue("The offset seems incorrect, got " + o3, o3 > 50L);
- /** Once we have proper shutdown of streaming jobs, enable these tests
- Assert.assertEquals("The offset seems incorrect", 99L, PersistentKafkaSource.getOffset(zk, standardCC.groupId(), topicName, 0));
- Assert.assertEquals("The offset seems incorrect", 99L, PersistentKafkaSource.getOffset(zk, standardCC.groupId(), topicName, 1));
- Assert.assertEquals("The offset seems incorrect", 99L, PersistentKafkaSource.getOffset(zk, standardCC.groupId(), topicName, 2));*/
+ long o1 = -1, o2 = -1, o3 = -1;
+ if(TestPersistentKafkaSource.finalOffset[0] > 0) {
+ o1 = PersistentKafkaSource.getOffset(zk, standardCC.groupId(), topicName, 0);
+ Assert.assertTrue("The offset seems incorrect, got " + o1, o1 == TestPersistentKafkaSource.finalOffset[0]);
+ }
+ if(TestPersistentKafkaSource.finalOffset[1] > 0) {
+ o2 = PersistentKafkaSource.getOffset(zk, standardCC.groupId(), topicName, 1);
+ Assert.assertTrue("The offset seems incorrect, got " + o2, o2 == TestPersistentKafkaSource.finalOffset[1]);
+ }
+ if(TestPersistentKafkaSource.finalOffset[2] > 0) {
+ o3 = PersistentKafkaSource.getOffset(zk, standardCC.groupId(), topicName, 2);
+ Assert.assertTrue("The offset seems incorrect, got " + o3, o3 == TestPersistentKafkaSource.finalOffset[2]);
+ }
+ Assert.assertFalse("no offset has been set", TestPersistentKafkaSource.finalOffset[0] == 0 &&
+ TestPersistentKafkaSource.finalOffset[1] == 0 &&
+ TestPersistentKafkaSource.finalOffset[2] == 0);
+ LOG.info("Got final offsets from zookeeper o1={}, o2={}, o3={}", o1, o2, o3);
LOG.info("Manipulating offsets");
- // set the offset to 25, 50, and 75 for the three partitions
+ // set the offset to 50 for the three partitions
PersistentKafkaSource.setOffset(zk, standardCC.groupId(), topicName, 0, 50);
PersistentKafkaSource.setOffset(zk, standardCC.groupId(), topicName, 1, 50);
PersistentKafkaSource.setOffset(zk, standardCC.groupId(), topicName, 2, 50);
@@ -283,20 +378,16 @@ public class KafkaITCase {
private void readSequence(StreamExecutionEnvironment env, ConsumerConfig cc, final String topicName, final int valuesStartFrom, final int valuesCount, final int finalCount) throws Exception {
LOG.info("Reading sequence for verification until final count {}", finalCount);
- DataStream<Tuple2<Integer, Integer>> source = env.addSource(
- new PersistentKafkaSource<Tuple2<Integer, Integer>>(topicName, new Utils.TypeInformationSerializationSchema<Tuple2<Integer, Integer>>(new Tuple2<Integer, Integer>(1,1), env.getConfig()), cc)
- )
- //add a sleeper mapper. Since there is no good way of "shutting down" a running topology, we have
- // to play this trick. The problem is that we have to wait until all checkpoints are confirmed
- .map(new MapFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>() {
- private static final long serialVersionUID = 1L;
-
- @Override
- public Tuple2<Integer, Integer> map(Tuple2<Integer, Integer> value) throws Exception {
- Thread.sleep(150);
- return value;
- }
- }).setParallelism(3);
+ TestPersistentKafkaSource<Tuple2<Integer, Integer>> pks = new TestPersistentKafkaSource<Tuple2<Integer, Integer>>(topicName, new Utils.TypeInformationSerializationSchema<Tuple2<Integer, Integer>>(new Tuple2<Integer, Integer>(1, 1), env.getConfig()), cc);
+ DataStream<Tuple2<Integer, Integer>> source = env.addSource(pks).map(new MapFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>() {
+ // we need to slow down the source so that it can participate in a few checkpoints.
+ // Otherwise it would write its data into buffers and shut down.
+ @Override
+ public Tuple2<Integer, Integer> map(Tuple2<Integer, Integer> value) throws Exception {
+ Thread.sleep(50);
+ return value;
+ }
+ });
// verify data
DataStream<Integer> validIndexes = source.flatMap(new RichFlatMapFunction<Tuple2<Integer, Integer>, Integer>() {
@@ -312,7 +403,6 @@ public class KafkaITCase {
LOG.info("Reader " + getRuntimeContext().getIndexOfThisSubtask() + " got " + value + " count=" + count + "/" + finalCount);
// verify if we've seen everything
-
if (count == finalCount) {
LOG.info("Received all values");
for (int i = 0; i < values.length; i++) {
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/util/UtilsTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/util/UtilsTest.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/util/UtilsTest.java
new file mode 100644
index 0000000..32bd1d1
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/util/UtilsTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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.streaming.connectors.kafka.util;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.connectors.kafka.Utils;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class UtilsTest {
+
+ /**
+ * Ensure that the returned byte array has the expected size
+ */
+ @Test
+ public void testTypeInformationSerializationSchema() {
+ final ExecutionConfig ec = new ExecutionConfig();
+
+ Tuple2<Integer, Integer> test = new Tuple2<Integer, Integer>(1,666);
+
+ Utils.TypeInformationSerializationSchema<Tuple2<Integer, Integer>> ser = new Utils.TypeInformationSerializationSchema<Tuple2<Integer, Integer>>(test, ec);
+
+ byte[] res = ser.serialize(test);
+ Assert.assertEquals(8, res.length);
+
+ Tuple2<Integer, Integer> another = ser.deserialize(res);
+ Assert.assertEquals(test.f0, another.f0);
+ Assert.assertEquals(test.f1, another.f1);
+ }
+
+ @Test
+ public void testGrowing() {
+ final ExecutionConfig ec = new ExecutionConfig();
+
+ Tuple2<Integer, byte[]> test1 = new Tuple2<Integer, byte[]>(1, new byte[16]);
+
+ Utils.TypeInformationSerializationSchema<Tuple2<Integer, byte[]>> ser = new Utils.TypeInformationSerializationSchema<Tuple2<Integer, byte[]>>(test1, ec);
+
+ byte[] res = ser.serialize(test1);
+ Assert.assertEquals(24, res.length);
+ Tuple2<Integer, byte[]> another = ser.deserialize(res);
+ Assert.assertEquals(16, another.f1.length);
+
+ test1 = new Tuple2<Integer, byte[]>(1, new byte[26]);
+
+ res = ser.serialize(test1);
+ Assert.assertEquals(34, res.length);
+ another = ser.deserialize(res);
+ Assert.assertEquals(26, another.f1.length);
+
+ test1 = new Tuple2<Integer, byte[]>(1, new byte[1]);
+
+ res = ser.serialize(test1);
+ Assert.assertEquals(9, res.length);
+ another = ser.deserialize(res);
+ Assert.assertEquals(1, another.f1.length);
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointCommitter.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointCommitter.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointCommitter.java
deleted file mode 100644
index 306df71..0000000
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointCommitter.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.api.checkpoint;
-
-import java.io.Serializable;
-
-import org.apache.flink.runtime.state.StateHandle;
-
-/**
- * This interface must be implemented by functions/operations that want to receive
- * a commit notification once a checkpoint has been completely acknowledged by all
- * participants.
- */
-public interface CheckpointCommitter {
-
- /**
- * This method is called as a notification once a distributed checkpoint has been completed.
- *
- * Note that any exception during this method will not cause the checkpoint to
- * fail any more.
- *
- * @param checkpointId The ID of the checkpoint that has been completed.
- * @param stateName The name of the committed state
- * @param checkPointedState Handle to the state that was checkpointed with this checkpoint id.
- * @throws Exception
- */
- void commitCheckpoint(long checkpointId, String stateName, StateHandle<Serializable> checkPointedState) throws Exception;
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointNotifier.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointNotifier.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointNotifier.java
new file mode 100644
index 0000000..c2d2182
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointNotifier.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.api.checkpoint;
+
+/**
+ * This interface must be implemented by functions/operations that want to receive
+ * a commit notification once a checkpoint has been completely acknowledged by all
+ * participants.
+ */
+public interface CheckpointNotifier {
+
+ /**
+ * This method is called as a notification once a distributed checkpoint has been completed.
+ *
+ * Note that any exception during this method will not cause the checkpoint to
+ * fail any more.
+ *
+ * @param checkpointId The ID of the checkpoint that has been completed.
+ * @throws Exception
+ */
+ void notifyCheckpointComplete(long checkpointId) throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
index 4d541bc..c988150 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
@@ -394,15 +394,16 @@ public class StreamingJobGraphGenerator {
List<JobVertexID> ackVertices = new ArrayList<JobVertexID>(jobVertices.size());
// collect the vertices that receive "commit checkpoint" messages
- // currently, these are only the sources
+ // currently, these are all certices
List<JobVertexID> commitVertices = new ArrayList<JobVertexID>();
for (JobVertex vertex : jobVertices.values()) {
if (vertex.isInputVertex()) {
triggerVertices.add(vertex.getID());
- commitVertices.add(vertex.getID());
}
+ // TODO: add check whether the user function implements the checkpointing interface
+ commitVertices.add(vertex.getID());
ackVertices.add(vertex.getID());
}
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
index c128a7b..b2d9c91 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
@@ -30,7 +30,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.PartitionedStateHandle;
import org.apache.flink.runtime.state.StateHandle;
import org.apache.flink.runtime.state.StateHandleProvider;
-import org.apache.flink.streaming.api.checkpoint.CheckpointCommitter;
+import org.apache.flink.streaming.api.checkpoint.CheckpointNotifier;
import org.apache.flink.streaming.api.checkpoint.Checkpointed;
import org.apache.flink.streaming.api.state.StreamOperatorState;
import org.apache.flink.streaming.runtime.tasks.StreamingRuntimeContext;
@@ -134,11 +134,10 @@ public abstract class AbstractUdfStreamOperator<OUT, F extends Function & Serial
}
- public void confirmCheckpointCompleted(long checkpointId, String stateName,
- StateHandle<Serializable> checkpointedState) throws Exception {
- if (userFunction instanceof CheckpointCommitter) {
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ if (userFunction instanceof CheckpointNotifier) {
try {
- ((CheckpointCommitter) userFunction).commitCheckpoint(checkpointId, stateName, checkpointedState);
+ ((CheckpointNotifier) userFunction).notifyCheckpointComplete(checkpointId);
} catch (Exception e) {
throw new Exception("Error while confirming checkpoint " + checkpointId + " to the stream function", e);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StatefulStreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StatefulStreamOperator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StatefulStreamOperator.java
index 6b5a3e8..afc36e0 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StatefulStreamOperator.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StatefulStreamOperator.java
@@ -36,5 +36,5 @@ public interface StatefulStreamOperator<OUT> extends StreamOperator<OUT> {
Tuple2<StateHandle<Serializable>, Map<String, PartitionedStateHandle>> getStateSnapshotFromFunction(long checkpointId, long timestamp) throws Exception;
- void confirmCheckpointCompleted(long checkpointId, String stateName, StateHandle<Serializable> checkpointedState) throws Exception;
+ void notifyCheckpointComplete(long checkpointId) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index 7421a33..f98ed2d 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -22,7 +22,6 @@ import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.functors.NotNullPredicate;
@@ -33,7 +32,7 @@ import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.runtime.event.task.TaskEvent;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-import org.apache.flink.runtime.jobgraph.tasks.CheckpointCommittingOperator;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointNotificationOperator;
import org.apache.flink.runtime.jobgraph.tasks.CheckpointedOperator;
import org.apache.flink.runtime.jobgraph.tasks.OperatorStateCarrier;
import org.apache.flink.runtime.state.FileStateHandle;
@@ -41,7 +40,6 @@ import org.apache.flink.runtime.state.LocalStateHandle;
import org.apache.flink.runtime.state.PartitionedStateHandle;
import org.apache.flink.runtime.state.StateHandle;
import org.apache.flink.runtime.state.StateHandleProvider;
-import org.apache.flink.runtime.util.SerializedValue;
import org.apache.flink.runtime.util.event.EventListener;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.StatefulStreamOperator;
@@ -52,7 +50,7 @@ import org.slf4j.LoggerFactory;
public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends AbstractInvokable implements
- OperatorStateCarrier<StateHandle<Serializable>>, CheckpointedOperator, CheckpointCommittingOperator {
+ OperatorStateCarrier<StateHandle<Serializable>>, CheckpointedOperator, CheckpointNotificationOperator {
private static final Logger LOG = LoggerFactory.getLogger(StreamTask.class);
@@ -69,11 +67,11 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
protected volatile boolean isRunning = false;
protected List<StreamingRuntimeContext> contexts;
-
+
protected StreamingRuntimeContext headContext;
protected ClassLoader userClassLoader;
-
+
private EventListener<TaskEvent> superstepListener;
public StreamTask() {
@@ -86,11 +84,11 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
public void registerInputOutput() {
this.userClassLoader = getUserCodeClassLoader();
this.configuration = new StreamConfig(getTaskConfiguration());
-
+
streamOperator = configuration.getStreamOperator(userClassLoader);
outputHandler = new OutputHandler<OUT>(this);
-
+
if (streamOperator != null) {
// IterationHead and IterationTail don't have an Operator...
@@ -110,13 +108,13 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
public StreamingRuntimeContext createRuntimeContext(StreamConfig conf) {
Environment env = getEnvironment();
String operatorName = conf.getStreamOperator(userClassLoader).getClass().getSimpleName();
-
+
KeySelector<?,Serializable> statePartitioner = conf.getStatePartitioner(userClassLoader);
-
+
return new StreamingRuntimeContext(operatorName, env, getUserCodeClassLoader(),
getExecutionConfig(), statePartitioner, getStateHandleProvider());
}
-
+
private StateHandleProvider<Serializable> getStateHandleProvider() {
StateHandleProvider<Serializable> provider = configuration
@@ -134,7 +132,7 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
} catch (Exception e) {
throw new RuntimeException(backendName + " is not a valid state backend.\nSupported backends: jobmanager, filesystem.");
}
-
+
switch (backend) {
case JOBMANAGER:
LOG.info("State backend for state checkpoints is set to jobmanager.");
@@ -152,7 +150,7 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
default:
throw new RuntimeException("Backend " + backend + " is not supported yet.");
}
-
+
} else {
LOG.info("Using user defined state backend for streaming checkpoitns.");
return provider;
@@ -199,7 +197,7 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
@SuppressWarnings("unchecked")
@Override
public void setInitialState(StateHandle<Serializable> stateHandle) throws Exception {
-
+
// We retrieve end restore the states for the chained oeprators.
List<Tuple2<StateHandle<Serializable>, Map<String, PartitionedStateHandle>>> chainedStates = (List<Tuple2<StateHandle<Serializable>, Map<String, PartitionedStateHandle>>>) stateHandle.getState();
@@ -217,15 +215,15 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
@Override
public void triggerCheckpoint(long checkpointId, long timestamp) throws Exception {
-
+
synchronized (checkpointLock) {
if (isRunning) {
try {
LOG.debug("Starting checkpoint {} on task {}", checkpointId, getName());
-
+
// We wrap the states of the chained operators in a list, marking non-stateful oeprators with null
List<Tuple2<StateHandle<Serializable>, Map<String, PartitionedStateHandle>>> chainedStates = new ArrayList<Tuple2<StateHandle<Serializable>, Map<String, PartitionedStateHandle>>>();
-
+
// A wrapper handle is created for the List of statehandles
WrapperStateHandle stateHandle;
try {
@@ -240,17 +238,17 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
chainedStates.add(null);
}
}
-
+
stateHandle = CollectionUtils.exists(chainedStates,
NotNullPredicate.INSTANCE) ? new WrapperStateHandle(chainedStates) : null;
}
catch (Exception e) {
throw new Exception("Error while drawing snapshot of the user state.", e);
}
-
+
// now emit the checkpoint barriers
outputHandler.broadcastBarrier(checkpointId, timestamp);
-
+
// now confirm the checkpoint
if (stateHandle == null) {
getEnvironment().acknowledgeCheckpoint(checkpointId);
@@ -270,44 +268,24 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
- public void confirmCheckpoint(long checkpointId, SerializedValue<StateHandle<?>> stateHandle) throws Exception {
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ // we do nothing here so far. this should call commit on the source function, for example
synchronized (checkpointLock) {
- if (stateHandle != null) {
- List<Tuple2<StateHandle<Serializable>, Map<String, PartitionedStateHandle>>> chainedStates = (List<Tuple2<StateHandle<Serializable>, Map<String, PartitionedStateHandle>>>) stateHandle
- .deserializeValue(getUserCodeClassLoader()).getState();
-
- for (int i = 0; i < chainedStates.size(); i++) {
- Tuple2<StateHandle<Serializable>, Map<String, PartitionedStateHandle>> chainedState = chainedStates
- .get(i);
- StreamOperator<?> chainedOperator = outputHandler.getChainedOperators().get(i);
-
- if (chainedState != null) {
- if (chainedState.f0 != null) {
- ((StatefulStreamOperator) chainedOperator).confirmCheckpointCompleted(
- checkpointId, null, chainedState.f0);
- }
+ if (streamOperator instanceof StatefulStreamOperator) {
+ ((StatefulStreamOperator) streamOperator).notifyCheckpointComplete(checkpointId);
+ }
- if (chainedState.f1 != null) {
- if (chainedOperator instanceof StatefulStreamOperator) {
- for (Entry<String, PartitionedStateHandle> stateEntry : chainedState.f1
- .entrySet()) {
- for (StateHandle<Serializable> handle : stateEntry.getValue()
- .getState().values()) {
- ((StatefulStreamOperator) chainedOperator)
- .confirmCheckpointCompleted(checkpointId,
- stateEntry.getKey(), handle);
- }
- }
- }
- }
+ if (hasChainedOperators) {
+ for (StreamOperator<?> chainedOperator : outputHandler.getChainedOperators()) {
+ if (chainedOperator instanceof StatefulStreamOperator) {
+ ((StatefulStreamOperator) chainedOperator).notifyCheckpointComplete(checkpointId);
}
}
}
-
}
}
-
-
+
+
// ------------------------------------------------------------------------
// Utilities
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-staging/flink-streaming/flink-streaming-core/src/test/resources/log4j-test.properties
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/resources/log4j-test.properties b/flink-staging/flink-streaming/flink-streaming-core/src/test/resources/log4j-test.properties
index 0b686e5..fcee1b5 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/resources/log4j-test.properties
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/resources/log4j-test.properties
@@ -17,7 +17,7 @@
################################################################################
# Set root logger level to DEBUG and its only appender to A1.
-log4j.rootLogger=OFF, A1
+log4j.rootLogger=INFO, A1
# A1 is set to be a ConsoleAppender.
log4j.appender.A1=org.apache.log4j.ConsoleAppender
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamCheckpointingITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamCheckpointingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamCheckpointingITCase.java
index a826eff..1730c63 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamCheckpointingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamCheckpointingITCase.java
@@ -23,26 +23,35 @@ import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.IOException;
+import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import org.apache.flink.api.common.functions.RichFilterFunction;
+import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.state.OperatorState;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.StateHandle;
+import org.apache.flink.streaming.api.checkpoint.CheckpointNotifier;
import org.apache.flink.streaming.api.checkpoint.Checkpointed;
import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.test.util.ForkableFlinkMiniCluster;
+import org.apache.flink.util.Collector;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* A simple test that runs a streaming topology with checkpointing enabled.
@@ -87,9 +96,8 @@ public class StreamCheckpointingITCase {
fail("Failed to stop test cluster: " + e.getMessage());
}
}
-
-
-
+
+
/**
* Runs the following program:
*
http://git-wip-us.apache.org/repos/asf/flink/blob/aa5e5b30/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YarnTestBase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YarnTestBase.java b/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YarnTestBase.java
index 43285b5..23b8940 100644
--- a/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YarnTestBase.java
+++ b/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YarnTestBase.java
@@ -467,7 +467,11 @@ public abstract class YarnTestBase {
// check if thread died
if(!runner.isAlive()) {
sendOutput();
- Assert.fail("Runner thread died before the test was finished. Return value = " +runner.getReturnValue());
+ if(runner.getReturnValue() != 0) {
+ Assert.fail("Runner thread died before the test was finished. Return value = " + runner.getReturnValue());
+ } else {
+ LOG.info("Runner stopped earlier than expected with return value = 0");
+ }
}
}
sendOutput();
[3/7] flink git commit: [FLINK-2235] Fix tests to allow exception in
'EnvironmentInformationTest.getSizeOfFreeHeapMemory()' if Xmx is not set.
Posted by se...@apache.org.
[FLINK-2235] Fix tests to allow exception in 'EnvironmentInformationTest.getSizeOfFreeHeapMemory()' if Xmx is not set.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b0b0cb50
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b0b0cb50
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b0b0cb50
Branch: refs/heads/master
Commit: b0b0cb50d63e2f831196fc9219858467a07d8302
Parents: 910e8c4
Author: Stephan Ewen <se...@apache.org>
Authored: Mon Jul 13 16:22:04 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Jul 13 17:54:31 2015 +0200
----------------------------------------------------------------------
.../flink/runtime/util/EnvironmentInformationTest.java | 12 ++++++++++--
1 file changed, 10 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b0b0cb50/flink-runtime/src/test/java/org/apache/flink/runtime/util/EnvironmentInformationTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/util/EnvironmentInformationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/util/EnvironmentInformationTest.java
index 771c57c..3890a81 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/util/EnvironmentInformationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/util/EnvironmentInformationTest.java
@@ -30,12 +30,20 @@ public class EnvironmentInformationTest {
public void testJavaMemory() {
try {
long fullHeap = EnvironmentInformation.getMaxJvmHeapMemory();
- long free = EnvironmentInformation.getSizeOfFreeHeapMemory();
long freeWithGC = EnvironmentInformation.getSizeOfFreeHeapMemoryWithDefrag();
assertTrue(fullHeap > 0);
- assertTrue(free >= 0);
assertTrue(freeWithGC >= 0);
+
+ try {
+ long free = EnvironmentInformation.getSizeOfFreeHeapMemory();
+ assertTrue(free >= 0);
+ }
+ catch (RuntimeException e) {
+ // this may only occur if the Xmx is not set
+ assertEquals(Long.MAX_VALUE, EnvironmentInformation.getMaxJvmHeapMemory());
+ }
+
// we cannot make these assumptions, because the test JVM may grow / shrink during the GC
// assertTrue(free <= fullHeap);
[5/7] flink git commit: [FLINK-2329] [runtime] Introduces
InstanceGateway as an abstraction to communicate with the TaskManager.
Posted by se...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerSlotSharingTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerSlotSharingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerSlotSharingTest.java
index f987e07..3e90123 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerSlotSharingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerSlotSharingTest.java
@@ -32,11 +32,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.flink.runtime.instance.SimpleSlot;
-import akka.actor.ActorSystem;
-import akka.testkit.JavaTestKit;
import org.apache.flink.runtime.testingUtils.TestingUtils;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -45,21 +41,7 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
* Tests for the scheduler when scheduling tasks in slot sharing groups.
*/
public class SchedulerSlotSharingTest {
- private static ActorSystem system;
- @BeforeClass
- public static void setup(){
- system = ActorSystem.create("TestingActorSystem", TestingUtils.testConfig());
- TestingUtils.setCallingThreadDispatcher(system);
- }
-
- @AfterClass
- public static void teardown(){
- TestingUtils.setGlobalExecutionContext();
- JavaTestKit.shutdownActorSystem(system);
- }
-
-
@Test
public void scheduleSingleVertexType() {
try {
@@ -67,7 +49,7 @@ public class SchedulerSlotSharingTest {
SlotSharingGroup sharingGroup = new SlotSharingGroup(jid1);
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
Instance i1 = getRandomInstance(2);
Instance i2 = getRandomInstance(2);
scheduler.newInstanceAvailable(i1);
@@ -154,7 +136,7 @@ public class SchedulerSlotSharingTest {
SlotSharingGroup sharingGroup = new SlotSharingGroup(jid1, jid2);
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
scheduler.newInstanceAvailable(getRandomInstance(2));
scheduler.newInstanceAvailable(getRandomInstance(2));
@@ -274,7 +256,7 @@ public class SchedulerSlotSharingTest {
SlotSharingGroup sharingGroup = new SlotSharingGroup(jid1, jid2);
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
scheduler.newInstanceAvailable(getRandomInstance(2));
scheduler.newInstanceAvailable(getRandomInstance(2));
@@ -339,7 +321,7 @@ public class SchedulerSlotSharingTest {
SlotSharingGroup sharingGroup = new SlotSharingGroup(jid1, jid2, jid3);
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
scheduler.newInstanceAvailable(getRandomInstance(2));
scheduler.newInstanceAvailable(getRandomInstance(2));
@@ -450,7 +432,7 @@ public class SchedulerSlotSharingTest {
SlotSharingGroup sharingGroup = new SlotSharingGroup(jid1, jid2);
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
scheduler.newInstanceAvailable(getRandomInstance(2));
// schedule 1 tasks from the first vertex group and 2 from the second
@@ -502,7 +484,7 @@ public class SchedulerSlotSharingTest {
SlotSharingGroup sharingGroup = new SlotSharingGroup(jid1, jid2);
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
scheduler.newInstanceAvailable(getRandomInstance(3));
scheduler.newInstanceAvailable(getRandomInstance(2));
@@ -649,7 +631,7 @@ public class SchedulerSlotSharingTest {
Instance i1 = getRandomInstance(2);
Instance i2 = getRandomInstance(2);
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
scheduler.newInstanceAvailable(i1);
scheduler.newInstanceAvailable(i2);
@@ -699,7 +681,7 @@ public class SchedulerSlotSharingTest {
Instance i1 = getRandomInstance(2);
Instance i2 = getRandomInstance(2);
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
scheduler.newInstanceAvailable(i1);
scheduler.newInstanceAvailable(i2);
@@ -749,7 +731,7 @@ public class SchedulerSlotSharingTest {
Instance i1 = getRandomInstance(2);
Instance i2 = getRandomInstance(2);
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
scheduler.newInstanceAvailable(i1);
scheduler.newInstanceAvailable(i2);
@@ -795,7 +777,6 @@ public class SchedulerSlotSharingTest {
@Test
public void testSequentialAllocateAndRelease() {
- TestingUtils.setGlobalExecutionContext();
try {
final JobVertexID jid1 = new JobVertexID();
final JobVertexID jid2 = new JobVertexID();
@@ -804,7 +785,7 @@ public class SchedulerSlotSharingTest {
final SlotSharingGroup sharingGroup = new SlotSharingGroup(jid1, jid2, jid3, jid4);
- final Scheduler scheduler = new Scheduler();
+ final Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
scheduler.newInstanceAvailable(getRandomInstance(4));
// allocate something from group 1 and 2 interleaved with schedule for group 3
@@ -853,15 +834,13 @@ public class SchedulerSlotSharingTest {
catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
- }finally{
- TestingUtils.setCallingThreadDispatcher(system);
}
}
@Test
public void testConcurrentAllocateAndRelease() {
final ExecutorService executor = Executors.newFixedThreadPool(20);
- TestingUtils.setGlobalExecutionContext();
+
try {
for (int run = 0; run < 50; run++) {
final JobVertexID jid1 = new JobVertexID();
@@ -871,7 +850,7 @@ public class SchedulerSlotSharingTest {
final SlotSharingGroup sharingGroup = new SlotSharingGroup(jid1, jid2, jid3, jid4);
- final Scheduler scheduler = new Scheduler();
+ final Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
scheduler.newInstanceAvailable(getRandomInstance(4));
final AtomicInteger enumerator1 = new AtomicInteger();
@@ -1030,10 +1009,6 @@ public class SchedulerSlotSharingTest {
e.printStackTrace();
fail(e.getMessage());
}
- finally {
- executor.shutdownNow();
- TestingUtils.setCallingThreadDispatcher(system);
- }
}
@Test
@@ -1046,7 +1021,7 @@ public class SchedulerSlotSharingTest {
SlotSharingGroup sharingGroup = new SlotSharingGroup(jid1, jid2, jid3, jid4);
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
scheduler.newInstanceAvailable(getRandomInstance(4));
// schedule one task for the first and second vertex
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerTestUtils.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerTestUtils.java
index 694b88b..2de0635 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerTestUtils.java
@@ -29,9 +29,9 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.atomic.AtomicInteger;
-import akka.actor.ActorRef;
import org.apache.flink.runtime.executiongraph.Execution;
import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.instance.DummyInstanceGateway;
import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.InstanceConnectionInfo;
@@ -66,7 +66,7 @@ public class SchedulerTestUtils {
final long GB = 1024L*1024*1024;
HardwareDescription resources = new HardwareDescription(4, 4*GB, 3*GB, 2*GB);
- return new Instance(ActorRef.noSender(), ci, new InstanceID(), resources, numSlots);
+ return new Instance(DummyInstanceGateway.INSTANCE, ci, new InstanceID(), resources, numSlots);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java
index a67cd00..442ddcf 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java
@@ -39,6 +39,7 @@ import org.apache.flink.runtime.memorymanager.DefaultMemoryManager;
import org.apache.flink.runtime.memorymanager.MemoryManager;
import org.apache.flink.runtime.messages.TaskManagerMessages;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.junit.Test;
import scala.Option;
import scala.Tuple2;
@@ -89,7 +90,10 @@ public class TaskManagerComponentsStartupShutdownTest {
final MemoryManager memManager = new DefaultMemoryManager(32 * BUFFER_SIZE, 1, BUFFER_SIZE, false);
final IOManager ioManager = new IOManagerAsync(TMP_DIR);
- final NetworkEnvironment network = new NetworkEnvironment(timeout, netConf);
+ final NetworkEnvironment network = new NetworkEnvironment(
+ TestingUtils.defaultExecutionContext(),
+ timeout,
+ netConf);
final int numberOfSlots = 1;
// create the task manager
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.scala
index a1ca43c..b3090e6 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.scala
@@ -18,36 +18,27 @@
package org.apache.flink.runtime.executiongraph
-import akka.actor.{Props, ActorSystem}
-import akka.testkit.TestKit
import org.apache.flink.api.common.JobID
import org.apache.flink.configuration.Configuration
import org.apache.flink.runtime.akka.AkkaUtils
+import org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.SimpleInstanceGateway
import org.apache.flink.runtime.jobgraph.{JobStatus, JobGraph, JobVertex}
import org.apache.flink.runtime.jobmanager.Tasks
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler
import org.apache.flink.runtime.testingUtils.TestingUtils
-import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
+import org.scalatest.{Matchers, WordSpecLike}
-class ExecutionGraphRestartTest(_system: ActorSystem) extends TestKit(_system) with WordSpecLike
-with Matchers with BeforeAndAfterAll {
-
- def this() = this(ActorSystem("TestingActorSystem", TestingUtils.testConfig))
-
- override def afterAll(): Unit = {
- TestKit.shutdownActorSystem(system)
- }
+class ExecutionGraphRestartTest extends WordSpecLike with Matchers {
val NUM_TASKS = 31
"The execution graph" must {
"be manually restartable" in {
try {
- val tm = system.actorOf(Props(classOf[ExecutionGraphTestUtils
- .SimpleAcknowledgingTaskManager], "TaskManager"))
- val instance = ExecutionGraphTestUtils.getInstance(tm)
+ val instance = ExecutionGraphTestUtils.getInstance(
+ new SimpleInstanceGateway(TestingUtils.directExecutionContext))
- val scheduler = new Scheduler
+ val scheduler = new Scheduler(TestingUtils.defaultExecutionContext)
scheduler.newInstanceAvailable(instance)
val sender = new JobVertex("Task")
@@ -56,7 +47,11 @@ with Matchers with BeforeAndAfterAll {
val jobGraph = new JobGraph("Pointwise job", sender)
- val eg = new ExecutionGraph(new JobID(), "test job", new Configuration(),
+ val eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext,
+ new JobID(),
+ "test job",
+ new Configuration(),
AkkaUtils.getDefaultTimeout)
eg.setNumberOfRetriesLeft(0)
eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources)
@@ -87,11 +82,10 @@ with Matchers with BeforeAndAfterAll {
"restart itself automatically" in {
try {
- val tm = system.actorOf(Props
- (classOf[ExecutionGraphTestUtils.SimpleAcknowledgingTaskManager], "TaskManager"))
- val instance = ExecutionGraphTestUtils.getInstance(tm)
+ val instance = ExecutionGraphTestUtils.getInstance(
+ new SimpleInstanceGateway(TestingUtils.directExecutionContext))
- val scheduler = new Scheduler
+ val scheduler = new Scheduler(TestingUtils.defaultExecutionContext)
scheduler.newInstanceAvailable(instance)
val sender = new JobVertex("Task")
@@ -100,7 +94,11 @@ with Matchers with BeforeAndAfterAll {
val jobGraph = new JobGraph("Pointwise job", sender)
- val eg = new ExecutionGraph(new JobID(), "Test job", new Configuration(),
+ val eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext,
+ new JobID(),
+ "Test job",
+ new Configuration(),
AkkaUtils.getDefaultTimeout)
eg.setNumberOfRetriesLeft(1)
eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources)
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/TaskManagerLossFailsTasksTest.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/TaskManagerLossFailsTasksTest.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/TaskManagerLossFailsTasksTest.scala
index 13199bc..aaa0025 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/TaskManagerLossFailsTasksTest.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/TaskManagerLossFailsTasksTest.scala
@@ -18,38 +18,27 @@
package org.apache.flink.runtime.executiongraph
-import akka.actor.{Props, ActorSystem}
-import akka.testkit.TestKit
import org.apache.flink.api.common.JobID
import org.apache.flink.configuration.Configuration
import org.apache.flink.runtime.akka.AkkaUtils
-import org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils
-.SimpleAcknowledgingTaskManager
+import org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.SimpleInstanceGateway
import org.apache.flink.runtime.jobgraph.{JobStatus, JobGraph, JobVertex}
import org.apache.flink.runtime.jobmanager.Tasks
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler
import org.apache.flink.runtime.testingUtils.TestingUtils
-import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
+import org.scalatest.{Matchers, WordSpecLike}
-class TaskManagerLossFailsTasksTest(_system: ActorSystem) extends TestKit(_system) with
-WordSpecLike with Matchers with BeforeAndAfterAll {
-
- def this() = this(ActorSystem("TestingActorSystem", TestingUtils.testConfig))
-
- override def afterAll(): Unit = {
- TestKit.shutdownActorSystem(system)
- }
+class TaskManagerLossFailsTasksTest extends WordSpecLike with Matchers {
"A task manager loss" must {
"fail the assigned tasks" in {
try {
- val tm1 = system.actorOf(Props(classOf[SimpleAcknowledgingTaskManager], "TaskManager1"))
- val tm2 = system.actorOf(Props(classOf[SimpleAcknowledgingTaskManager], "TaskManager2"))
-
- val instance1 = ExecutionGraphTestUtils.getInstance(tm1, 10)
- val instance2 = ExecutionGraphTestUtils.getInstance(tm2, 10)
+ val instance1 = ExecutionGraphTestUtils.getInstance(
+ new SimpleInstanceGateway(TestingUtils.defaultExecutionContext), 10)
+ val instance2 = ExecutionGraphTestUtils.getInstance(
+ new SimpleInstanceGateway(TestingUtils.defaultExecutionContext), 10)
- val scheduler = new Scheduler
+ val scheduler = new Scheduler(TestingUtils.defaultExecutionContext)
scheduler.newInstanceAvailable(instance1)
scheduler.newInstanceAvailable(instance2)
@@ -59,7 +48,11 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
val jobGraph = new JobGraph("Pointwise job", sender)
- val eg = new ExecutionGraph(new JobID(), "test job", new Configuration(),
+ val eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext,
+ new JobID(),
+ "test job",
+ new Configuration(),
AkkaUtils.getDefaultTimeout)
eg.setNumberOfRetriesLeft(0)
eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources)
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
index 766ea55..f7bf56a 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
@@ -177,12 +177,12 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
jm ! NotifyWhenJobStatus(jobGraph.getJobID, JobStatus.RESTARTING)
jm ! RequestWorkingTaskManager(jobGraph.getJobID)
- val WorkingTaskManager(tm) = expectMsgType[WorkingTaskManager]
+ val WorkingTaskManager(gatewayOption) = expectMsgType[WorkingTaskManager]
- tm match {
- case ActorRef.noSender => fail("There has to be at least one task manager on which" +
+ gatewayOption match {
+ case None => fail("There has to be at least one task manager on which" +
"the tasks are running.")
- case t => t ! PoisonPill
+ case Some(gateway) => gateway.tell(PoisonPill)
}
expectMsg(JobStatusIs(jobGraph.getJobID, JobStatus.RESTARTING))
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
index 11e93d6..219e5ae 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
@@ -60,16 +60,34 @@ class TestingCluster(userConfiguration: Configuration,
override def startJobManager(actorSystem: ActorSystem): ActorRef = {
- val (instanceManager, scheduler, libraryCacheManager, _, accumulatorManager,
- executionRetries, delayBetweenRetries,
- timeout, archiveCount) = JobManager.createJobManagerComponents(configuration)
+ val (executionContext,
+ instanceManager,
+ scheduler,
+ libraryCacheManager,
+ _,
+ accumulatorManager,
+ executionRetries,
+ delayBetweenRetries,
+ timeout,
+ archiveCount) = JobManager.createJobManagerComponents(configuration)
val testArchiveProps = Props(new MemoryArchivist(archiveCount) with TestingMemoryArchivist)
val archive = actorSystem.actorOf(testArchiveProps, JobManager.ARCHIVE_NAME)
- val jobManagerProps = Props(new JobManager(configuration, instanceManager, scheduler,
- libraryCacheManager, archive, accumulatorManager, executionRetries,
- delayBetweenRetries, timeout, streamingMode) with TestingJobManager)
+ val jobManagerProps = Props(
+ new JobManager(
+ configuration,
+ executionContext,
+ instanceManager,
+ scheduler,
+ libraryCacheManager,
+ archive,
+ accumulatorManager,
+ executionRetries,
+ delayBetweenRetries,
+ timeout,
+ streamingMode)
+ with TestingJobManager)
actorSystem.actorOf(jobManagerProps, JobManager.JOB_MANAGER_NAME)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
index 89e1d72..5747b7e 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
@@ -106,11 +106,10 @@ trait TestingJobManager extends ActorLogMessages with WrapAsScala {
case NotifyWhenJobRemoved(jobID) =>
- val tms = instanceManager.getAllRegisteredInstances.map(_.getTaskManager)
+ val gateways = instanceManager.getAllRegisteredInstances.map(_.getInstanceGateway)
- val responses = tms.map{
- tm =>
- (tm ? NotifyWhenJobRemoved(jobID))(timeout).mapTo[Boolean]
+ val responses = gateways.map{
+ gateway => gateway.ask(NotifyWhenJobRemoved(jobID), timeout).mapTo[Boolean]
}
import context.dispatcher
@@ -135,17 +134,17 @@ trait TestingJobManager extends ActorLogMessages with WrapAsScala {
currentJobs.get(jobID) match {
case Some((eg, _)) =>
if(eg.getAllExecutionVertices.isEmpty){
- sender ! WorkingTaskManager(ActorRef.noSender)
+ sender ! WorkingTaskManager(None)
} else {
val resource = eg.getAllExecutionVertices.head.getCurrentAssignedResource
if(resource == null){
- sender ! WorkingTaskManager(ActorRef.noSender)
+ sender ! WorkingTaskManager(None)
} else {
- sender ! WorkingTaskManager(resource.getInstance().getTaskManager)
+ sender ! WorkingTaskManager(Some(resource.getInstance().getInstanceGateway))
}
}
- case None => sender ! WorkingTaskManager(ActorRef.noSender)
+ case None => sender ! WorkingTaskManager(None)
}
case NotifyWhenJobStatus(jobID, state) =>
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
index f810749..241c6c0 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.testingUtils
import akka.actor.ActorRef
import org.apache.flink.api.common.JobID
import org.apache.flink.runtime.executiongraph.ExecutionGraph
+import org.apache.flink.runtime.instance.InstanceGateway
import org.apache.flink.runtime.jobgraph.JobStatus
object TestingJobManagerMessages {
@@ -43,7 +44,7 @@ object TestingJobManagerMessages {
case class NotifyWhenJobRemoved(jobID: JobID)
case class RequestWorkingTaskManager(jobID: JobID)
- case class WorkingTaskManager(taskManager: ActorRef)
+ case class WorkingTaskManager(gatewayOption: Option[InstanceGateway])
case class NotifyWhenJobStatus(jobID: JobID, state: JobStatus)
case class JobStatusIs(jobID: JobID, state: JobStatus)
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
index 3611633..914f37c 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
@@ -18,14 +18,12 @@
package org.apache.flink.runtime.testingUtils
-import akka.actor.{ActorRef, ActorSystem}
-import akka.testkit.CallingThreadDispatcher
+import com.google.common.util.concurrent.MoreExecutors
import com.typesafe.config.ConfigFactory
import org.apache.flink.configuration.{ConfigConstants, Configuration}
import org.apache.flink.runtime.akka.AkkaUtils
-import org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.ActionQueue
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext
@@ -67,19 +65,33 @@ object TestingUtils {
new TestingCluster(config)
}
- def setGlobalExecutionContext(): Unit = {
- AkkaUtils.globalExecutionContext = ExecutionContext.global
+ /** Returns the global [[ExecutionContext]] which is a [[scala.concurrent.forkjoin.ForkJoinPool]]
+ * with a default parallelism equal to the number of available cores.
+ *
+ * @return ExecutionContext.global
+ */
+ def defaultExecutionContext = ExecutionContext.global
+
+ /** Returns an [[ExecutionContext]] which uses the current thread to execute the runnable.
+ *
+ * @return Direct [[ExecutionContext]] which executes runnables directly
+ */
+ def directExecutionContext = ExecutionContext.fromExecutor(MoreExecutors.directExecutor())
+
+ /** @return A new [[QueuedActionExecutionContext]] */
+ def queuedActionExecutionContext = {
+ new QueuedActionExecutionContext(new ActionQueue())
}
- def setCallingThreadDispatcher(system: ActorSystem): Unit = {
- AkkaUtils.globalExecutionContext = system.dispatchers.lookup(CallingThreadDispatcher.Id)
- }
-
- def setExecutionContext(context: ExecutionContext): Unit = {
- AkkaUtils.globalExecutionContext = context
- }
+ /** [[ExecutionContext]] which queues [[Runnable]] up in an [[ActionQueue]] instead of
+ * execution them. If the automatic execution mode is activated, then the [[Runnable]] are
+ * executed.
+ *
+ * @param actionQueue
+ */
+ class QueuedActionExecutionContext private[testingUtils] (val actionQueue: ActionQueue)
+ extends ExecutionContext {
- class QueuedActionExecutionContext(queue: ActionQueue) extends ExecutionContext {
var automaticExecution = false
def toggleAutomaticExecution() = {
@@ -90,7 +102,7 @@ object TestingUtils {
if(automaticExecution){
runnable.run()
}else {
- queue.queueAction(runnable)
+ actionQueue.queueAction(runnable)
}
}
@@ -98,4 +110,26 @@ object TestingUtils {
t.printStackTrace()
}
}
+
+ /** Queue which stores [[Runnable]] */
+ class ActionQueue {
+ private val runnables = scala.collection.mutable.Queue[Runnable]()
+
+ def triggerNextAction {
+ val r = runnables.dequeue
+ r.run()
+ }
+
+ def popNextAction: Runnable = {
+ runnables.dequeue()
+ }
+
+ def queueAction(r: Runnable) {
+ runnables.enqueue(r)
+ }
+
+ def isEmpty: Boolean = {
+ runnables.isEmpty
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java b/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
index a0db4d6..e108970 100644
--- a/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
@@ -30,8 +30,8 @@ import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.StreamingMode;
-import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.hadoop.fs.FileSystem;
import org.junit.Assert;
import org.slf4j.Logger;
@@ -148,7 +148,7 @@ public class TestBaseUtils {
}
Future<Iterable<Object>> bcVariableManagerFutureResponses = Futures.sequence(
- bcVariableManagerResponseFutures, AkkaUtils.globalExecutionContext());
+ bcVariableManagerResponseFutures, TestingUtils.defaultExecutionContext());
Iterable<Object> responses = Await.result(bcVariableManagerFutureResponses, timeout);
@@ -158,7 +158,7 @@ public class TestBaseUtils {
}
Future<Iterable<Object>> numActiveConnectionsFutureResponses = Futures.sequence(
- numActiveConnectionsResponseFutures, AkkaUtils.globalExecutionContext());
+ numActiveConnectionsResponseFutures, TestingUtils.defaultExecutionContext());
responses = Await.result(numActiveConnectionsFutureResponses, timeout);
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
index 796ea09..0534178 100644
--- a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
+++ b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
@@ -80,16 +80,38 @@ class ForkableFlinkMiniCluster(userConfiguration: Configuration,
override def startJobManager(actorSystem: ActorSystem): ActorRef = {
- val (instanceManager, scheduler, libraryCacheManager, _, accumulatorManager,
- executionRetries, delayBetweenRetries,
- timeout, archiveCount) = JobManager.createJobManagerComponents(configuration)
+ val (executionContext,
+ instanceManager,
+ scheduler,
+ libraryCacheManager,
+ _,
+ accumulatorManager,
+ executionRetries,
+ delayBetweenRetries,
+ timeout,
+ archiveCount) = JobManager.createJobManagerComponents(configuration)
+
+ val testArchiveProps = Props(
+ new MemoryArchivist(
+ archiveCount)
+ with TestingMemoryArchivist)
- val testArchiveProps = Props(new MemoryArchivist(archiveCount) with TestingMemoryArchivist)
val archive = actorSystem.actorOf(testArchiveProps, JobManager.ARCHIVE_NAME)
- val jobManagerProps = Props(new JobManager(configuration, instanceManager, scheduler,
- libraryCacheManager, archive, accumulatorManager, executionRetries,
- delayBetweenRetries, timeout, streamingMode) with TestingJobManager)
+ val jobManagerProps = Props(
+ new JobManager(
+ configuration,
+ executionContext,
+ instanceManager,
+ scheduler,
+ libraryCacheManager,
+ archive,
+ accumulatorManager,
+ executionRetries,
+ delayBetweenRetries,
+ timeout,
+ streamingMode)
+ with TestingJobManager)
val jobManager = actorSystem.actorOf(jobManagerProps, JobManager.JOB_MANAGER_NAME)
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala
index 1884fab..c20f621 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala
@@ -110,9 +110,15 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
jm ! RequestWorkingTaskManager(jobID)
- val tm = expectMsgType[WorkingTaskManager].taskManager
- // kill one task manager
- tm ! PoisonPill
+ val gatewayOption = expectMsgType[WorkingTaskManager].gatewayOption
+
+ gatewayOption match {
+ case Some(gateway) =>
+ // kill one task manager
+ gateway.tell(PoisonPill)
+
+ case None => fail("Could not retrieve a working task manager.")
+ }
val failure = expectMsgType[Failure]
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/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 5dd197d..8cfeead 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
@@ -235,16 +235,34 @@ object ApplicationMaster {
// start all the components inside the job manager
LOG.debug("Starting JobManager components")
- val (instanceManager, scheduler, libraryCacheManager, archiveProps, accumulatorManager,
- executionRetries, delayBetweenRetries,
- timeout, _) = JobManager.createJobManagerComponents(configuration)
+ val (executionContext,
+ instanceManager,
+ scheduler,
+ libraryCacheManager,
+ archiveProps,
+ accumulatorManager,
+ executionRetries,
+ delayBetweenRetries,
+ timeout,
+ _) = JobManager.createJobManagerComponents(configuration)
// start the archiver
val archiver: ActorRef = jobManagerSystem.actorOf(archiveProps, JobManager.ARCHIVE_NAME)
- val jobManagerProps = Props(new JobManager(configuration, instanceManager, scheduler,
- libraryCacheManager, archiver, accumulatorManager, executionRetries,
- delayBetweenRetries, timeout, streamingMode) with ApplicationMasterActor)
+ val jobManagerProps = Props(
+ new JobManager(
+ configuration,
+ executionContext,
+ instanceManager,
+ scheduler,
+ libraryCacheManager,
+ archiver,
+ accumulatorManager,
+ executionRetries,
+ delayBetweenRetries,
+ timeout,
+ streamingMode)
+ with ApplicationMasterActor)
LOG.debug("Starting JobManager actor")
val jobManager = JobManager.startActor(jobManagerProps, jobManagerSystem)
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/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 411808b..3fb5e30 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
@@ -103,7 +103,7 @@ trait ApplicationMasterActor extends ActorLogMessages {
instanceManager.getAllRegisteredInstances.asScala foreach {
instance =>
- instance.getTaskManager ! StopYarnSession(status, diag)
+ instance.getInstanceGateway.tell(StopYarnSession(status, diag))
}
rmClientOption foreach {
[7/7] flink git commit: [FLINK-2329] [runtime] Introduces
InstanceGateway as an abstraction to communicate with the TaskManager.
Posted by se...@apache.org.
[FLINK-2329] [runtime] Introduces InstanceGateway as an abstraction to communicate with the TaskManager.
Replaces AkkaUtils.globalExecutionContext with instance dependent ExecutionContext.
This closes #893
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/2ccb5fdb
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/2ccb5fdb
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/2ccb5fdb
Branch: refs/heads/master
Commit: 2ccb5fdb47aa0e3766fd7fbd17a41feaca29fcbc
Parents: aa5e5b3
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue Jul 7 11:41:44 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Jul 13 17:54:31 2015 +0200
----------------------------------------------------------------------
.../flink/runtime/executiongraph/Execution.java | 89 ++-
.../runtime/executiongraph/ExecutionGraph.java | 55 +-
.../runtime/executiongraph/ExecutionVertex.java | 40 +-
.../runtime/instance/AkkaInstanceGateway.java | 111 ++++
.../apache/flink/runtime/instance/Instance.java | 34 +-
.../flink/runtime/instance/InstanceGateway.java | 82 +++
.../flink/runtime/instance/InstanceManager.java | 3 +-
.../runtime/io/network/NetworkEnvironment.java | 34 +-
.../runtime/jobmanager/scheduler/Scheduler.java | 11 +-
.../jobmanager/web/SetupInfoServlet.java | 3 +-
.../web-docs-infoserver/js/taskmanager.js | 2 +-
.../apache/flink/runtime/akka/AkkaUtils.scala | 2 -
.../flink/runtime/jobmanager/JobManager.scala | 95 +++-
.../runtime/minicluster/FlinkMiniCluster.scala | 20 +-
.../flink/runtime/taskmanager/TaskManager.scala | 34 +-
.../executiongraph/AllVerticesIteratorTest.java | 2 +
.../ExecutionGraphConstructionTest.java | 54 +-
.../ExecutionGraphDeploymentTest.java | 59 +-
.../executiongraph/ExecutionGraphTestUtils.java | 121 ++--
.../ExecutionStateProgressTest.java | 31 +-
.../ExecutionVertexCancelTest.java | 563 ++++++++-----------
.../ExecutionVertexDeploymentTest.java | 115 +---
.../ExecutionVertexSchedulingTest.java | 35 +-
.../executiongraph/LocalInputSplitsTest.java | 53 +-
.../executiongraph/PointwisePatternTest.java | 50 +-
.../TerminalStateDeadlockTest.java | 10 +-
.../VertexLocationConstraintTest.java | 91 +--
.../executiongraph/VertexSlotSharingTest.java | 7 +-
.../instance/BaseTestingInstanceGateway.java | 94 ++++
.../runtime/instance/DummyInstanceGateway.java | 57 ++
.../flink/runtime/instance/InstanceTest.java | 7 +-
.../flink/runtime/instance/SimpleSlotTest.java | 4 +-
.../io/network/NetworkEnvironmentTest.java | 6 +-
.../ScheduleWithCoLocationHintTest.java | 37 +-
.../scheduler/SchedulerIsolatedTasksTest.java | 31 +-
.../scheduler/SchedulerSlotSharingTest.java | 51 +-
.../scheduler/SchedulerTestUtils.java | 4 +-
...askManagerComponentsStartupShutdownTest.java | 6 +-
.../ExecutionGraphRestartTest.scala | 40 +-
.../TaskManagerLossFailsTasksTest.scala | 33 +-
.../runtime/jobmanager/RecoveryITCase.scala | 8 +-
.../runtime/testingUtils/TestingCluster.scala | 30 +-
.../testingUtils/TestingJobManager.scala | 15 +-
.../TestingJobManagerMessages.scala | 3 +-
.../runtime/testingUtils/TestingUtils.scala | 62 +-
.../apache/flink/test/util/TestBaseUtils.java | 6 +-
.../test/util/ForkableFlinkMiniCluster.scala | 36 +-
.../taskmanager/TaskManagerFailsITCase.scala | 12 +-
.../apache/flink/yarn/ApplicationMaster.scala | 30 +-
.../flink/yarn/ApplicationMasterActor.scala | 2 +-
50 files changed, 1399 insertions(+), 981 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/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 76a58e8..af67c3f 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
@@ -18,13 +18,9 @@
package org.apache.flink.runtime.executiongraph;
-import akka.actor.ActorRef;
import akka.dispatch.OnComplete;
import akka.dispatch.OnFailure;
-import akka.pattern.Patterns;
-import akka.util.Timeout;
import org.apache.flink.runtime.JobException;
-import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.deployment.InputChannelDeploymentDescriptor;
import org.apache.flink.runtime.deployment.PartialInputChannelDeploymentDescriptor;
import org.apache.flink.runtime.deployment.ResultPartitionLocation;
@@ -32,6 +28,7 @@ import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.InstanceConnectionInfo;
+import org.apache.flink.runtime.instance.InstanceGateway;
import org.apache.flink.runtime.instance.SimpleSlot;
import org.apache.flink.runtime.io.network.ConnectionID;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
@@ -50,6 +47,7 @@ import org.apache.flink.runtime.taskmanager.Task;
import org.apache.flink.runtime.util.SerializedValue;
import org.apache.flink.util.ExceptionUtils;
import org.slf4j.Logger;
+import scala.concurrent.ExecutionContext;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
@@ -131,9 +129,20 @@ public class Execution implements Serializable {
private SerializedValue<StateHandle<?>> operatorState;
+ /** The execution context which is used to execute futures. */
+ @SuppressWarnings("NonSerializableFieldInSerializableClass")
+ private ExecutionContext executionContext;
+
// --------------------------------------------------------------------------------------------
- public Execution(ExecutionVertex vertex, int attemptNumber, long startTimestamp, FiniteDuration timeout) {
+ public Execution(
+ ExecutionContext executionContext,
+ ExecutionVertex vertex,
+ int attemptNumber,
+ long startTimestamp,
+ FiniteDuration timeout) {
+ this.executionContext = checkNotNull(executionContext);
+
this.vertex = checkNotNull(vertex);
this.attemptId = new ExecutionAttemptID();
@@ -200,6 +209,8 @@ public class Execution implements Serializable {
}
assignedResource = null;
+ executionContext = null;
+
partialInputChannelDeploymentDescriptors.clear();
partialInputChannelDeploymentDescriptors = null;
}
@@ -338,8 +349,9 @@ public class Execution implements Serializable {
vertex.getExecutionGraph().registerExecution(this);
final Instance instance = slot.getInstance();
- final Future<Object> deployAction = Patterns.ask(instance.getTaskManager(),
- new SubmitTask(deployment), new Timeout(timeout));
+ final InstanceGateway gateway = instance.getInstanceGateway();
+
+ final Future<Object> deployAction = gateway.ask(new SubmitTask(deployment), timeout);
deployAction.onComplete(new OnComplete<Object>(){
@@ -366,7 +378,7 @@ public class Execution implements Serializable {
}
}
}
- }, AkkaUtils.globalExecutionContext());
+ }, executionContext);
}
catch (Throwable t) {
markFailed(t);
@@ -402,7 +414,7 @@ public class Execution implements Serializable {
else if (current == FINISHED || current == FAILED) {
// nothing to do any more. finished failed before it could be cancelled.
// in any case, the task is removed from the TaskManager already
- sendFailIntermediateResultPartitionsRPCCall();
+ sendFailIntermediateResultPartitionsRpcCall();
return;
}
@@ -485,7 +497,7 @@ public class Execution implements Serializable {
return true;
}
- }, AkkaUtils.globalExecutionContext());
+ }, executionContext);
// double check to resolve race conditions
if(consumerVertex.getExecutionState() == RUNNING){
@@ -533,7 +545,7 @@ public class Execution implements Serializable {
final UpdatePartitionInfo updateTaskMessage = new UpdateTaskSinglePartitionInfo(
consumer.getAttemptId(), partition.getIntermediateResult().getId(), descriptor);
- sendUpdateTaskRpcCall(consumerSlot, updateTaskMessage);
+ sendUpdatePartitionInfoRpcCall(consumerSlot, updateTaskMessage);
}
// ----------------------------------------------------------------
// Consumer is scheduled or deploying => cache input channel
@@ -689,11 +701,12 @@ public class Execution implements Serializable {
inputChannelDeploymentDescriptors.add(partialInputChannelDeploymentDescriptor.createInputChannelDeploymentDescriptor(this));
}
- UpdatePartitionInfo updateTaskMessage =
- createUpdateTaskMultiplePartitionInfos(attemptId, resultIDs,
- inputChannelDeploymentDescriptors);
+ UpdatePartitionInfo updateTaskMessage = createUpdateTaskMultiplePartitionInfos(
+ attemptId,
+ resultIDs,
+ inputChannelDeploymentDescriptors);
- sendUpdateTaskRpcCall(assignedResource, updateTaskMessage);
+ sendUpdatePartitionInfoRpcCall(assignedResource, updateTaskMessage);
}
}
@@ -804,14 +817,23 @@ public class Execution implements Serializable {
}
}
+ /**
+ * This method sends a CancelTask message to the instance of the assigned slot.
+ *
+ * The sending is tried up to NUM_CANCEL_CALL_TRIES times.
+ */
private void sendCancelRpcCall() {
final SimpleSlot slot = this.assignedResource;
if (slot != null) {
- Future<Object> cancelResult = AkkaUtils.retry(slot.getInstance().getTaskManager(), new
- CancelTask(attemptId), NUM_CANCEL_CALL_TRIES,
- AkkaUtils.globalExecutionContext(), timeout);
+ final InstanceGateway gateway = slot.getInstance().getInstanceGateway();
+
+ Future<Object> cancelResult = gateway.retry(
+ new CancelTask(attemptId),
+ NUM_CANCEL_CALL_TRIES,
+ timeout,
+ executionContext);
cancelResult.onComplete(new OnComplete<Object>() {
@@ -827,35 +849,40 @@ public class Execution implements Serializable {
}
}
}
- }, AkkaUtils.globalExecutionContext());
+ }, executionContext);
}
}
- private void sendFailIntermediateResultPartitionsRPCCall() {
+ private void sendFailIntermediateResultPartitionsRpcCall() {
final SimpleSlot slot = this.assignedResource;
if (slot != null) {
final Instance instance = slot.getInstance();
if (instance.isAlive()) {
- try {
- // TODO For some tests this could be a problem when querying too early if all resources were released
- instance.getTaskManager().tell(new FailIntermediateResultPartitions(attemptId), ActorRef.noSender());
- } catch (Throwable t) {
- fail(new Exception("Intermediate result partition could not be failed.", t));
- }
+ final InstanceGateway gateway = instance.getInstanceGateway();
+
+ // TODO For some tests this could be a problem when querying too early if all resources were released
+ gateway.tell(new FailIntermediateResultPartitions(attemptId));
}
}
}
- private void sendUpdateTaskRpcCall(final SimpleSlot consumerSlot,
- final UpdatePartitionInfo updateTaskMsg) {
+ /**
+ * Sends an UpdatePartitionInfo message to the instance of the consumerSlot.
+ *
+ * @param consumerSlot Slot to whose instance the message will be sent
+ * @param updatePartitionInfo UpdatePartitionInfo message
+ */
+ private void sendUpdatePartitionInfoRpcCall(
+ final SimpleSlot consumerSlot,
+ final UpdatePartitionInfo updatePartitionInfo) {
if (consumerSlot != null) {
final Instance instance = consumerSlot.getInstance();
+ final InstanceGateway gateway = instance.getInstanceGateway();
- Future<Object> futureUpdate = Patterns.ask(instance.getTaskManager(), updateTaskMsg,
- new Timeout(timeout));
+ Future<Object> futureUpdate = gateway.ask(updatePartitionInfo, timeout);
futureUpdate.onFailure(new OnFailure() {
@Override
@@ -863,7 +890,7 @@ public class Execution implements Serializable {
fail(new IllegalStateException("Update task on instance " + instance +
" failed due to:", failure));
}
- }, AkkaUtils.globalExecutionContext());
+ }, executionContext);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/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 84cbab7..47b7ae2 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
@@ -24,7 +24,6 @@ import akka.actor.ActorSystem;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.JobException;
-import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
import org.apache.flink.runtime.execution.ExecutionState;
@@ -45,6 +44,7 @@ import org.apache.flink.util.InstantiationUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import scala.concurrent.ExecutionContext;
import scala.concurrent.duration.FiniteDuration;
import java.io.Serializable;
@@ -197,6 +197,10 @@ public class ExecutionGraph implements Serializable {
@SuppressWarnings("NonSerializableFieldInSerializableClass")
private CheckpointCoordinator checkpointCoordinator;
+ /** The execution context which is used to execute futures. */
+ @SuppressWarnings("NonSerializableFieldInSerializableClass")
+ private ExecutionContext executionContext;
+
// ------ Fields that are only relevant for archived execution graphs ------------
private ExecutionConfig executionConfig;
@@ -207,17 +211,38 @@ public class ExecutionGraph implements Serializable {
/**
* This constructor is for tests only, because it does not include class loading information.
*/
- ExecutionGraph(JobID jobId, String jobName, Configuration jobConfig, FiniteDuration timeout) {
- this(jobId, jobName, jobConfig, timeout, new ArrayList<BlobKey>(), ExecutionGraph.class.getClassLoader());
- }
-
- public ExecutionGraph(JobID jobId, String jobName, Configuration jobConfig, FiniteDuration timeout,
- List<BlobKey> requiredJarFiles, ClassLoader userClassLoader) {
-
- if (jobId == null || jobName == null || jobConfig == null || userClassLoader == null) {
+ ExecutionGraph(
+ ExecutionContext executionContext,
+ JobID jobId,
+ String jobName,
+ Configuration jobConfig,
+ FiniteDuration timeout) {
+ this(
+ executionContext,
+ jobId,
+ jobName,
+ jobConfig,
+ timeout,
+ new ArrayList<BlobKey>(),
+ ExecutionGraph.class.getClassLoader()
+ );
+ }
+
+ public ExecutionGraph(
+ ExecutionContext executionContext,
+ JobID jobId,
+ String jobName,
+ Configuration jobConfig,
+ FiniteDuration timeout,
+ List<BlobKey> requiredJarFiles,
+ ClassLoader userClassLoader) {
+
+ if (executionContext == null || jobId == null || jobName == null || jobConfig == null || userClassLoader == null) {
throw new NullPointerException();
}
+ this.executionContext = executionContext;
+
this.jobID = jobId;
this.jobName = jobName;
this.jobConfiguration = jobConfig;
@@ -451,6 +476,15 @@ public class ExecutionGraph implements Serializable {
return this.stateTimestamps[status.ordinal()];
}
+ /**
+ * Returns the ExecutionContext associated with this ExecutionGraph.
+ *
+ * @return ExecutionContext associated with this ExecutionGraph
+ */
+ public ExecutionContext getExecutionContext() {
+ return executionContext;
+ }
+
// --------------------------------------------------------------------------------------------
// Actions
// --------------------------------------------------------------------------------------------
@@ -629,6 +663,7 @@ public class ExecutionGraph implements Serializable {
userClassLoader = null;
scheduler = null;
checkpointCoordinator = null;
+ executionContext = null;
for (ExecutionJobVertex vertex : verticesInCreationOrder) {
vertex.prepareForArchiving();
@@ -719,7 +754,7 @@ public class ExecutionGraph implements Serializable {
restart();
return null;
}
- }, AkkaUtils.globalExecutionContext());
+ }, executionContext);
break;
}
else if (numberOfRetriesLeft <= 0 && transitionState(current, JobStatus.FAILED, failureCause)) {
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
index a70fa7d..f9001cf 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
@@ -18,8 +18,6 @@
package org.apache.flink.runtime.executiongraph;
-import akka.actor.ActorRef;
-
import org.apache.flink.runtime.JobException;
import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.deployment.InputChannelDeploymentDescriptor;
@@ -30,6 +28,7 @@ import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.InstanceConnectionInfo;
+import org.apache.flink.runtime.instance.InstanceGateway;
import org.apache.flink.runtime.instance.SimpleSlot;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
@@ -101,14 +100,20 @@ public class ExecutionVertex implements Serializable {
// --------------------------------------------------------------------------------------------
- public ExecutionVertex(ExecutionJobVertex jobVertex, int subTaskIndex,
- IntermediateResult[] producedDataSets, FiniteDuration timeout) {
+ public ExecutionVertex(
+ ExecutionJobVertex jobVertex,
+ int subTaskIndex,
+ IntermediateResult[] producedDataSets,
+ FiniteDuration timeout) {
this(jobVertex, subTaskIndex, producedDataSets, timeout, System.currentTimeMillis());
}
- public ExecutionVertex(ExecutionJobVertex jobVertex, int subTaskIndex,
- IntermediateResult[] producedDataSets, FiniteDuration timeout,
- long createTimestamp) {
+ public ExecutionVertex(
+ ExecutionJobVertex jobVertex,
+ int subTaskIndex,
+ IntermediateResult[] producedDataSets,
+ FiniteDuration timeout,
+ long createTimestamp) {
this.jobVertex = jobVertex;
this.subTaskIndex = subTaskIndex;
@@ -125,7 +130,12 @@ public class ExecutionVertex implements Serializable {
this.priorExecutions = new CopyOnWriteArrayList<Execution>();
- this.currentExecution = new Execution(this, 0, createTimestamp, timeout);
+ this.currentExecution = new Execution(
+ getExecutionGraph().getExecutionContext(),
+ this,
+ 0,
+ createTimestamp,
+ timeout);
// create a co-location scheduling hint, if necessary
CoLocationGroup clg = jobVertex.getCoLocationGroup();
@@ -416,8 +426,12 @@ public class ExecutionVertex implements Serializable {
if (state == FINISHED || state == CANCELED || state == FAILED) {
priorExecutions.add(execution);
- currentExecution = new Execution(this, execution.getAttemptNumber()+1,
- System.currentTimeMillis(), timeout);
+ currentExecution = new Execution(
+ getExecutionGraph().getExecutionContext(),
+ this,
+ execution.getAttemptNumber()+1,
+ System.currentTimeMillis(),
+ timeout);
CoLocationGroup grp = jobVertex.getCoLocationGroup();
if (grp != null) {
@@ -455,9 +469,9 @@ public class ExecutionVertex implements Serializable {
// send only if we actually have a target
if (slot != null) {
- ActorRef taskManager = slot.getInstance().getTaskManager();
- if (taskManager != null) {
- taskManager.tell(message, ActorRef.noSender());
+ InstanceGateway gateway = slot.getInstance().getInstanceGateway();
+ if (gateway != null) {
+ gateway.tell(message);
}
}
else {
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/main/java/org/apache/flink/runtime/instance/AkkaInstanceGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/AkkaInstanceGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/AkkaInstanceGateway.java
new file mode 100644
index 0000000..b7d60c5
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/AkkaInstanceGateway.java
@@ -0,0 +1,111 @@
+/*
+ * 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.instance;
+
+import akka.actor.ActorRef;
+import akka.pattern.Patterns;
+import akka.util.Timeout;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import scala.concurrent.ExecutionContext;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+
+/**
+ * InstanceGateway implementation which uses Akka to communicate with remote instances.
+ */
+public class AkkaInstanceGateway implements InstanceGateway {
+
+ /** ActorRef of the remote instance */
+ private final ActorRef taskManager;
+
+ public AkkaInstanceGateway(ActorRef taskManager) {
+ this.taskManager = taskManager;
+ }
+
+ /**
+ * Sends a message asynchronously and returns its response. The response to the message is
+ * returned as a future.
+ *
+ * @param message Message to be sent
+ * @param timeout Timeout until the Future is completed with an AskTimeoutException
+ * @return Future which contains the response to the sent message
+ */
+ @Override
+ public Future<Object> ask(Object message, FiniteDuration timeout) {
+ return Patterns.ask(taskManager, message, new Timeout(timeout));
+ }
+
+ /**
+ * Sends a message asynchronously without a result.
+ *
+ * @param message Message to be sent
+ */
+ @Override
+ public void tell(Object message) {
+ taskManager.tell(message, ActorRef.noSender());
+ }
+
+ /**
+ * Forwards a message. For the receiver of this message it looks as if sender has sent the
+ * message.
+ *
+ * @param message Message to be sent
+ * @param sender Sender of the forwarded message
+ */
+ @Override
+ public void forward(Object message, ActorRef sender) {
+ taskManager.tell(message, sender);
+ }
+
+ /**
+ * Retries to send asynchronously a message up to numberRetries times. The response to this
+ * message is returned as a future. The message is re-sent if the number of retries is not yet
+ * exceeded and if an exception occurred while sending it.
+ *
+ * @param message Message to be sent
+ * @param numberRetries Number of times to retry sending the message
+ * @param timeout Timeout for each sending attempt
+ * @param executionContext ExecutionContext which is used to send the message multiple times
+ * @return Future of the response to the sent message
+ */
+ @Override
+ public Future<Object> retry(
+ Object message,
+ int numberRetries,
+ FiniteDuration timeout,
+ ExecutionContext executionContext) {
+
+ return AkkaUtils.retry(
+ taskManager,
+ message,
+ numberRetries,
+ executionContext,
+ timeout);
+ }
+
+ /**
+ * Returns the ActorPath of the remote instance.
+ *
+ * @return ActorPath of the remote instance.
+ */
+ @Override
+ public String path() {
+ return taskManager.path().toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
index 39caf08..1c44b5d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
@@ -25,8 +25,6 @@ import java.util.List;
import java.util.Queue;
import java.util.Set;
-import akka.actor.ActorRef;
-
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobmanager.scheduler.SlotAvailabilityListener;
import org.slf4j.Logger;
@@ -43,8 +41,8 @@ public class Instance {
/** The lock on which to synchronize allocations and failure state changes */
private final Object instanceLock = new Object();
- /** The actor ref to the task manager represented by this taskManager. */
- private final ActorRef taskManager;
+ /** The instacne gateway to communicate with the instance */
+ private final InstanceGateway instanceGateway;
/** The instance connection information for the data transfer. */
private final InstanceConnectionInfo connectionInfo;
@@ -81,15 +79,19 @@ public class Instance {
/**
* Constructs an instance reflecting a registered TaskManager.
*
- * @param taskManager The actor reference of the represented task manager.
+ * @param instanceGateway The instance gateway to communicate with the remote instance
* @param connectionInfo The remote connection where the task manager receives requests.
* @param id The id under which the taskManager is registered.
* @param resources The resources available on the machine.
* @param numberOfSlots The number of task slots offered by this taskManager.
*/
- public Instance(ActorRef taskManager, InstanceConnectionInfo connectionInfo, InstanceID id,
- HardwareDescription resources, int numberOfSlots) {
- this.taskManager = taskManager;
+ public Instance(
+ InstanceGateway instanceGateway,
+ InstanceConnectionInfo connectionInfo,
+ InstanceID id,
+ HardwareDescription resources,
+ int numberOfSlots) {
+ this.instanceGateway = instanceGateway;
this.connectionInfo = connectionInfo;
this.instanceId = id;
this.resources = resources;
@@ -327,12 +329,14 @@ public class Instance {
}
}
- public ActorRef getTaskManager() {
- return taskManager;
- }
-
- public String getPath(){
- return taskManager.path().toString();
+ /**
+ * Returns the InstanceGateway of this Instance. This gateway can be used to communicate with
+ * it.
+ *
+ * @return InstanceGateway associated with this instance
+ */
+ public InstanceGateway getInstanceGateway() {
+ return instanceGateway;
}
public InstanceConnectionInfo getInstanceConnectionInfo() {
@@ -386,6 +390,6 @@ public class Instance {
@Override
public String toString() {
return String.format("%s @ %s - %d slots - URL: %s", instanceId, connectionInfo.getHostname(),
- numberOfSlots, (taskManager != null ? taskManager.path() : "ActorRef.noSender"));
+ numberOfSlots, (instanceGateway != null ? instanceGateway.path() : "No instance gateway"));
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/main/java/org/apache/flink/runtime/instance/InstanceGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/InstanceGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/InstanceGateway.java
new file mode 100644
index 0000000..a30b2f6
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/InstanceGateway.java
@@ -0,0 +1,82 @@
+/*
+ * 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.instance;
+
+import akka.actor.ActorRef;
+import scala.concurrent.ExecutionContext;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+
+/**
+ * Interface to abstract the communication with an Instance.
+ *
+ * It allows to avoid direct interaction with an ActorRef.
+ */
+public interface InstanceGateway {
+
+ /**
+ * Sends a message asynchronously and returns its response. The response to the message is
+ * returned as a future.
+ *
+ * @param message Message to be sent
+ * @param timeout Timeout until the Future is completed with an AskTimeoutException
+ * @return Future which contains the response to the sent message
+ */
+ Future<Object> ask(Object message, FiniteDuration timeout);
+
+ /**
+ * Sends a message asynchronously without a result.
+ *
+ * @param message Message to be sent
+ */
+ void tell(Object message);
+
+ /**
+ * Forwards a message. For the receiver of this message it looks as if sender has sent the
+ * message.
+ *
+ * @param message Message to be sent
+ * @param sender Sender of the forwarded message
+ */
+ void forward(Object message, ActorRef sender);
+
+ /**
+ * Retries to send asynchronously a message up to numberRetries times. The response to this
+ * message is returned as a future. The message is re-sent if the number of retries is not yet
+ * exceeded and if an exception occurred while sending it.
+ *
+ * @param message Message to be sent
+ * @param numberRetries Number of times to retry sending the message
+ * @param timeout Timeout for each sending attempt
+ * @param executionContext ExecutionContext which is used to send the message multiple times
+ * @return Future of the response to the sent message
+ */
+ Future<Object> retry(
+ Object message,
+ int numberRetries,
+ FiniteDuration timeout,
+ ExecutionContext executionContext);
+
+ /**
+ * Returns the path of the remote instance.
+ *
+ * @return Path of the remote instance.
+ */
+ String path();
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/main/java/org/apache/flink/runtime/instance/InstanceManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/InstanceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/InstanceManager.java
index c1800bd..4f6c7a6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/InstanceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/InstanceManager.java
@@ -149,8 +149,9 @@ public class InstanceManager {
id = new InstanceID();
} while (registeredHostsById.containsKey(id));
+ InstanceGateway instanceGateway = new AkkaInstanceGateway(taskManager);
- Instance host = new Instance(taskManager, connectionInfo, id, resources, numberOfSlots);
+ Instance host = new Instance(instanceGateway, connectionInfo, id, resources, numberOfSlots);
registeredHostsById.put(id, host);
registeredHostsByConnection.put(taskManager, host);
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
index c082c6a..0ffc889 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
@@ -23,7 +23,6 @@ import akka.dispatch.OnFailure;
import akka.pattern.Patterns;
import akka.util.Timeout;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.io.disk.iomanager.IOManager.IOMode;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
@@ -46,6 +45,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Option;
import scala.Tuple2;
+import scala.concurrent.ExecutionContext;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
@@ -89,11 +89,20 @@ public class NetworkEnvironment {
private boolean isShutdown;
/**
+ * ExecutionEnvironment which is used to execute remote calls with the
+ * {@link JobManagerResultPartitionConsumableNotifier}
+ */
+ private final ExecutionContext executionContext;
+
+ /**
* Initializes all network I/O components.
*/
- public NetworkEnvironment(FiniteDuration jobManagerTimeout,
- NetworkEnvironmentConfiguration config) throws IOException {
+ public NetworkEnvironment(
+ ExecutionContext executionContext,
+ FiniteDuration jobManagerTimeout,
+ NetworkEnvironmentConfiguration config) throws IOException {
+ this.executionContext = executionContext;
this.configuration = checkNotNull(config);
this.jobManagerTimeout = checkNotNull(jobManagerTimeout);
@@ -182,7 +191,10 @@ public class NetworkEnvironment {
this.partitionManager = new ResultPartitionManager();
this.taskEventDispatcher = new TaskEventDispatcher();
this.partitionConsumableNotifier = new JobManagerResultPartitionConsumableNotifier(
- jobManagerRef, taskManagerRef, new Timeout(jobManagerTimeout));
+ executionContext,
+ jobManagerRef,
+ taskManagerRef,
+ new Timeout(jobManagerTimeout));
this.partitionStateChecker = new JobManagerPartitionStateChecker(
jobManagerRef, taskManagerRef);
@@ -414,6 +426,12 @@ public class NetworkEnvironment {
*/
private static class JobManagerResultPartitionConsumableNotifier implements ResultPartitionConsumableNotifier {
+ /**
+ * {@link ExecutionContext} which is used for the failure handler of {@link ScheduleOrUpdateConsumers}
+ * messages.
+ */
+ private final ExecutionContext executionContext;
+
private final ActorRef jobManager;
private final ActorRef taskManager;
@@ -421,8 +439,12 @@ public class NetworkEnvironment {
private final Timeout jobManagerMessageTimeout;
public JobManagerResultPartitionConsumableNotifier(
- ActorRef jobManager, ActorRef taskManager, Timeout jobManagerMessageTimeout) {
+ ExecutionContext executionContext,
+ ActorRef jobManager,
+ ActorRef taskManager,
+ Timeout jobManagerMessageTimeout) {
+ this.executionContext = executionContext;
this.jobManager = jobManager;
this.taskManager = taskManager;
this.jobManagerMessageTimeout = jobManagerMessageTimeout;
@@ -448,7 +470,7 @@ public class NetworkEnvironment {
taskManager.tell(failMsg, ActorRef.noSender());
}
- }, AkkaUtils.globalExecutionContext());
+ }, executionContext);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/Scheduler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/Scheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/Scheduler.java
index 940082e..cb99e52 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/Scheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/Scheduler.java
@@ -37,7 +37,6 @@ import akka.dispatch.Futures;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
-import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.instance.SlotSharingGroupAssignment;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.instance.SharedSlot;
@@ -50,6 +49,7 @@ import org.apache.flink.util.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import scala.concurrent.ExecutionContext;
/**
* The scheduler is responsible for distributing the ready-to-run tasks among instances and slots.
@@ -95,12 +95,17 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
/** The number of slot allocations where locality could not be respected */
private int nonLocalizedAssignments;
+ /** The ExecutionContext which is used to execute newSlotAvailable futures. */
+ private final ExecutionContext executionContext;
+
// ------------------------------------------------------------------------
/**
* Creates a new scheduler.
*/
- public Scheduler() {}
+ public Scheduler(ExecutionContext executionContext) {
+ this.executionContext = executionContext;
+ }
/**
* Shuts the scheduler down. After shut down no more tasks can be added to the scheduler.
@@ -519,7 +524,7 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener {
handleNewSlot();
return null;
}
- }, AkkaUtils.globalExecutionContext());
+ }, executionContext);
}
private void handleNewSlot() {
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/SetupInfoServlet.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/SetupInfoServlet.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/SetupInfoServlet.java
index 4e028d4..567d15a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/SetupInfoServlet.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/SetupInfoServlet.java
@@ -146,8 +146,7 @@ public class SetupInfoServlet extends HttpServlet {
long time = new Date().getTime() - instance.getLastHeartBeat();
try {
- objInner.put("inetAdress", instance.getInstanceConnectionInfo().getInetAdress());
- objInner.put("ipcPort", instance.getTaskManager().path().address().hostPort());
+ objInner.put("path", instance.getInstanceGateway().path());
objInner.put("dataPort", instance.getInstanceConnectionInfo().dataPort());
objInner.put("timeSinceLastHeartbeat", time / 1000);
objInner.put("slotsNumber", instance.getTotalNumberOfSlots());
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/main/resources/web-docs-infoserver/js/taskmanager.js
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/resources/web-docs-infoserver/js/taskmanager.js b/flink-runtime/src/main/resources/web-docs-infoserver/js/taskmanager.js
index 1ea9a41..68e4278 100644
--- a/flink-runtime/src/main/resources/web-docs-infoserver/js/taskmanager.js
+++ b/flink-runtime/src/main/resources/web-docs-infoserver/js/taskmanager.js
@@ -224,7 +224,7 @@ function processTMdata(json) {
"</div>";
var content = "<tr id=\""+tmRowIdCssName+"\">" +
- "<td style=\"width:20%\">"+tm.inetAdress+" <br> IPC Port: "+tm.ipcPort+", Data Port: "+tm.dataPort+"</td>" + // first row: TaskManager
+ "<td style=\"width:20%\">"+tm.path+" <br> Data Port: "+tm.dataPort+"</td>" + // first row: TaskManager
"<td id=\""+tmRowIdCssName+"-memory\">"+tmMemoryBox+"</td>" + // second row: memory statistics
"<td id=\""+tmRowIdCssName+"-info\"><i>Loading Information</i></td>" + // Information
"</tr>";
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
index 7ffaddd..d38e503 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
@@ -41,8 +41,6 @@ object AkkaUtils {
val INF_TIMEOUT = 21474835 seconds
- var globalExecutionContext: ExecutionContext = ExecutionContext.global
-
/**
* Creates a local actor system without remoting.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/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 dc1599a..3b4ce15 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
@@ -60,6 +60,7 @@ import org.apache.flink.util.{ExceptionUtils, InstantiationUtil}
import scala.collection.JavaConverters._
import scala.concurrent._
import scala.concurrent.duration._
+import scala.concurrent.forkjoin.ForkJoinPool
import scala.language.postfixOps
/**
@@ -89,16 +90,18 @@ import scala.language.postfixOps
* - [[JobStatusChanged]] indicates that the status of job (RUNNING, CANCELING, FINISHED, etc.) has
* changed. This message is sent by the ExecutionGraph.
*/
-class JobManager(protected val flinkConfiguration: Configuration,
- protected val instanceManager: InstanceManager,
- protected val scheduler: FlinkScheduler,
- protected val libraryCacheManager: BlobLibraryCacheManager,
- protected val archive: ActorRef,
- protected val accumulatorManager: AccumulatorManager,
- protected val defaultExecutionRetries: Int,
- protected val delayBetweenRetries: Long,
- protected val timeout: FiniteDuration,
- protected val mode: StreamingMode)
+class JobManager(
+ protected val flinkConfiguration: Configuration,
+ protected val executionContext: ExecutionContext,
+ protected val instanceManager: InstanceManager,
+ protected val scheduler: FlinkScheduler,
+ protected val libraryCacheManager: BlobLibraryCacheManager,
+ protected val archive: ActorRef,
+ protected val accumulatorManager: AccumulatorManager,
+ protected val defaultExecutionRetries: Int,
+ protected val delayBetweenRetries: Long,
+ protected val timeout: FiniteDuration,
+ protected val mode: StreamingMode)
extends Actor with ActorLogMessages with ActorSynchronousLogging {
/** List of current jobs running jobs */
@@ -117,7 +120,7 @@ class JobManager(protected val flinkConfiguration: Configuration,
// disconnect the registered task managers
instanceManager.getAllRegisteredInstances.asScala.foreach {
- _.getTaskManager ! Disconnect("JobManager is shutting down")
+ _.getInstanceGateway().tell(Disconnect("JobManager is shutting down"))
}
archive ! PoisonPill
@@ -136,7 +139,6 @@ class JobManager(protected val flinkConfiguration: Configuration,
}
log.debug(s"Job manager ${self.path} is completely stopped.")
-
}
/**
@@ -411,8 +413,8 @@ class JobManager(protected val flinkConfiguration: Configuration,
case message: AccumulatorMessage => handleAccumulatorMessage(message)
case RequestStackTrace(instanceID) =>
- val taskManager = instanceManager.getRegisteredInstanceById(instanceID).getTaskManager
- taskManager forward SendStackTrace
+ val gateway = instanceManager.getRegisteredInstanceById(instanceID).getInstanceGateway
+ gateway.forward(SendStackTrace, sender)
case Terminated(taskManager) =>
if (instanceManager.isRegistered(taskManager)) {
@@ -480,10 +482,18 @@ class JobManager(protected val flinkConfiguration: Configuration,
}
// see if there already exists an ExecutionGraph for the corresponding job ID
- executionGraph = currentJobs.getOrElseUpdate(jobGraph.getJobID,
- (new ExecutionGraph(jobGraph.getJobID, jobGraph.getName,
- jobGraph.getJobConfiguration, timeout, jobGraph.getUserJarBlobKeys, userCodeLoader),
- JobInfo(sender(), System.currentTimeMillis())))._1
+ executionGraph = currentJobs.getOrElseUpdate(
+ jobGraph.getJobID,
+ (new ExecutionGraph(
+ executionContext,
+ jobGraph.getJobID,
+ jobGraph.getName,
+ jobGraph.getJobConfiguration,
+ timeout,
+ jobGraph.getUserJarBlobKeys,
+ userCodeLoader),
+ JobInfo(sender(), System.currentTimeMillis()))
+ )._1
// configure the execution graph
val jobNumberRetries = if (jobGraph.getNumberOfExecutionRetries >= 0) {
@@ -1046,8 +1056,8 @@ object JobManager {
* @param configuration The configuration from which to parse the config values.
* @return The members for a default JobManager.
*/
- def createJobManagerComponents(configuration: Configuration) :
- (InstanceManager, FlinkScheduler, BlobLibraryCacheManager,
+ def createJobManagerComponents(configuration: Configuration)
+ : (ExecutionContext, InstanceManager, FlinkScheduler, BlobLibraryCacheManager,
Props, AccumulatorManager, Int, Long, FiniteDuration, Int) = {
val timeout: FiniteDuration = AkkaUtils.getTimeout(configuration)
@@ -1083,6 +1093,8 @@ object JobManager {
val accumulatorManager: AccumulatorManager = new AccumulatorManager(Math.min(1, archiveCount))
+ val executionContext = ExecutionContext.fromExecutor(new ForkJoinPool())
+
var blobServer: BlobServer = null
var instanceManager: InstanceManager = null
var scheduler: FlinkScheduler = null
@@ -1091,7 +1103,7 @@ object JobManager {
try {
blobServer = new BlobServer(configuration)
instanceManager = new InstanceManager()
- scheduler = new FlinkScheduler()
+ scheduler = new FlinkScheduler(executionContext)
libraryCacheManager = new BlobLibraryCacheManager(blobServer, cleanupInterval)
instanceManager.addInstanceListener(scheduler)
@@ -1114,8 +1126,16 @@ object JobManager {
}
}
- (instanceManager, scheduler, libraryCacheManager, archiveProps, accumulatorManager,
- executionRetries, delayBetweenRetries, timeout, archiveCount)
+ (executionContext,
+ instanceManager,
+ scheduler,
+ libraryCacheManager,
+ archiveProps,
+ accumulatorManager,
+ executionRetries,
+ delayBetweenRetries,
+ timeout,
+ archiveCount)
}
/**
@@ -1154,9 +1174,16 @@ object JobManager {
archiverActorName: Option[String],
streamingMode: StreamingMode): (ActorRef, ActorRef) = {
- val (instanceManager, scheduler, libraryCacheManager, archiveProps, accumulatorManager,
- executionRetries, delayBetweenRetries,
- timeout, _) = createJobManagerComponents(configuration)
+ val (executionContext,
+ instanceManager,
+ scheduler,
+ libraryCacheManager,
+ archiveProps,
+ accumulatorManager,
+ executionRetries,
+ delayBetweenRetries,
+ timeout,
+ _) = createJobManagerComponents(configuration)
// start the archiver with the given name, or without (avoid name conflicts)
val archiver: ActorRef = archiverActorName match {
@@ -1164,9 +1191,19 @@ object JobManager {
case None => actorSystem.actorOf(archiveProps)
}
- val jobManagerProps = Props(classOf[JobManager], configuration, instanceManager, scheduler,
- libraryCacheManager, archiver, accumulatorManager, executionRetries,
- delayBetweenRetries, timeout, streamingMode)
+ val jobManagerProps = Props(
+ classOf[JobManager],
+ configuration,
+ executionContext,
+ instanceManager,
+ scheduler,
+ libraryCacheManager,
+ archiver,
+ accumulatorManager,
+ executionRetries,
+ delayBetweenRetries,
+ timeout,
+ streamingMode)
val jobManager: ActorRef = jobMangerActorName match {
case Some(actorName) => actorSystem.actorOf(jobManagerProps, actorName)
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
index 73a37de..49c701e 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
@@ -34,7 +34,7 @@ import org.apache.flink.runtime.messages.TaskManagerMessages.NotifyWhenRegistere
import org.slf4j.LoggerFactory
import scala.concurrent.duration.FiniteDuration
-import scala.concurrent.{Future, Await}
+import scala.concurrent.{ExecutionContext, Future, Await}
/**
* Abstract base class for Flink's mini cluster. The mini cluster starts a
@@ -48,9 +48,10 @@ import scala.concurrent.{Future, Await}
* @param streamingMode True, if the system should be started in streaming mode, false if
* in pure batch mode.
*/
-abstract class FlinkMiniCluster(val userConfiguration: Configuration,
- val singleActorSystem: Boolean,
- val streamingMode: StreamingMode) {
+abstract class FlinkMiniCluster(
+ val userConfiguration: Configuration,
+ val singleActorSystem: Boolean,
+ val streamingMode: StreamingMode) {
def this(userConfiguration: Configuration, singleActorSystem: Boolean)
= this(userConfiguration, singleActorSystem, StreamingMode.BATCH_ONLY)
@@ -157,7 +158,7 @@ abstract class FlinkMiniCluster(val userConfiguration: Configuration,
val future = gracefulStop(jobManagerActor, timeout)
- implicit val executionContext = AkkaUtils.globalExecutionContext
+ implicit val executionContext = ExecutionContext.global
Await.ready(Future.sequence(future +: futures), timeout)
@@ -179,7 +180,7 @@ abstract class FlinkMiniCluster(val userConfiguration: Configuration,
}
def waitForTaskManagersToBeRegistered(): Unit = {
- implicit val executionContext = AkkaUtils.globalExecutionContext
+ implicit val executionContext = ExecutionContext.global
val futures = taskManagerActors map {
taskManager => (taskManager ? NotifyWhenRegisteredAtJobManager)(timeout)
@@ -196,8 +197,11 @@ abstract class FlinkMiniCluster(val userConfiguration: Configuration,
}
@throws(classOf[JobExecutionException])
- def submitJobAndWait(jobGraph: JobGraph, printUpdates: Boolean, timeout: FiniteDuration)
- : SerializedJobExecutionResult = {
+ def submitJobAndWait(
+ jobGraph: JobGraph,
+ printUpdates: Boolean,
+ timeout: FiniteDuration)
+ : SerializedJobExecutionResult = {
val clientActorSystem = if (singleActorSystem) jobManagerActorSystem
else JobClient.startJobClientActorSystem(configuration)
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/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 520decd..44a0b04 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
@@ -65,6 +65,7 @@ import org.apache.flink.runtime.util.{ZooKeeperUtil, MathUtils, EnvironmentInfor
import scala.concurrent._
import scala.concurrent.duration._
+import scala.concurrent.forkjoin.ForkJoinPool
import scala.util.{Failure, Success}
import scala.collection.JavaConverters._
@@ -1363,14 +1364,16 @@ object TaskManager {
@throws(classOf[IllegalConfigurationException])
@throws(classOf[IOException])
@throws(classOf[Exception])
- def startTaskManagerComponentsAndActor(configuration: Configuration,
- actorSystem: ActorSystem,
- taskManagerHostname: String,
- taskManagerActorName: Option[String],
- jobManagerPath: Option[String],
- localTaskManagerCommunication: Boolean,
- streamingMode: StreamingMode,
- taskManagerClass: Class[_ <: TaskManager]): ActorRef = {
+ def startTaskManagerComponentsAndActor(
+ configuration: Configuration,
+ actorSystem: ActorSystem,
+ taskManagerHostname: String,
+ taskManagerActorName: Option[String],
+ jobManagerPath: Option[String],
+ localTaskManagerCommunication: Boolean,
+ streamingMode: StreamingMode,
+ taskManagerClass: Class[_ <: TaskManager])
+ : ActorRef = {
// get and check the JobManager config
val jobManagerAkkaUrl: String = jobManagerPath.getOrElse {
@@ -1380,17 +1383,20 @@ object TaskManager {
}
val (taskManagerConfig : TaskManagerConfiguration,
- netConfig: NetworkEnvironmentConfiguration,
- connectionInfo: InstanceConnectionInfo)
-
- = parseTaskManagerConfiguration(configuration, taskManagerHostname,
- localTaskManagerCommunication)
+ netConfig: NetworkEnvironmentConfiguration,
+ connectionInfo: InstanceConnectionInfo
+ ) = parseTaskManagerConfiguration(
+ configuration,
+ taskManagerHostname,
+ localTaskManagerCommunication)
// pre-start checks
checkTempDirs(taskManagerConfig.tmpDirPaths)
+ val executionContext = ExecutionContext.fromExecutor(new ForkJoinPool())
+
// we start the network first, to make sure it can allocate its buffers first
- val network = new NetworkEnvironment(taskManagerConfig.timeout, netConfig)
+ val network = new NetworkEnvironment(executionContext, taskManagerConfig.timeout, netConfig)
// computing the amount of memory to use depends on how much memory is available
// it strictly needs to happen AFTER the network stack has been initialized
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/AllVerticesIteratorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/AllVerticesIteratorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/AllVerticesIteratorTest.java
index 693e014..1e66d81 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/AllVerticesIteratorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/AllVerticesIteratorTest.java
@@ -22,6 +22,7 @@ import java.util.Arrays;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
@@ -43,6 +44,7 @@ public class AllVerticesIteratorTest {
v4.setParallelism(2);
ExecutionGraph eg = Mockito.mock(ExecutionGraph.class);
+ Mockito.when(eg.getExecutionContext()).thenReturn(TestingUtils.directExecutionContext());
ExecutionJobVertex ejv1 = new ExecutionJobVertex(eg, v1, 1,
AkkaUtils.getDefaultTimeout());
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphConstructionTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphConstructionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphConstructionTest.java
index a4bd03c..a47ea77 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphConstructionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphConstructionTest.java
@@ -28,6 +28,7 @@ import static org.mockito.Mockito.when;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.junit.Test;
import org.mockito.Matchers;
@@ -99,7 +100,12 @@ public class ExecutionGraphConstructionTest {
List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2, v3, v4, v5));
- ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ jobName,
+ cfg,
+ AkkaUtils.getDefaultTimeout());
try {
eg.attachJobGraph(ordered);
}
@@ -137,7 +143,12 @@ public class ExecutionGraphConstructionTest {
List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2, v3));
- ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ jobName,
+ cfg,
+ AkkaUtils.getDefaultTimeout());
try {
eg.attachJobGraph(ordered);
}
@@ -198,7 +209,12 @@ public class ExecutionGraphConstructionTest {
List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2, v3));
- ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ jobName,
+ cfg,
+ AkkaUtils.getDefaultTimeout());
try {
eg.attachJobGraph(ordered);
}
@@ -446,7 +462,12 @@ public class ExecutionGraphConstructionTest {
List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1));
- ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ jobName,
+ cfg,
+ AkkaUtils.getDefaultTimeout());
try {
eg.attachJobGraph(ordered);
}
@@ -496,7 +517,12 @@ public class ExecutionGraphConstructionTest {
List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2, v3, v5, v4));
- ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ jobName,
+ cfg,
+ AkkaUtils.getDefaultTimeout());
try {
eg.attachJobGraph(ordered);
fail("Attached wrong jobgraph");
@@ -551,7 +577,11 @@ public class ExecutionGraphConstructionTest {
List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2, v3, v4, v5));
- ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg,
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ jobName,
+ cfg,
AkkaUtils.getDefaultTimeout());
try {
eg.attachJobGraph(ordered);
@@ -591,7 +621,11 @@ public class ExecutionGraphConstructionTest {
List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2, v3));
- ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg,
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ jobName,
+ cfg,
AkkaUtils.getDefaultTimeout());
try {
@@ -657,7 +691,11 @@ public class ExecutionGraphConstructionTest {
JobGraph jg = new JobGraph(jobId, jobName, v1, v2, v3, v4, v5, v6, v7, v8);
- ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg,
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ jobName,
+ cfg,
AkkaUtils.getDefaultTimeout());
eg.attachJobGraph(jg.getVerticesSortedTopologicallyFromSources());
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/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 03a41b4..cff7146 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
@@ -30,12 +30,6 @@ import java.util.Comparator;
import java.util.List;
import java.util.Map;
-import akka.actor.Actor;
-import akka.actor.ActorRef;
-import akka.actor.ActorSystem;
-import akka.actor.Props;
-import akka.testkit.JavaTestKit;
-import akka.testkit.TestActorRef;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.deployment.InputGateDeploymentDescriptor;
@@ -51,29 +45,13 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
import org.apache.flink.runtime.operators.RegularPactTask;
import org.apache.flink.runtime.testingUtils.TestingUtils;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
import org.junit.Test;
public class ExecutionGraphDeploymentTest {
- private static ActorSystem system;
-
- @BeforeClass
- public static void setup() {
- system = ActorSystem.create("TestingActorSystem", TestingUtils.testConfig());
- }
-
- @AfterClass
- public static void teardown() {
- JavaTestKit.shutdownActorSystem(system);
- system = null;
- }
-
@Test
public void testBuildDeploymentDescriptor() {
try {
- TestingUtils.setCallingThreadDispatcher(system);
final JobID jobId = new JobID();
final JobVertexID jid1 = new JobVertexID();
@@ -100,7 +78,11 @@ public class ExecutionGraphDeploymentTest {
v3.connectNewDataSetAsInput(v2, DistributionPattern.ALL_TO_ALL);
v4.connectNewDataSetAsInput(v2, DistributionPattern.ALL_TO_ALL);
- ExecutionGraph eg = new ExecutionGraph(jobId, "some job", new Configuration(),
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ "some job",
+ new Configuration(),
AkkaUtils.getDefaultTimeout());
List<JobVertex> ordered = Arrays.asList(v1, v2, v3, v4);
@@ -110,15 +92,9 @@ public class ExecutionGraphDeploymentTest {
ExecutionJobVertex ejv = eg.getAllVertices().get(jid2);
ExecutionVertex vertex = ejv.getTaskVertices()[3];
- // create synchronous task manager
- final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
- Props.create(ExecutionGraphTestUtils
- .SimpleAcknowledgingTaskManager.class));
-
- ExecutionGraphTestUtils.SimpleAcknowledgingTaskManager tm = (ExecutionGraphTestUtils
- .SimpleAcknowledgingTaskManager) simpleTaskManager.underlyingActor();
+ ExecutionGraphTestUtils.SimpleInstanceGateway instanceGateway = new ExecutionGraphTestUtils.SimpleInstanceGateway(TestingUtils.directExecutionContext());
- final Instance instance = getInstance(simpleTaskManager);
+ final Instance instance = getInstance(instanceGateway);
final SimpleSlot slot = instance.allocateSimpleSlot(jobId);
@@ -128,7 +104,7 @@ public class ExecutionGraphDeploymentTest {
assertEquals(ExecutionState.DEPLOYING, vertex.getExecutionState());
- TaskDeploymentDescriptor descr = tm.lastTDD;
+ TaskDeploymentDescriptor descr = instanceGateway.lastTDD;
assertNotNull(descr);
assertEquals(jobId, descr.getJobID());
@@ -152,9 +128,6 @@ public class ExecutionGraphDeploymentTest {
e.printStackTrace();
fail(e.getMessage());
}
- finally {
- TestingUtils.setGlobalExecutionContext();
- }
}
@Test
@@ -307,19 +280,23 @@ public class ExecutionGraphDeploymentTest {
v2.setInvokableClass(RegularPactTask.class);
// execution graph that executes actions synchronously
- ExecutionGraph eg = new ExecutionGraph(jobId, "some job", new Configuration(),
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.directExecutionContext(),
+ jobId,
+ "some job",
+ new Configuration(),
AkkaUtils.getDefaultTimeout());
eg.setQueuedSchedulingAllowed(false);
List<JobVertex> ordered = Arrays.asList(v1, v2);
eg.attachJobGraph(ordered);
- // create a mock taskmanager that accepts deployment calls
- ActorRef tm = system.actorOf(Props.create(ExecutionGraphTestUtils.SimpleAcknowledgingTaskManager.class));
-
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
for (int i = 0; i < dop1 + dop2; i++) {
- scheduler.newInstanceAvailable(ExecutionGraphTestUtils.getInstance(tm));
+ scheduler.newInstanceAvailable(
+ ExecutionGraphTestUtils.getInstance(
+ new ExecutionGraphTestUtils.SimpleInstanceGateway(
+ TestingUtils.directExecutionContext())));
}
assertEquals(dop1 + dop2, scheduler.getNumberOfAvailableSlots());
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/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 a77a09e..8a63060 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
@@ -26,17 +26,16 @@ import java.lang.reflect.Field;
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;
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.BaseTestingInstanceGateway;
import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.InstanceConnectionInfo;
+import org.apache.flink.runtime.instance.InstanceGateway;
import org.apache.flink.runtime.instance.InstanceID;
import org.apache.flink.runtime.instance.SimpleSlot;
import org.apache.flink.runtime.jobgraph.JobVertex;
@@ -49,9 +48,11 @@ import org.apache.flink.runtime.messages.TaskMessages.SubmitTask;
import org.apache.flink.runtime.messages.TaskMessages.FailIntermediateResultPartitions;
import org.apache.flink.runtime.messages.TaskMessages.CancelTask;
import org.apache.flink.runtime.messages.TaskMessages.TaskOperationResult;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.mockito.Matchers;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
+import scala.concurrent.ExecutionContext;
public class ExecutionGraphTestUtils {
@@ -100,103 +101,95 @@ public class ExecutionGraphTestUtils {
// utility mocking methods
// --------------------------------------------------------------------------------------------
- public static Instance getInstance(final ActorRef taskManager) throws
- Exception {
- return getInstance(taskManager, 1);
+ public static Instance getInstance(final InstanceGateway gateway) throws Exception {
+ return getInstance(gateway, 1);
}
- public static Instance getInstance(final ActorRef taskManager, final int numberOfSlots) throws Exception {
+ public static Instance getInstance(final InstanceGateway gateway, final int numberOfSlots) throws Exception {
HardwareDescription hardwareDescription = new HardwareDescription(4, 2L*1024*1024*1024, 1024*1024*1024, 512*1024*1024);
InetAddress address = InetAddress.getByName("127.0.0.1");
InstanceConnectionInfo connection = new InstanceConnectionInfo(address, 10001);
-
- return new Instance(taskManager, connection, new InstanceID(), hardwareDescription, numberOfSlots);
+
+ return new Instance(gateway, connection, new InstanceID(), hardwareDescription, numberOfSlots);
}
- public static class SimpleAcknowledgingTaskManager extends UntypedActor {
+ public static class SimpleInstanceGateway extends BaseTestingInstanceGateway {
public TaskDeploymentDescriptor lastTDD;
+
+ public SimpleInstanceGateway(ExecutionContext executionContext){
+ super(executionContext);
+ }
+
@Override
- public void onReceive(Object msg) throws Exception {
- if (msg instanceof SubmitTask) {
- SubmitTask submitTask = (SubmitTask) msg;
+ public Object handleMessage(Object message) {
+ Object result = null;
+ if(message instanceof SubmitTask) {
+ SubmitTask submitTask = (SubmitTask) message;
lastTDD = submitTask.tasks();
- getSender().tell(Messages.getAcknowledge(), getSelf());
- } else if (msg instanceof CancelTask) {
- CancelTask cancelTask = (CancelTask) msg;
- getSender().tell(new TaskOperationResult(cancelTask.attemptID(), true), getSelf());
- }
- else if (msg instanceof FailIntermediateResultPartitions) {
- getSender().tell(new Object(), getSelf());
+ result = Messages.getAcknowledge();
+ } else if(message instanceof CancelTask) {
+ CancelTask cancelTask = (CancelTask) message;
+
+ result = new TaskOperationResult(cancelTask.attemptID(), true);
+ } else if(message instanceof FailIntermediateResultPartitions) {
+ result = new Object();
}
+
+ return result;
}
}
- public static final String ERROR_MESSAGE = "test_failure_error_message";
+ public static class SimpleFailingInstanceGateway extends BaseTestingInstanceGateway {
+ public SimpleFailingInstanceGateway(ExecutionContext executionContext) {
+ super(executionContext);
+ }
- public static class SimpleFailingTaskManager extends UntypedActor {
@Override
- public void onReceive(Object msg) throws Exception {
- if (msg instanceof SubmitTask) {
- 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());
+ public Object handleMessage(Object message) throws Exception {
+ if(message instanceof SubmitTask) {
+ throw new Exception(ERROR_MESSAGE);
+ } else if (message instanceof CancelTask) {
+ CancelTask cancelTask = (CancelTask) message;
+
+ return new TaskOperationResult(cancelTask.attemptID(), true);
+ } else {
+ return null;
}
}
}
-
- public static ExecutionJobVertex getExecutionVertex(JobVertexID id) throws JobException {
+
+ public static final String ERROR_MESSAGE = "test_failure_error_message";
+
+ public static ExecutionJobVertex getExecutionVertex(JobVertexID id, ExecutionContext executionContext) throws JobException {
JobVertex ajv = new JobVertex("TestVertex", id);
ajv.setInvokableClass(mock(AbstractInvokable.class).getClass());
-
- ExecutionGraph graph = new ExecutionGraph(new JobID(), "test job", new Configuration(),
+
+ ExecutionGraph graph = new ExecutionGraph(
+ executionContext,
+ new JobID(),
+ "test job",
+ new Configuration(),
AkkaUtils.getDefaultTimeout());
-
+
ExecutionJobVertex ejv = spy(new ExecutionJobVertex(graph, ajv, 1,
AkkaUtils.getDefaultTimeout()));
-
+
Answer<Void> noop = new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) {
return null;
}
};
-
+
doAnswer(noop).when(ejv).vertexCancelled(Matchers.anyInt());
doAnswer(noop).when(ejv).vertexFailed(Matchers.anyInt(), Matchers.any(Throwable.class));
doAnswer(noop).when(ejv).vertexFinished(Matchers.anyInt());
-
+
return ejv;
}
- // --------------------------------------------------------------------------------------------
-
- public static final class ActionQueue {
-
- private final LinkedList<Runnable> runnables = new LinkedList<Runnable>();
-
- public void triggerNextAction() {
- Runnable r = runnables.remove();
- r.run();
- }
-
- public void triggerLatestAction(){
- Runnable r = runnables.removeLast();
- r.run();
- }
-
- public Runnable popNextAction() {
- Runnable r = runnables.remove();
- return r;
- }
-
- public void queueAction(Runnable r) {
- this.runnables.add(r);
- }
-
- public boolean isEmpty(){
- return runnables.isEmpty();
- }
+ public static ExecutionJobVertex getExecutionVertex(JobVertexID id) throws JobException {
+ return getExecutionVertex(id, TestingUtils.defaultExecutionContext());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
index 7787ab4..f47e92c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
@@ -24,10 +24,6 @@ import static org.mockito.Mockito.mock;
import java.util.Arrays;
-import akka.actor.ActorRef;
-import akka.actor.ActorSystem;
-import akka.actor.Props;
-import akka.testkit.JavaTestKit;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.instance.SimpleSlot;
@@ -37,24 +33,10 @@ 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.testingUtils.TestingUtils;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
import org.junit.Test;
public class ExecutionStateProgressTest {
- private static ActorSystem system;
-
- @BeforeClass
- public static void setup(){
- system = ActorSystem.create("TestingActorSystem", TestingUtils.testConfig());
- }
-
- @AfterClass
- public static void teardown(){
- JavaTestKit.shutdownActorSystem(system);
- }
-
@Test
public void testAccumulatedStateFinished() {
try {
@@ -65,8 +47,13 @@ public class ExecutionStateProgressTest {
ajv.setParallelism(3);
ajv.setInvokableClass(mock(AbstractInvokable.class).getClass());
- ExecutionGraph graph = new ExecutionGraph(jid, "test job", new Configuration(),
+ ExecutionGraph graph = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jid,
+ "test job",
+ new Configuration(),
AkkaUtils.getDefaultTimeout());
+
graph.attachJobGraph(Arrays.asList(ajv));
setGraphStatus(graph, JobStatus.RUNNING);
@@ -74,9 +61,11 @@ public class ExecutionStateProgressTest {
ExecutionJobVertex ejv = graph.getJobVertex(vid);
// mock resources and mock taskmanager
- ActorRef taskManager = system.actorOf(Props.create(SimpleAcknowledgingTaskManager.class));
for (ExecutionVertex ee : ejv.getTaskVertices()) {
- SimpleSlot slot = getInstance(taskManager).allocateSimpleSlot(jid);
+ SimpleSlot slot = getInstance(
+ new SimpleInstanceGateway(
+ TestingUtils.defaultExecutionContext())
+ ).allocateSimpleSlot(jid);
ee.deployToSlot(slot);
}
[2/7] flink git commit: [FLINK-2235] fix calculation of free memory
for local execution
Posted by se...@apache.org.
[FLINK-2235] fix calculation of free memory for local execution
The Java runtime may return Long.MAX_VALUE for a call to the maxMemory()
method of the Runtime class. In these cases, we can get hold of the
physical memory size of the operating system by using a proprietary
Oracle JDK method. Otherwise, we fail with an explanatory exception.
The Oracle JVM defines the max memory to be 1/4 of the physical memory
if the user has more than 192 Megabytes which is assumed here.
This closes #859
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/910e8c43
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/910e8c43
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/910e8c43
Branch: refs/heads/master
Commit: 910e8c438477f27c6d20392d00dd0cfbe8496c8b
Parents: 2ccb5fd
Author: Maximilian Michels <mx...@apache.org>
Authored: Mon Jun 22 17:47:11 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Jul 13 17:54:31 2015 +0200
----------------------------------------------------------------------
.../runtime/util/EnvironmentInformation.java | 19 ++++++++++++++++++-
1 file changed, 18 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/910e8c43/flink-runtime/src/main/java/org/apache/flink/runtime/util/EnvironmentInformation.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/EnvironmentInformation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/EnvironmentInformation.java
index 630e227..736b2b5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/EnvironmentInformation.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/EnvironmentInformation.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.util;
import java.io.InputStream;
import java.lang.management.ManagementFactory;
+import java.lang.management.OperatingSystemMXBean;
import java.lang.management.RuntimeMXBean;
import java.lang.reflect.Method;
import java.util.List;
@@ -138,7 +139,23 @@ public class EnvironmentInformation {
*/
public static long getSizeOfFreeHeapMemory() {
Runtime r = Runtime.getRuntime();
- return r.maxMemory() - r.totalMemory() + r.freeMemory();
+ long maxMemory = r.maxMemory();
+
+ if (maxMemory == Long.MAX_VALUE) {
+ // amount of free memory unknown
+ try {
+ // workaround for Oracle JDK
+ OperatingSystemMXBean operatingSystemMXBean = ManagementFactory.getOperatingSystemMXBean();
+ Class<?> clazz = Class.forName("com.sun.management.OperatingSystemMXBean");
+ Method method = clazz.getMethod("getTotalPhysicalMemorySize");
+ maxMemory = (Long) method.invoke(operatingSystemMXBean) / 4;
+ } catch (Throwable e) {
+ throw new RuntimeException("Could not determine the amount of free memory.\n" +
+ "Please set the maximum memory for the JVM, e.g. -Xmx512M for 512 megabytes.");
+ }
+ }
+
+ return maxMemory - r.totalMemory() + r.freeMemory();
}
/**
[4/7] flink git commit: [hotfix] Fix error messages in
EnvironmentInformation when accessing user information
Posted by se...@apache.org.
[hotfix] Fix error messages in EnvironmentInformation when accessing user information
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/84b39dcb
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/84b39dcb
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/84b39dcb
Branch: refs/heads/master
Commit: 84b39dcb50167368d45882a88842760237fa3560
Parents: b0b0cb5
Author: Stephan Ewen <se...@apache.org>
Authored: Mon Jul 13 16:27:01 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Jul 13 17:54:31 2015 +0200
----------------------------------------------------------------------
.../apache/flink/runtime/util/EnvironmentInformation.java | 10 +++++++---
1 file changed, 7 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/84b39dcb/flink-runtime/src/main/java/org/apache/flink/runtime/util/EnvironmentInformation.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/EnvironmentInformation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/EnvironmentInformation.java
index 736b2b5..dde422b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/EnvironmentInformation.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/EnvironmentInformation.java
@@ -89,10 +89,14 @@ public class EnvironmentInformation {
try {
return UserGroupInformation.getCurrentUser().getShortUserName();
}
+ catch (LinkageError e) {
+ // hadoop classes are not in the classpath
+ LOG.debug("Cannot determine user/group information using Hadoop utils. " +
+ "Hadoop classes not loaded or compatible", e);
+ }
catch (Throwable t) {
- if (LOG.isDebugEnabled() && !(t instanceof ClassNotFoundException)) {
- LOG.debug("Cannot determine user/group information using Hadoop utils.", t);
- }
+ // some other error occurred that we should log and make known
+ LOG.warn("Error while accessing user/group information via Hadoop utils.", t);
}
String user = System.getProperty("user.name");
[6/7] flink git commit: [FLINK-2329] [runtime] Introduces
InstanceGateway as an abstraction to communicate with the TaskManager.
Posted by se...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/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 eb0aab4..9db330b 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
@@ -24,18 +24,11 @@ import static org.mockito.Mockito.mock;
import java.io.IOException;
-import akka.actor.Actor;
-import akka.testkit.TestActorRef;
-import akka.actor.ActorRef;
-import akka.actor.ActorSystem;
-import akka.actor.Props;
-import akka.actor.Status;
-import akka.actor.UntypedActor;
-import akka.japi.Creator;
-import akka.testkit.JavaTestKit;
-
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.instance.BaseTestingInstanceGateway;
+import org.apache.flink.runtime.instance.DummyInstanceGateway;
+import org.apache.flink.runtime.instance.InstanceGateway;
import org.apache.flink.runtime.instance.SimpleSlot;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.api.common.JobID;
@@ -46,26 +39,12 @@ import org.apache.flink.runtime.messages.TaskMessages;
import org.apache.flink.runtime.messages.TaskMessages.TaskOperationResult;
import org.apache.flink.runtime.testingUtils.TestingUtils;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
import org.junit.Test;
+import scala.concurrent.ExecutionContext;
@SuppressWarnings("serial")
public class ExecutionVertexCancelTest {
- private static ActorSystem system;
-
- @BeforeClass
- public static void setup(){
- system = ActorSystem.create("TestingActorSystem", TestingUtils.testConfig());
- }
-
- @AfterClass
- public static void teardown(){
- JavaTestKit.shutdownActorSystem(system);
- system = null;
- }
-
// --------------------------------------------------------------------------------------------
// Canceling in different states
// --------------------------------------------------------------------------------------------
@@ -127,388 +106,350 @@ public class ExecutionVertexCancelTest {
@Test
public void testCancelConcurrentlyToDeploying_CallsNotOvertaking() {
- new JavaTestKit(system){{
- try {
- final JobVertexID jid = new JobVertexID();
- final ActionQueue actions = new ActionQueue();
+ try {
+ final JobVertexID jid = new JobVertexID();
- TestingUtils.setExecutionContext(new TestingUtils.QueuedActionExecutionContext(
- actions));
+ final TestingUtils.QueuedActionExecutionContext executionContext = TestingUtils.queuedActionExecutionContext();
+ final TestingUtils.ActionQueue actions = executionContext.actionQueue();
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
- final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
- AkkaUtils.getDefaultTimeout());
- final ExecutionAttemptID execId = vertex.getCurrentExecutionAttempt().getAttemptId();
+ final ExecutionJobVertex ejv = getExecutionVertex(
+ jid,
+ executionContext
+ );
- setVertexState(vertex, ExecutionState.SCHEDULED);
- assertEquals(ExecutionState.SCHEDULED, vertex.getExecutionState());
+ final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
+ AkkaUtils.getDefaultTimeout());
+ final ExecutionAttemptID execId = vertex.getCurrentExecutionAttempt().getAttemptId();
- ActorRef taskManager = TestActorRef.create(system, Props.create(new
- CancelSequenceTaskManagerCreator(new TaskOperationResult(execId, true),
- new TaskOperationResult(execId, false))));
+ setVertexState(vertex, ExecutionState.SCHEDULED);
+ assertEquals(ExecutionState.SCHEDULED, vertex.getExecutionState());
- Instance instance = getInstance(taskManager);
- SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ InstanceGateway instanceGateway = new CancelSequenceInstanceGateway(
+ executionContext,
+ new TaskOperationResult(execId, true),
+ new TaskOperationResult(execId, false));
- vertex.deployToSlot(slot);
+ Instance instance = getInstance(instanceGateway);
+ SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ vertex.deployToSlot(slot);
- assertEquals(ExecutionState.DEPLOYING, vertex.getExecutionState());
+ assertEquals(ExecutionState.DEPLOYING, vertex.getExecutionState());
- vertex.cancel();
+ vertex.cancel();
- assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
+ assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
- // first action happens (deploy)
- actions.triggerNextAction();
- assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
+ // first action happens (deploy)
+ actions.triggerNextAction();
+ assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
- // the deploy call found itself in canceling after it returned and needs to send a cancel call
- // the call did not yet execute, so it is still in canceling
- assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
+ // the deploy call found itself in canceling after it returned and needs to send a cancel call
+ // the call did not yet execute, so it is still in canceling
+ assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
- // second action happens (cancel call from cancel function)
- actions.triggerNextAction();
+ // second action happens (cancel call from cancel function)
+ actions.triggerNextAction();
- // TaskManager reports back (canceling done)
- vertex.getCurrentExecutionAttempt().cancelingComplete();
+ // TaskManager reports back (canceling done)
+ vertex.getCurrentExecutionAttempt().cancelingComplete();
- // should properly set state to cancelled
- assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
+ // should properly set state to cancelled
+ assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
- // trigger the correction canceling call
- actions.triggerNextAction();
- assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
+ // trigger the correction canceling call
+ actions.triggerNextAction();
+ assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
- assertTrue(slot.isReleased());
+ assertTrue(slot.isReleased());
- assertNull(vertex.getFailureCause());
+ assertNull(vertex.getFailureCause());
- assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
- assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
- assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELED) > 0);
- }
- catch(Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
- finally {
- TestingUtils.setGlobalExecutionContext();
- }
- }};
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELED) > 0);
+ }
+ catch(Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
}
@Test
public void testCancelConcurrentlyToDeploying_CallsOvertaking() {
- new JavaTestKit(system){
- {
- try {
- final JobVertexID jid = new JobVertexID();
- final ActionQueue actions = new ActionQueue();
+ try {
+ final JobVertexID jid = new JobVertexID();
- TestingUtils.setExecutionContext(new TestingUtils
- .QueuedActionExecutionContext(actions));
+ final TestingUtils.QueuedActionExecutionContext executionContext = TestingUtils.queuedActionExecutionContext();
+ final TestingUtils.ActionQueue actions = executionContext.actionQueue();
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
+ final ExecutionJobVertex ejv = getExecutionVertex(jid, executionContext);
- final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
- AkkaUtils.getDefaultTimeout());
- final ExecutionAttemptID execId = vertex.getCurrentExecutionAttempt().getAttemptId();
+ final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
+ AkkaUtils.getDefaultTimeout());
+ final ExecutionAttemptID execId = vertex.getCurrentExecutionAttempt().getAttemptId();
- setVertexState(vertex, ExecutionState.SCHEDULED);
- assertEquals(ExecutionState.SCHEDULED, vertex.getExecutionState());
+ setVertexState(vertex, ExecutionState.SCHEDULED);
+ assertEquals(ExecutionState.SCHEDULED, vertex.getExecutionState());
- // task manager cancel sequence mock actor
- // first return NOT SUCCESS (task not found, cancel call overtook deploy call), then success (cancel call after deploy call)
- TestActorRef<? extends Actor> taskManager = TestActorRef.create(system, Props.create(new
- CancelSequenceTaskManagerCreator(new
- TaskOperationResult(execId, false), new TaskOperationResult(execId, true))));
+ // task manager cancel sequence mock actor
+ // first return NOT SUCCESS (task not found, cancel call overtook deploy call), then success (cancel call after deploy call)
+ InstanceGateway instanceGateway = new CancelSequenceInstanceGateway(
+ executionContext,
+ new TaskOperationResult(execId, false),
+ new TaskOperationResult(execId, true)
+ );
- Instance instance = getInstance(taskManager);
- SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ Instance instance = getInstance(instanceGateway);
+ SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
- vertex.deployToSlot(slot);
+ vertex.deployToSlot(slot);
- assertEquals(ExecutionState.DEPLOYING, vertex.getExecutionState());
+ assertEquals(ExecutionState.DEPLOYING, vertex.getExecutionState());
- vertex.cancel();
+ vertex.cancel();
- assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
+ assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
- // first action happens (deploy)
- Runnable deployAction = actions.popNextAction();
- Runnable cancelAction = actions.popNextAction();
+ // first action happens (deploy)
+ Runnable deployAction = actions.popNextAction();
+ Runnable cancelAction = actions.popNextAction();
- // cancel call first
- cancelAction.run();
- // process onComplete callback
- actions.triggerNextAction();
+ // cancel call first
+ cancelAction.run();
+ // process onComplete callback
+ actions.triggerNextAction();
- // did not find the task, not properly cancelled, stay in canceling
- assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
+ // did not find the task, not properly cancelled, stay in canceling
+ assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
- // deploy action next
- deployAction.run();
+ // deploy action next
+ deployAction.run();
- // the deploy call found itself in canceling after it returned and needs to send a cancel call
- // the call did not yet execute, so it is still in canceling
- assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
+ // the deploy call found itself in canceling after it returned and needs to send a cancel call
+ // the call did not yet execute, so it is still in canceling
+ assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
- vertex.getCurrentExecutionAttempt().cancelingComplete();
+ vertex.getCurrentExecutionAttempt().cancelingComplete();
- assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
+ assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
- assertTrue(slot.isReleased());
+ assertTrue(slot.isReleased());
- assertNull(vertex.getFailureCause());
+ assertNull(vertex.getFailureCause());
- assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
- assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
- assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELED) > 0);
- } catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }finally{
- TestingUtils.setGlobalExecutionContext();
- }
- }
- };
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELED) > 0);
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
}
@Test
public void testCancelFromRunning() {
- new JavaTestKit(system) {
- {
- try {
- TestingUtils.setCallingThreadDispatcher(system);
- final JobVertexID jid = new JobVertexID();
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
+ try {
+ final JobVertexID jid = new JobVertexID();
+ final ExecutionJobVertex ejv = getExecutionVertex(jid, TestingUtils.directExecutionContext());
- final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
- AkkaUtils.getDefaultTimeout());
- final ExecutionAttemptID execId = vertex.getCurrentExecutionAttempt().getAttemptId();
+ final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
+ AkkaUtils.getDefaultTimeout());
+ final ExecutionAttemptID execId = vertex.getCurrentExecutionAttempt().getAttemptId();
- final TestActorRef<? extends Actor> taskManager = TestActorRef.create(system,
- Props.create(new CancelSequenceTaskManagerCreator(new
- TaskOperationResult(execId, true))));
+ InstanceGateway instanceGateway = new CancelSequenceInstanceGateway(
+ TestingUtils.directExecutionContext(),
+ new TaskOperationResult(execId, true));
- Instance instance = getInstance(taskManager);
- SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ Instance instance = getInstance(instanceGateway);
+ SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
- setVertexState(vertex, ExecutionState.RUNNING);
- setVertexResource(vertex, slot);
+ setVertexState(vertex, ExecutionState.RUNNING);
+ setVertexResource(vertex, slot);
- assertEquals(ExecutionState.RUNNING, vertex.getExecutionState());
+ assertEquals(ExecutionState.RUNNING, vertex.getExecutionState());
- vertex.cancel();
- vertex.getCurrentExecutionAttempt().cancelingComplete(); // responce by task manager once actially canceled
+ vertex.cancel();
+ vertex.getCurrentExecutionAttempt().cancelingComplete(); // responce by task manager once actially canceled
- assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
+ assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
- assertTrue(slot.isReleased());
+ assertTrue(slot.isReleased());
- assertNull(vertex.getFailureCause());
+ assertNull(vertex.getFailureCause());
- assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
- assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
- assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELED) > 0);
- } catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }finally{
- TestingUtils.setGlobalExecutionContext();
- }
- }
- };
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELED) > 0);
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
}
@Test
public void testRepeatedCancelFromRunning() {
- new JavaTestKit(system) {
- {
- try {
- TestingUtils.setCallingThreadDispatcher(system);
+ try {
- final JobVertexID jid = new JobVertexID();
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
+ final JobVertexID jid = new JobVertexID();
+ final ExecutionJobVertex ejv = getExecutionVertex(jid, TestingUtils.directExecutionContext());
- final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
- AkkaUtils.getDefaultTimeout());
- final ExecutionAttemptID execId = vertex.getCurrentExecutionAttempt().getAttemptId();
+ final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
+ AkkaUtils.getDefaultTimeout());
+ final ExecutionAttemptID execId = vertex.getCurrentExecutionAttempt().getAttemptId();
- final ActorRef taskManager = TestActorRef.create(system, Props.create(new
- CancelSequenceTaskManagerCreator(new
- TaskOperationResult(execId, true))));
+ final InstanceGateway instanceGateway = new CancelSequenceInstanceGateway(
+ TestingUtils.directExecutionContext(),
+ new TaskOperationResult(execId, true)
+ );
- Instance instance = getInstance(taskManager);
- SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ Instance instance = getInstance(instanceGateway);
+ SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
- setVertexState(vertex, ExecutionState.RUNNING);
- setVertexResource(vertex, slot);
+ setVertexState(vertex, ExecutionState.RUNNING);
+ setVertexResource(vertex, slot);
- assertEquals(ExecutionState.RUNNING, vertex.getExecutionState());
+ assertEquals(ExecutionState.RUNNING, vertex.getExecutionState());
- vertex.cancel();
+ vertex.cancel();
- assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
+ assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
- vertex.cancel();
+ vertex.cancel();
- assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
+ assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
- // callback by TaskManager after canceling completes
- vertex.getCurrentExecutionAttempt().cancelingComplete();
+ // callback by TaskManager after canceling completes
+ vertex.getCurrentExecutionAttempt().cancelingComplete();
- assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
+ assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
- assertTrue(slot.isReleased());
+ assertTrue(slot.isReleased());
- assertNull(vertex.getFailureCause());
+ assertNull(vertex.getFailureCause());
- assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
- assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
- assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELED) > 0);
- } catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }finally{
- TestingUtils.setGlobalExecutionContext();
- }
- }
- };
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELED) > 0);
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
}
@Test
public void testCancelFromRunningDidNotFindTask() {
// this may happen when the task finished or failed while the call was in progress
- new JavaTestKit(system) {
- {
- try {
- TestingUtils.setCallingThreadDispatcher(system);
- final JobVertexID jid = new JobVertexID();
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
+ try {
+ final JobVertexID jid = new JobVertexID();
+ final ExecutionJobVertex ejv = getExecutionVertex(jid, TestingUtils.directExecutionContext());
- final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
- AkkaUtils.getDefaultTimeout());
- final ExecutionAttemptID execId = vertex.getCurrentExecutionAttempt().getAttemptId();
+ final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
+ AkkaUtils.getDefaultTimeout());
+ final ExecutionAttemptID execId = vertex.getCurrentExecutionAttempt().getAttemptId();
- final ActorRef taskManager = TestActorRef.create(system,Props.create(new
- CancelSequenceTaskManagerCreator(new
- TaskOperationResult(execId, false))));
- Instance instance = getInstance(taskManager);
- SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ final InstanceGateway instanceGateway = new CancelSequenceInstanceGateway(
+ TestingUtils.directExecutionContext(),
+ new TaskOperationResult(execId, false)
+ );
- setVertexState(vertex, ExecutionState.RUNNING);
- setVertexResource(vertex, slot);
+ Instance instance = getInstance(instanceGateway);
+ SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
- assertEquals(ExecutionState.RUNNING, vertex.getExecutionState());
+ setVertexState(vertex, ExecutionState.RUNNING);
+ setVertexResource(vertex, slot);
- vertex.cancel();
+ assertEquals(ExecutionState.RUNNING, vertex.getExecutionState());
- assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
+ vertex.cancel();
- assertNull(vertex.getFailureCause());
+ assertEquals(ExecutionState.CANCELING, vertex.getExecutionState());
- assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
- assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
- } catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }finally{
- TestingUtils.setGlobalExecutionContext();
- }
- }
- };
+ assertNull(vertex.getFailureCause());
+
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
}
@Test
public void testCancelCallFails() {
- new JavaTestKit(system) {
- {
- try {
- TestingUtils.setCallingThreadDispatcher(system);
- final JobVertexID jid = new JobVertexID();
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
+ try {
+ final JobVertexID jid = new JobVertexID();
+ final ExecutionJobVertex ejv = getExecutionVertex(jid, TestingUtils.directExecutionContext());
- final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
- AkkaUtils.getDefaultTimeout());
+ final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
+ AkkaUtils.getDefaultTimeout());
- final ActorRef taskManager = TestActorRef.create(system, Props.create(new
- CancelSequenceTaskManagerCreator()));
+ final InstanceGateway gateway = new CancelSequenceInstanceGateway(TestingUtils.directExecutionContext());
- Instance instance = getInstance(taskManager);
- SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ Instance instance = getInstance(gateway);
+ SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
- setVertexState(vertex, ExecutionState.RUNNING);
- setVertexResource(vertex, slot);
+ setVertexState(vertex, ExecutionState.RUNNING);
+ setVertexResource(vertex, slot);
- assertEquals(ExecutionState.RUNNING, vertex.getExecutionState());
+ assertEquals(ExecutionState.RUNNING, vertex.getExecutionState());
- vertex.cancel();
+ vertex.cancel();
- assertEquals(ExecutionState.FAILED, vertex.getExecutionState());
+ assertEquals(ExecutionState.FAILED, vertex.getExecutionState());
- assertTrue(slot.isReleased());
+ assertTrue(slot.isReleased());
- assertNotNull(vertex.getFailureCause());
+ assertNotNull(vertex.getFailureCause());
- assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
- assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
- assertTrue(vertex.getStateTimestamp(ExecutionState.FAILED) > 0);
- } catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }finally{
- TestingUtils.setGlobalExecutionContext();
- }
- }
- };
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
+ assertTrue(vertex.getStateTimestamp(ExecutionState.FAILED) > 0);
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
}
@Test
public void testSendCancelAndReceiveFail() {
- new JavaTestKit(system) {
- {
- try {
- final JobVertexID jid = new JobVertexID();
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
+ try {
+ final JobVertexID jid = new JobVertexID();
+ final ExecutionJobVertex ejv = getExecutionVertex(jid);
- final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
- AkkaUtils.getDefaultTimeout());
- final ExecutionAttemptID execID = vertex.getCurrentExecutionAttempt().getAttemptId();
+ final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
+ AkkaUtils.getDefaultTimeout());
+ final ExecutionAttemptID execID = vertex.getCurrentExecutionAttempt().getAttemptId();
- final ActorRef taskManager = system.actorOf(
- Props.create(new CancelSequenceTaskManagerCreator(
- new TaskOperationResult(execID, true)
- )));
+ final InstanceGateway gateway = new CancelSequenceInstanceGateway(
+ TestingUtils.defaultExecutionContext(),
+ new TaskOperationResult(execID, true));
- Instance instance = getInstance(taskManager);
- SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+ Instance instance = getInstance(gateway);
+ SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
- setVertexState(vertex, ExecutionState.RUNNING);
- setVertexResource(vertex, slot);
+ setVertexState(vertex, ExecutionState.RUNNING);
+ setVertexResource(vertex, slot);
- assertEquals(ExecutionState.RUNNING, vertex.getExecutionState());
+ assertEquals(ExecutionState.RUNNING, vertex.getExecutionState());
- vertex.cancel();
+ vertex.cancel();
- assertTrue(vertex.getExecutionState() == ExecutionState.CANCELING || vertex.getExecutionState() == ExecutionState.FAILED);
+ assertTrue(vertex.getExecutionState() == ExecutionState.CANCELING || vertex.getExecutionState() == ExecutionState.FAILED);
- vertex.getCurrentExecutionAttempt().markFailed(new Throwable("test"));
+ vertex.getCurrentExecutionAttempt().markFailed(new Throwable("test"));
- assertTrue(vertex.getExecutionState() == ExecutionState.CANCELED || vertex.getExecutionState() == ExecutionState.FAILED);
+ assertTrue(vertex.getExecutionState() == ExecutionState.CANCELED || vertex.getExecutionState() == ExecutionState.FAILED);
- assertTrue(slot.isReleased());
+ assertTrue(slot.isReleased());
- assertEquals(0, vertex.getExecutionGraph().getRegisteredExecutions().size());
- } catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
- }
- };
+ assertEquals(0, vertex.getExecutionGraph().getRegisteredExecutions().size());
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
}
// --------------------------------------------------------------------------------------------
@@ -541,7 +482,7 @@ public class ExecutionVertexCancelTest {
// deploying after canceling from CREATED needs to raise an exception, because
// the scheduler (or any caller) needs to know that the slot should be released
try {
- Instance instance = getInstance(ActorRef.noSender());
+ Instance instance = getInstance(DummyInstanceGateway.INSTANCE);
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
vertex.deployToSlot(slot);
@@ -584,7 +525,7 @@ public class ExecutionVertexCancelTest {
AkkaUtils.getDefaultTimeout());
setVertexState(vertex, ExecutionState.CANCELING);
- Instance instance = getInstance(ActorRef.noSender());
+ Instance instance = getInstance(DummyInstanceGateway.INSTANCE);
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
vertex.deployToSlot(slot);
@@ -600,7 +541,7 @@ public class ExecutionVertexCancelTest {
ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
- Instance instance = getInstance(ActorRef.noSender());
+ Instance instance = getInstance(DummyInstanceGateway.INSTANCE);
SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
setVertexResource(vertex, slot);
@@ -621,39 +562,33 @@ public class ExecutionVertexCancelTest {
}
}
- public static class CancelSequenceTaskManagerCreator implements Creator<CancelSequenceTaskManager> {
- private final TaskOperationResult[] results;
- public CancelSequenceTaskManagerCreator(TaskOperationResult ... results){
- this.results = results;
- }
-
- @Override
- public CancelSequenceTaskManager create() throws Exception {
- return new CancelSequenceTaskManager(results);
- }
- }
-
- public static class CancelSequenceTaskManager extends UntypedActor{
+ public static class CancelSequenceInstanceGateway extends BaseTestingInstanceGateway {
private final TaskOperationResult[] results;
- private int index;
+ private int index = -1;
- public CancelSequenceTaskManager(TaskOperationResult[] results){
- this.results = results;
- index = -1;
+ public CancelSequenceInstanceGateway(ExecutionContext executionContext, TaskOperationResult ... result) {
+ super(executionContext);
+ this.results = result;
}
@Override
- public void onReceive(Object message) throws Exception {
- if(message instanceof TaskMessages.SubmitTask){
- getSender().tell(Messages.getAcknowledge(), getSelf());
- }else if(message instanceof TaskMessages.CancelTask){
+ public Object handleMessage(Object message) throws Exception {
+ Object result;
+ if(message instanceof TaskMessages.SubmitTask) {
+ result = Messages.getAcknowledge();
+ } else if(message instanceof TaskMessages.CancelTask) {
index++;
+
if(index >= results.length){
- getSender().tell(new Status.Failure(new IOException("RPC call failed.")), getSelf());
- }else {
- getSender().tell(results[index], getSelf());
+ throw new IOException("RPC call failed.");
+ } else {
+ result = results[index];
}
+ } else {
+ result = null;
}
+
+ return result;
}
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/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 eadf328..431c3a9 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,13 +22,6 @@ 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;
-import akka.testkit.JavaTestKit;
-import akka.testkit.TestActorRef;
-
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.instance.Instance;
@@ -37,37 +30,20 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.messages.TaskMessages.TaskOperationResult;
import org.apache.flink.runtime.testingUtils.TestingUtils;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
import org.junit.Test;
public class ExecutionVertexDeploymentTest {
- private static ActorSystem system;
-
- @BeforeClass
- public static void setup(){
- system = ActorSystem.create("TestingActorSystem", TestingUtils.testConfig());
- }
-
- @AfterClass
- public static void teardown(){
- JavaTestKit.shutdownActorSystem(system);
- }
-
@Test
public void testDeployCall() {
try {
final JobVertexID jid = new JobVertexID();
- TestingUtils.setCallingThreadDispatcher(system);
- ActorRef tm = TestActorRef.create(system, Props.create(SimpleAcknowledgingTaskManager
- .class));
-
final ExecutionJobVertex ejv = getExecutionVertex(jid);
// mock taskmanager to simply accept the call
- Instance instance = getInstance(tm);
+ Instance instance = getInstance(
+ new SimpleInstanceGateway(TestingUtils.directExecutionContext()));
final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
@@ -95,24 +71,17 @@ public class ExecutionVertexDeploymentTest {
e.printStackTrace();
fail(e.getMessage());
}
- finally {
- TestingUtils.setGlobalExecutionContext();
- }
}
@Test
public void testDeployWithSynchronousAnswer() {
try {
- TestingUtils.setCallingThreadDispatcher(system);
-
final JobVertexID jid = new JobVertexID();
- final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
- Props.create(SimpleAcknowledgingTaskManager.class));
-
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
+ final ExecutionJobVertex ejv = getExecutionVertex(jid, TestingUtils.directExecutionContext());
- final Instance instance = getInstance(simpleTaskManager);
+ final Instance instance = getInstance(
+ new SimpleInstanceGateway(TestingUtils.directExecutionContext()));
final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
@@ -143,9 +112,6 @@ public class ExecutionVertexDeploymentTest {
e.printStackTrace();
fail(e.getMessage());
}
- finally {
- TestingUtils.setGlobalExecutionContext();
- }
}
@Test
@@ -157,10 +123,8 @@ public class ExecutionVertexDeploymentTest {
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
- final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
- Props.create(SimpleAcknowledgingTaskManager.class));
-
- final Instance instance = getInstance(simpleTaskManager);
+ final Instance instance = getInstance(
+ new SimpleInstanceGateway(TestingUtils.defaultExecutionContext()));
final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
@@ -200,18 +164,14 @@ public class ExecutionVertexDeploymentTest {
@Test
public void testDeployFailedSynchronous() {
try {
- TestingUtils.setCallingThreadDispatcher(system);
-
final JobVertexID jid = new JobVertexID();
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
+ final ExecutionJobVertex ejv = getExecutionVertex(jid, TestingUtils.directExecutionContext());
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
- final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
- Props.create(SimpleFailingTaskManager.class));
-
- final Instance instance = getInstance(simpleTaskManager);
+ final Instance instance = getInstance(
+ new SimpleFailingInstanceGateway(TestingUtils.directExecutionContext()));
final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
@@ -230,9 +190,6 @@ public class ExecutionVertexDeploymentTest {
e.printStackTrace();
fail(e.getMessage());
}
- finally {
- TestingUtils.setGlobalExecutionContext();
- }
}
@Test
@@ -243,10 +200,8 @@ public class ExecutionVertexDeploymentTest {
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
- final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
- Props.create(SimpleFailingTaskManager.class));
-
- final Instance instance = getInstance(simpleTaskManager);
+ final Instance instance = getInstance(
+ new SimpleFailingInstanceGateway(TestingUtils.directExecutionContext()));
final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
@@ -280,20 +235,16 @@ public class ExecutionVertexDeploymentTest {
public void testFailExternallyDuringDeploy() {
try {
final JobVertexID jid = new JobVertexID();
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
- final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
- AkkaUtils.getDefaultTimeout());
+ final TestingUtils.QueuedActionExecutionContext ec = TestingUtils.queuedActionExecutionContext();
+ final TestingUtils.ActionQueue queue = ec.actionQueue();
- final ActionQueue queue = new ActionQueue();
- final TestingUtils.QueuedActionExecutionContext ec = new TestingUtils
- .QueuedActionExecutionContext(queue);
+ final ExecutionJobVertex ejv = getExecutionVertex(jid, ec);
- TestingUtils.setExecutionContext(ec);
+ final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
+ AkkaUtils.getDefaultTimeout());
- final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system,
- Props.create(SimpleAcknowledgingTaskManager.class));
- final Instance instance = getInstance(simpleTaskManager);
+ final Instance instance = getInstance(new SimpleInstanceGateway(TestingUtils.directExecutionContext()));
final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
@@ -307,6 +258,7 @@ public class ExecutionVertexDeploymentTest {
assertEquals(testError, vertex.getFailureCause());
queue.triggerNextAction();
+ queue.triggerNextAction();
assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
assertTrue(vertex.getStateTimestamp(ExecutionState.DEPLOYING) > 0);
@@ -316,34 +268,29 @@ public class ExecutionVertexDeploymentTest {
e.printStackTrace();
fail(e.getMessage());
}
- finally {
- TestingUtils.setGlobalExecutionContext();
- }
}
@Test
public void testFailCallOvertakesDeploymentAnswer() {
try {
- ActionQueue queue = new ActionQueue();
- TestingUtils.QueuedActionExecutionContext context = new TestingUtils
- .QueuedActionExecutionContext(queue);
-
- TestingUtils.setExecutionContext(context);
+ final TestingUtils.QueuedActionExecutionContext context = TestingUtils.queuedActionExecutionContext();
+ final TestingUtils.ActionQueue queue = context.actionQueue();
final JobVertexID jid = new JobVertexID();
- final ExecutionJobVertex ejv = getExecutionVertex(jid);
+ final ExecutionJobVertex ejv = getExecutionVertex(jid, context);
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
final ExecutionAttemptID eid = vertex.getCurrentExecutionAttempt().getAttemptId();
- final TestActorRef<? extends Actor> simpleTaskManager = TestActorRef.create(system, Props.create(new
- ExecutionVertexCancelTest.CancelSequenceTaskManagerCreator(new
- TaskOperationResult(eid, false), new TaskOperationResult(eid, true))));
+ final Instance instance = getInstance(
+ new ExecutionVertexCancelTest.CancelSequenceInstanceGateway(
+ context,
+ new TaskOperationResult(eid, false),
+ new TaskOperationResult(eid, true)));
- final Instance instance = getInstance(simpleTaskManager);
final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
assertEquals(ExecutionState.CREATED, vertex.getExecutionState());
@@ -361,11 +308,14 @@ public class ExecutionVertexDeploymentTest {
Runnable cancel1 = queue.popNextAction();
cancel1.run();
- // execute onComplete callback
+ // execute onComplete callback of cancel
queue.triggerNextAction();
deploy.run();
+ // execute onComplete callback of deploy
+ queue.triggerNextAction();
+
assertEquals(ExecutionState.FAILED, vertex.getExecutionState());
assertEquals(testError, vertex.getFailureCause());
@@ -380,8 +330,5 @@ public class ExecutionVertexDeploymentTest {
e.printStackTrace();
fail(e.getMessage());
}
- finally {
- TestingUtils.setGlobalExecutionContext();
- }
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/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 1e9c30b..8ea7017 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
@@ -23,14 +23,9 @@ import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.ge
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;
-import akka.actor.ActorRef;
-import akka.actor.ActorSystem;
-import akka.actor.Props;
-import akka.testkit.JavaTestKit;
-import akka.testkit.TestActorRef;
-
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.instance.DummyInstanceGateway;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.SimpleSlot;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -38,28 +33,12 @@ import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
import org.apache.flink.runtime.jobmanager.scheduler.ScheduledUnit;
import org.apache.flink.runtime.jobmanager.scheduler.SlotAllocationFuture;
import org.apache.flink.runtime.testingUtils.TestingUtils;
-
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.Matchers;
public class ExecutionVertexSchedulingTest {
- private static ActorSystem system;
-
- @BeforeClass
- public static void setup(){
- system = ActorSystem.create("TestingActorSystem", TestingUtils.testConfig());
- }
-
- @AfterClass
- public static void teardown(){
- JavaTestKit.shutdownActorSystem(system);
- system = null;
- }
-
@Test
public void testSlotReleasedWhenScheduledImmediately() {
try {
@@ -68,7 +47,7 @@ public class ExecutionVertexSchedulingTest {
AkkaUtils.getDefaultTimeout());
// a slot than cannot be deployed to
- final Instance instance = getInstance(ActorRef.noSender());
+ final Instance instance = getInstance(DummyInstanceGateway.INSTANCE);
final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
slot.releaseSlot();
@@ -98,7 +77,7 @@ public class ExecutionVertexSchedulingTest {
AkkaUtils.getDefaultTimeout());
// a slot than cannot be deployed to
- final Instance instance = getInstance(ActorRef.noSender());
+ final Instance instance = getInstance(DummyInstanceGateway.INSTANCE);
final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
slot.releaseSlot();
@@ -134,11 +113,7 @@ public class ExecutionVertexSchedulingTest {
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0],
AkkaUtils.getDefaultTimeout());
- TestingUtils.setCallingThreadDispatcher(system);
- ActorRef tm = TestActorRef.create(system, Props.create(ExecutionGraphTestUtils
- .SimpleAcknowledgingTaskManager.class));
-
- final Instance instance = getInstance(tm);
+ final Instance instance = getInstance(new ExecutionGraphTestUtils.SimpleInstanceGateway(TestingUtils.defaultExecutionContext()));
final SimpleSlot slot = instance.allocateSimpleSlot(ejv.getJobId());
Scheduler scheduler = mock(Scheduler.class);
@@ -153,8 +128,6 @@ public class ExecutionVertexSchedulingTest {
catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
- }finally{
- TestingUtils.setGlobalExecutionContext();
}
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/LocalInputSplitsTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/LocalInputSplitsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/LocalInputSplitsTest.java
index 0d2ffeb..b4a7e63 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/LocalInputSplitsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/LocalInputSplitsTest.java
@@ -39,39 +39,14 @@ import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
import org.apache.flink.runtime.operators.testutils.DummyInvokable;
import org.apache.flink.runtime.testingUtils.TestingUtils;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
import org.junit.Test;
import scala.concurrent.duration.FiniteDuration;
-import akka.actor.Actor;
-import akka.actor.ActorSystem;
-import akka.actor.Props;
-import akka.testkit.JavaTestKit;
-import akka.testkit.TestActorRef;
public class LocalInputSplitsTest {
private static final FiniteDuration TIMEOUT = new FiniteDuration(100, TimeUnit.SECONDS);
- private static ActorSystem system;
-
- private static TestActorRef<? extends Actor> taskManager;
-
-
- @BeforeClass
- public static void setup() {
- system = ActorSystem.create("TestingActorSystem", TestingUtils.testConfig());
- taskManager = TestActorRef.create(system,
- Props.create(ExecutionGraphTestUtils.SimpleAcknowledgingTaskManager.class));
- }
-
- @AfterClass
- public static void teardown() {
- JavaTestKit.shutdownActorSystem(system);
- system = null;
- }
-
// --------------------------------------------------------------------------------------------
@Test
@@ -290,14 +265,18 @@ public class LocalInputSplitsTest {
JobGraph jobGraph = new JobGraph("test job", vertex);
- ExecutionGraph eg = new ExecutionGraph(jobGraph.getJobID(),
- jobGraph.getName(), jobGraph.getJobConfiguration(), TIMEOUT);
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobGraph.getJobID(),
+ jobGraph.getName(),
+ jobGraph.getJobConfiguration(),
+ TIMEOUT);
eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources());
eg.setQueuedSchedulingAllowed(false);
// create a scheduler with 6 instances where always two are on the same host
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
Instance i1 = getInstance(new byte[] {10,0,1,1}, 12345, "host1", 1);
Instance i2 = getInstance(new byte[] {10,0,1,1}, 12346, "host1", 1);
Instance i3 = getInstance(new byte[] {10,0,1,2}, 12345, "host2", 1);
@@ -349,8 +328,12 @@ public class LocalInputSplitsTest {
JobGraph jobGraph = new JobGraph("test job", vertex);
- ExecutionGraph eg = new ExecutionGraph(jobGraph.getJobID(),
- jobGraph.getName(), jobGraph.getJobConfiguration(), TIMEOUT);
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobGraph.getJobID(),
+ jobGraph.getName(),
+ jobGraph.getJobConfiguration(),
+ TIMEOUT);
eg.setQueuedSchedulingAllowed(false);
eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources());
@@ -370,7 +353,7 @@ public class LocalInputSplitsTest {
}
private static Scheduler getScheduler(int numInstances, int numSlotsPerInstance) throws Exception {
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
for (int i = 0; i < numInstances; i++) {
byte[] ipAddress = new byte[] { 10, 0, 1, (byte) (1 + i) };
@@ -393,7 +376,13 @@ public class LocalInputSplitsTest {
when(connection.getHostname()).thenReturn(hostname);
when(connection.getFQDNHostname()).thenReturn(hostname);
- return new Instance(taskManager, connection, new InstanceID(), hardwareDescription, slots);
+ return new Instance(
+ new ExecutionGraphTestUtils.SimpleInstanceGateway(
+ TestingUtils.defaultExecutionContext()),
+ connection,
+ new InstanceID(),
+ hardwareDescription,
+ slots);
}
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
index 4677bf8..3cedb63 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
@@ -23,6 +23,7 @@ import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.junit.Test;
import java.util.ArrayList;
@@ -56,7 +57,12 @@ public class PointwisePatternTest {
List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2));
- ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ jobName,
+ cfg,
+ AkkaUtils.getDefaultTimeout());
try {
eg.attachJobGraph(ordered);
}
@@ -91,7 +97,12 @@ public class PointwisePatternTest {
List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2));
- ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ jobName,
+ cfg,
+ AkkaUtils.getDefaultTimeout());
try {
eg.attachJobGraph(ordered);
}
@@ -127,7 +138,12 @@ public class PointwisePatternTest {
List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2));
- ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ jobName,
+ cfg,
+ AkkaUtils.getDefaultTimeout());
try {
eg.attachJobGraph(ordered);
}
@@ -164,7 +180,12 @@ public class PointwisePatternTest {
List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2));
- ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ jobName,
+ cfg,
+ AkkaUtils.getDefaultTimeout());
try {
eg.attachJobGraph(ordered);
}
@@ -199,7 +220,12 @@ public class PointwisePatternTest {
List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2));
- ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ jobName,
+ cfg,
+ AkkaUtils.getDefaultTimeout());
try {
eg.attachJobGraph(ordered);
}
@@ -254,7 +280,12 @@ public class PointwisePatternTest {
List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2));
- ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ jobName,
+ cfg,
+ AkkaUtils.getDefaultTimeout());
try {
eg.attachJobGraph(ordered);
}
@@ -300,7 +331,12 @@ public class PointwisePatternTest {
List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2));
- ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ jobName,
+ cfg,
+ AkkaUtils.getDefaultTimeout());
try {
eg.attachJobGraph(ordered);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TerminalStateDeadlockTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TerminalStateDeadlockTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TerminalStateDeadlockTest.java
index 376ff14..b779d79 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TerminalStateDeadlockTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TerminalStateDeadlockTest.java
@@ -18,11 +18,10 @@
package org.apache.flink.runtime.executiongraph;
-import akka.actor.ActorRef;
-
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.instance.DummyInstanceGateway;
import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.InstanceConnectionInfo;
@@ -34,6 +33,7 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
import org.apache.flink.runtime.operators.testutils.DummyInvokable;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.junit.Test;
import scala.concurrent.duration.FiniteDuration;
@@ -78,7 +78,7 @@ public class TerminalStateDeadlockTest {
InstanceConnectionInfo ci = new InstanceConnectionInfo(address, 12345);
HardwareDescription resources = new HardwareDescription(4, 4000000, 3000000, 2000000);
- Instance instance = new Instance(ActorRef.noSender(), ci, new InstanceID(), resources, 4);
+ Instance instance = new Instance(DummyInstanceGateway.INSTANCE, ci, new InstanceID(), resources, 4);
this.resource = instance.allocateSimpleSlot(new JobID());
}
@@ -116,7 +116,7 @@ public class TerminalStateDeadlockTest {
vertices = Arrays.asList(v1, v2);
}
- final Scheduler scheduler = new Scheduler();
+ final Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
final Executor executor = Executors.newFixedThreadPool(4);
@@ -181,7 +181,7 @@ public class TerminalStateDeadlockTest {
private volatile boolean done;
TestExecGraph(JobID jobId) {
- super(jobId, "test graph", EMPTY_CONFIG, TIMEOUT);
+ super(TestingUtils.defaultExecutionContext(), jobId, "test graph", EMPTY_CONFIG, TIMEOUT);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexLocationConstraintTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexLocationConstraintTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexLocationConstraintTest.java
index 756b9a4..3305254 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexLocationConstraintTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexLocationConstraintTest.java
@@ -26,6 +26,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.TimeUnit;
+import org.apache.flink.runtime.instance.DummyInstanceGateway;
import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.InstanceConnectionInfo;
@@ -39,42 +40,14 @@ import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
import org.apache.flink.runtime.operators.testutils.DummyInvokable;
import org.apache.flink.runtime.testingUtils.TestingUtils;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
import org.junit.Test;
import scala.concurrent.duration.FiniteDuration;
-import akka.actor.Actor;
-import akka.actor.ActorRef;
-import akka.actor.ActorSystem;
-import akka.actor.Props;
-import akka.testkit.JavaTestKit;
-import akka.testkit.TestActorRef;
public class VertexLocationConstraintTest {
private static final FiniteDuration timeout = new FiniteDuration(100, TimeUnit.SECONDS);
- private static ActorSystem system;
-
- private static TestActorRef<? extends Actor> taskManager;
-
-
- @BeforeClass
- public static void setup() {
- system = ActorSystem.create("TestingActorSystem", TestingUtils.testConfig());
-
- taskManager = TestActorRef.create(system,
- Props.create(ExecutionGraphTestUtils.SimpleAcknowledgingTaskManager.class));
- }
-
- @AfterClass
- public static void teardown() {
- JavaTestKit.shutdownActorSystem(system);
- system = null;
- }
-
-
@Test
public void testScheduleWithConstraint1() {
try {
@@ -91,7 +64,7 @@ public class VertexLocationConstraintTest {
Instance instance2 = getInstance(address2, 6789, hostname2);
Instance instance3 = getInstance(address3, 6789, hostname3);
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
scheduler.newInstanceAvailable(instance1);
scheduler.newInstanceAvailable(instance2);
scheduler.newInstanceAvailable(instance3);
@@ -102,7 +75,12 @@ public class VertexLocationConstraintTest {
jobVertex.setParallelism(2);
JobGraph jg = new JobGraph("test job", jobVertex);
- ExecutionGraph eg = new ExecutionGraph(jg.getJobID(), jg.getName(), jg.getJobConfiguration(), timeout);
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jg.getJobID(),
+ jg.getName(),
+ jg.getJobConfiguration(),
+ timeout);
eg.attachJobGraph(Collections.singletonList(jobVertex));
ExecutionJobVertex ejv = eg.getAllVertices().get(jobVertex.getID());
@@ -157,7 +135,7 @@ public class VertexLocationConstraintTest {
Instance instance2 = getInstance(address2, 6789, hostname2);
Instance instance3 = getInstance(address3, 6789, hostname3);
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
scheduler.newInstanceAvailable(instance1);
scheduler.newInstanceAvailable(instance2);
scheduler.newInstanceAvailable(instance3);
@@ -168,7 +146,12 @@ public class VertexLocationConstraintTest {
jobVertex.setParallelism(2);
JobGraph jg = new JobGraph("test job", jobVertex);
- ExecutionGraph eg = new ExecutionGraph(jg.getJobID(), jg.getName(), jg.getJobConfiguration(), timeout);
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jg.getJobID(),
+ jg.getName(),
+ jg.getJobConfiguration(),
+ timeout);
eg.attachJobGraph(Collections.singletonList(jobVertex));
ExecutionJobVertex ejv = eg.getAllVertices().get(jobVertex.getID());
@@ -219,7 +202,7 @@ public class VertexLocationConstraintTest {
Instance instance2 = getInstance(address2, 6789, hostname2);
Instance instance3 = getInstance(address3, 6789, hostname3);
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
scheduler.newInstanceAvailable(instance1);
scheduler.newInstanceAvailable(instance2);
scheduler.newInstanceAvailable(instance3);
@@ -238,7 +221,12 @@ public class VertexLocationConstraintTest {
JobGraph jg = new JobGraph("test job", jobVertex1, jobVertex2);
- ExecutionGraph eg = new ExecutionGraph(jg.getJobID(), jg.getName(), jg.getJobConfiguration(), timeout);
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jg.getJobID(),
+ jg.getName(),
+ jg.getJobConfiguration(),
+ timeout);
eg.attachJobGraph(Arrays.asList(jobVertex1, jobVertex2));
ExecutionJobVertex ejv = eg.getAllVertices().get(jobVertex1.getID());
@@ -290,7 +278,7 @@ public class VertexLocationConstraintTest {
Instance instance1 = getInstance(address1, 6789, hostname1);
Instance instance2 = getInstance(address2, 6789, hostname2);
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
scheduler.newInstanceAvailable(instance1);
// prepare the execution graph
@@ -299,7 +287,12 @@ public class VertexLocationConstraintTest {
jobVertex.setParallelism(1);
JobGraph jg = new JobGraph("test job", jobVertex);
- ExecutionGraph eg = new ExecutionGraph(jg.getJobID(), jg.getName(), jg.getJobConfiguration(), timeout);
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jg.getJobID(),
+ jg.getName(),
+ jg.getJobConfiguration(),
+ timeout);
eg.attachJobGraph(Collections.singletonList(jobVertex));
ExecutionJobVertex ejv = eg.getAllVertices().get(jobVertex.getID());
@@ -343,7 +336,7 @@ public class VertexLocationConstraintTest {
Instance instance1 = getInstance(address1, 6789, hostname1);
Instance instance2 = getInstance(address2, 6789, hostname2);
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
scheduler.newInstanceAvailable(instance1);
// prepare the execution graph
@@ -362,7 +355,12 @@ public class VertexLocationConstraintTest {
jobVertex1.setSlotSharingGroup(sharingGroup);
jobVertex2.setSlotSharingGroup(sharingGroup);
- ExecutionGraph eg = new ExecutionGraph(jg.getJobID(), jg.getName(), jg.getJobConfiguration(), timeout);
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jg.getJobID(),
+ jg.getName(),
+ jg.getJobConfiguration(),
+ timeout);
eg.attachJobGraph(Arrays.asList(jobVertex1, jobVertex2));
ExecutionJobVertex ejv = eg.getAllVertices().get(jobVertex1.getID());
@@ -395,12 +393,17 @@ public class VertexLocationConstraintTest {
JobVertex vertex = new JobVertex("test vertex", new JobVertexID());
JobGraph jg = new JobGraph("test job", vertex);
- ExecutionGraph eg = new ExecutionGraph(jg.getJobID(), jg.getName(), jg.getJobConfiguration(), timeout);
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ jg.getJobID(),
+ jg.getName(),
+ jg.getJobConfiguration(),
+ timeout);
eg.attachJobGraph(Collections.singletonList(vertex));
ExecutionVertex ev = eg.getAllVertices().get(vertex.getID()).getTaskVertices()[0];
- Instance instance = ExecutionGraphTestUtils.getInstance(ActorRef.noSender());
+ Instance instance = ExecutionGraphTestUtils.getInstance(DummyInstanceGateway.INSTANCE);
ev.setLocationConstraintHosts(Collections.singletonList(instance));
assertNotNull(ev.getPreferredLocations());
@@ -431,6 +434,12 @@ public class VertexLocationConstraintTest {
when(connection.getHostname()).thenReturn(hostname);
when(connection.getFQDNHostname()).thenReturn(hostname);
- return new Instance(taskManager, connection, new InstanceID(), hardwareDescription, 1);
+ return new Instance(
+ new ExecutionGraphTestUtils.SimpleInstanceGateway(
+ TestingUtils.defaultExecutionContext()),
+ connection,
+ new InstanceID(),
+ hardwareDescription,
+ 1);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexSlotSharingTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexSlotSharingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexSlotSharingTest.java
index a1d6d03..d9e422c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexSlotSharingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexSlotSharingTest.java
@@ -30,6 +30,7 @@ import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.junit.Test;
public class VertexSlotSharingTest {
@@ -68,7 +69,11 @@ public class VertexSlotSharingTest {
List<JobVertex> vertices = new ArrayList<JobVertex>(Arrays.asList(v1, v2, v3, v4, v5));
- ExecutionGraph eg = new ExecutionGraph(new JobID(), "test job", new Configuration(),
+ ExecutionGraph eg = new ExecutionGraph(
+ TestingUtils.defaultExecutionContext(),
+ new JobID(),
+ "test job",
+ new Configuration(),
AkkaUtils.getDefaultTimeout());
eg.attachJobGraph(vertices);
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/instance/BaseTestingInstanceGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/BaseTestingInstanceGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/BaseTestingInstanceGateway.java
new file mode 100644
index 0000000..e9f8259
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/BaseTestingInstanceGateway.java
@@ -0,0 +1,94 @@
+/*
+ * 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.instance;
+
+import akka.actor.ActorPath;
+import akka.actor.ActorRef;
+import akka.dispatch.Futures;
+import scala.concurrent.ExecutionContext;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+
+import java.util.concurrent.Callable;
+
+/**
+ * Abstract base class for testing {@link InstanceGateway} instances. The implementing subclass
+ * only has to provide an implementation for handleMessage which contains the logic to treat
+ * different messages.
+ */
+abstract public class BaseTestingInstanceGateway implements InstanceGateway {
+ /**
+ * {@link ExecutionContext} which is used to execute the futures.
+ */
+ private final ExecutionContext executionContext;
+
+ public BaseTestingInstanceGateway(ExecutionContext executionContext) {
+ this.executionContext = executionContext;
+ }
+
+ @Override
+ public Future<Object> ask(Object message, FiniteDuration timeout) {
+ try {
+ final Object result = handleMessage(message);
+
+ return Futures.future(new Callable<Object>() {
+ @Override
+ public Object call() throws Exception {
+ return result;
+ }
+ }, executionContext);
+
+ } catch (final Exception e) {
+ // if an exception occurred in the handleMessage method then return it as part of the future
+ return Futures.future(new Callable<Object>() {
+ @Override
+ public Object call() throws Exception {
+ throw e;
+ }
+ }, executionContext);
+ }
+ }
+
+ /**
+ * Handles the supported messages by this InstanceGateway
+ *
+ * @param message Message to handle
+ * @return Result
+ * @throws Exception
+ */
+ abstract public Object handleMessage(Object message) throws Exception;
+
+ @Override
+ public void tell(Object message) {}
+
+ @Override
+ public void forward(Object message, ActorRef sender) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Future<Object> retry(Object message, int numberRetries, FiniteDuration timeout, ExecutionContext executionContext) {
+ return ask(message, timeout);
+ }
+
+ @Override
+ public String path() {
+ throw new UnsupportedOperationException();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/instance/DummyInstanceGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/DummyInstanceGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/DummyInstanceGateway.java
new file mode 100644
index 0000000..5941201
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/DummyInstanceGateway.java
@@ -0,0 +1,57 @@
+/*
+ * 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.instance;
+
+import akka.actor.ActorPath;
+import akka.actor.ActorRef;
+import scala.concurrent.ExecutionContext;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+
+/**
+ * Dummy {@link InstanceGateway} implementation used for testing.
+ */
+public class DummyInstanceGateway implements InstanceGateway {
+ public static final DummyInstanceGateway INSTANCE = new DummyInstanceGateway();
+
+ @Override
+ public Future<Object> ask(Object message, FiniteDuration timeout) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void tell(Object message) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void forward(Object message, ActorRef sender) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Future<Object> retry(Object message, int numberRetries, FiniteDuration timeout, ExecutionContext executionContext) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String path() {
+ return "DummyInstanceGateway";
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/instance/InstanceTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/InstanceTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/InstanceTest.java
index 595ac7e..c075a17 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/InstanceTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/InstanceTest.java
@@ -23,7 +23,6 @@ import static org.junit.Assert.*;
import java.lang.reflect.Method;
import java.net.InetAddress;
-import akka.actor.ActorRef;
import org.apache.flink.api.common.JobID;
import org.junit.Test;
@@ -39,7 +38,7 @@ public class InstanceTest {
InetAddress address = InetAddress.getByName("127.0.0.1");
InstanceConnectionInfo connection = new InstanceConnectionInfo(address, 10001);
- Instance instance = new Instance(ActorRef.noSender(), connection, new InstanceID(), hardwareDescription, 4);
+ Instance instance = new Instance(DummyInstanceGateway.INSTANCE, connection, new InstanceID(), hardwareDescription, 4);
assertEquals(4, instance.getTotalNumberOfSlots());
assertEquals(4, instance.getNumberOfAvailableSlots());
@@ -100,7 +99,7 @@ public class InstanceTest {
InetAddress address = InetAddress.getByName("127.0.0.1");
InstanceConnectionInfo connection = new InstanceConnectionInfo(address, 10001);
- Instance instance = new Instance(ActorRef.noSender(), connection, new InstanceID(), hardwareDescription, 3);
+ Instance instance = new Instance(DummyInstanceGateway.INSTANCE, connection, new InstanceID(), hardwareDescription, 3);
assertEquals(3, instance.getNumberOfAvailableSlots());
@@ -130,7 +129,7 @@ public class InstanceTest {
InetAddress address = InetAddress.getByName("127.0.0.1");
InstanceConnectionInfo connection = new InstanceConnectionInfo(address, 10001);
- Instance instance = new Instance(ActorRef.noSender(), connection, new InstanceID(), hardwareDescription, 3);
+ Instance instance = new Instance(DummyInstanceGateway.INSTANCE, connection, new InstanceID(), hardwareDescription, 3);
assertEquals(3, instance.getNumberOfAvailableSlots());
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SimpleSlotTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SimpleSlotTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SimpleSlotTest.java
index e075ab6..29ec7b5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SimpleSlotTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SimpleSlotTest.java
@@ -23,8 +23,6 @@ import static org.junit.Assert.*;
import java.net.InetAddress;
-import akka.actor.ActorRef;
-
import org.apache.flink.runtime.executiongraph.Execution;
import org.apache.flink.api.common.JobID;
@@ -148,7 +146,7 @@ public class SimpleSlotTest {
InetAddress address = InetAddress.getByName("127.0.0.1");
InstanceConnectionInfo connection = new InstanceConnectionInfo(address, 10001);
- Instance instance = new Instance(ActorRef.noSender(), connection, new InstanceID(), hardwareDescription, 1);
+ Instance instance = new Instance(DummyInstanceGateway.INSTANCE, connection, new InstanceID(), hardwareDescription, 1);
return instance.allocateSimpleSlot(new JobID());
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
index 7739dea..6affdcc 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
@@ -27,6 +27,7 @@ import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.netty.NettyConfig;
import org.apache.flink.runtime.net.NetUtils;
import org.apache.flink.runtime.taskmanager.NetworkEnvironmentConfiguration;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.junit.Test;
import org.mockito.Mockito;
import scala.Some;
@@ -58,7 +59,10 @@ public class NetworkEnvironmentTest {
NUM_BUFFERS, BUFFER_SIZE, IOManager.IOMode.SYNC, new Some<NettyConfig>(nettyConf),
new Tuple2<Integer, Integer>(0, 0));
- NetworkEnvironment env = new NetworkEnvironment(new FiniteDuration(30, TimeUnit.SECONDS), config);
+ NetworkEnvironment env = new NetworkEnvironment(
+ TestingUtils.defaultExecutionContext(),
+ new FiniteDuration(30, TimeUnit.SECONDS),
+ config);
assertFalse(env.isShutdown());
assertFalse(env.isAssociated());
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduleWithCoLocationHintTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduleWithCoLocationHintTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduleWithCoLocationHintTest.java
index a15e477..676b2a3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduleWithCoLocationHintTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduleWithCoLocationHintTest.java
@@ -27,41 +27,22 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import akka.actor.ActorSystem;
-import akka.testkit.JavaTestKit;
-
import org.apache.flink.runtime.instance.SimpleSlot;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.testingUtils.TestingUtils;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
import org.junit.Test;
public class ScheduleWithCoLocationHintTest {
- private static ActorSystem system;
-
- @BeforeClass
- public static void setup(){
- system = ActorSystem.create("TestingActorSystem", TestingUtils.testConfig());
- TestingUtils.setCallingThreadDispatcher(system);
- }
-
- @AfterClass
- public static void teardown(){
- TestingUtils.setGlobalExecutionContext();
- JavaTestKit.shutdownActorSystem(system);
- }
-
@Test
public void scheduleAllSharedAndCoLocated() {
try {
JobVertexID jid1 = new JobVertexID();
JobVertexID jid2 = new JobVertexID();
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
scheduler.newInstanceAvailable(getRandomInstance(2));
scheduler.newInstanceAvailable(getRandomInstance(2));
@@ -187,7 +168,7 @@ public class ScheduleWithCoLocationHintTest {
JobVertexID jid3 = new JobVertexID();
JobVertexID jid4 = new JobVertexID();
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
Instance i1 = getRandomInstance(1);
Instance i2 = getRandomInstance(1);
@@ -231,7 +212,7 @@ public class ScheduleWithCoLocationHintTest {
JobVertexID jid2 = new JobVertexID();
JobVertexID jid3 = new JobVertexID();
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
Instance i1 = getRandomInstance(1);
Instance i2 = getRandomInstance(1);
@@ -276,7 +257,7 @@ public class ScheduleWithCoLocationHintTest {
JobVertexID jid3 = new JobVertexID();
JobVertexID jid4 = new JobVertexID();
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
scheduler.newInstanceAvailable(getRandomInstance(1));
scheduler.newInstanceAvailable(getRandomInstance(1));
scheduler.newInstanceAvailable(getRandomInstance(1));
@@ -338,7 +319,7 @@ public class ScheduleWithCoLocationHintTest {
JobVertexID jid2 = new JobVertexID();
JobVertexID jid3 = new JobVertexID();
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
Instance i1 = getRandomInstance(1);
Instance i2 = getRandomInstance(1);
@@ -407,7 +388,7 @@ public class ScheduleWithCoLocationHintTest {
JobVertexID jid1 = new JobVertexID();
JobVertexID jid2 = new JobVertexID();
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
Instance i1 = getRandomInstance(1);
Instance i2 = getRandomInstance(1);
@@ -461,7 +442,7 @@ public class ScheduleWithCoLocationHintTest {
JobVertexID jid2 = new JobVertexID();
JobVertexID jidx = new JobVertexID();
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
Instance i1 = getRandomInstance(1);
Instance i2 = getRandomInstance(1);
@@ -519,7 +500,7 @@ public class ScheduleWithCoLocationHintTest {
JobVertexID jid1 = new JobVertexID();
JobVertexID jid2 = new JobVertexID();
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
Instance i1 = getRandomInstance(1);
Instance i2 = getRandomInstance(1);
@@ -581,7 +562,7 @@ public class ScheduleWithCoLocationHintTest {
JobVertexID jid1 = new JobVertexID();
JobVertexID jid2 = new JobVertexID();
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.directExecutionContext());
Instance i1 = getRandomInstance(1);
Instance i2 = getRandomInstance(1);
http://git-wip-us.apache.org/repos/asf/flink/blob/2ccb5fdb/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerIsolatedTasksTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerIsolatedTasksTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerIsolatedTasksTest.java
index d19299b..2ee53d9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerIsolatedTasksTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerIsolatedTasksTest.java
@@ -25,11 +25,7 @@ import static org.apache.flink.runtime.jobmanager.scheduler.SchedulerTestUtils.g
import static org.junit.Assert.*;
import org.apache.flink.runtime.instance.SimpleSlot;
-import akka.actor.ActorSystem;
-import akka.testkit.JavaTestKit;
import org.apache.flink.runtime.testingUtils.TestingUtils;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
import org.junit.Test;
import java.util.ArrayList;
@@ -47,24 +43,11 @@ import org.apache.flink.runtime.instance.Instance;
* Tests for the {@link Scheduler} when scheduling individual tasks.
*/
public class SchedulerIsolatedTasksTest {
- private static ActorSystem system;
- @BeforeClass
- public static void setup(){
- system = ActorSystem.create("TestingActorSystem", TestingUtils.testConfig());
- TestingUtils.setCallingThreadDispatcher(system);
- }
-
- @AfterClass
- public static void teardown(){
- TestingUtils.setGlobalExecutionContext();
- JavaTestKit.shutdownActorSystem(system);
- }
-
@Test
public void testAddAndRemoveInstance() {
try {
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
Instance i1 = getRandomInstance(2);
Instance i2 = getRandomInstance(2);
@@ -128,7 +111,7 @@ public class SchedulerIsolatedTasksTest {
@Test
public void testScheduleImmediately() {
try {
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
assertEquals(0, scheduler.getNumberOfAvailableSlots());
scheduler.newInstanceAvailable(getRandomInstance(2));
@@ -197,12 +180,10 @@ public class SchedulerIsolatedTasksTest {
final int NUM_SLOTS_PER_INSTANCE = 3;
final int NUM_TASKS_TO_SCHEDULE = 2000;
- TestingUtils.setGlobalExecutionContext();
-
try {
// note: since this test asynchronously releases slots, the executor needs release workers.
// doing the release call synchronous can lead to a deadlock
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
for (int i = 0; i < NUM_INSTANCES; i++) {
scheduler.newInstanceAvailable(getRandomInstance((int) (Math.random() * NUM_SLOTS_PER_INSTANCE) + 1));
@@ -287,15 +268,13 @@ public class SchedulerIsolatedTasksTest {
} catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
- } finally {
- TestingUtils.setCallingThreadDispatcher(system);
}
}
@Test
public void testScheduleWithDyingInstances() {
try {
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
Instance i1 = getRandomInstance(2);
Instance i2 = getRandomInstance(2);
@@ -355,7 +334,7 @@ public class SchedulerIsolatedTasksTest {
@Test
public void testSchedulingLocation() {
try {
- Scheduler scheduler = new Scheduler();
+ Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
Instance i1 = getRandomInstance(2);
Instance i2 = getRandomInstance(2);