You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2019/02/20 09:15:58 UTC
[flink] branch master updated: [FLINK-11663] Remove control flow
break point from Execution#releaseAssignedResource
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 76ecd7a [FLINK-11663] Remove control flow break point from Execution#releaseAssignedResource
76ecd7a is described below
commit 76ecd7abd239d6690e5e6ea9afe8262c7a389f26
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Tue Feb 19 16:59:38 2019 +0100
[FLINK-11663] Remove control flow break point from Execution#releaseAssignedResource
Since LogicalSlot#releaseSlot returns a future which is always completed from the main thread
it is no longer necessary to merge the future back into the main thread by using FutureUtils#
whenCompleteAsyncIfNotDone. The advantage is that LogicalSlot#releaseSlot will atomically release
the assigned Execution.
This closes #7755.
---
.../flink/runtime/executiongraph/Execution.java | 7 ++--
.../executiongraph/ExecutionGraphRestartTest.java | 10 +++--
.../runtime/executiongraph/ExecutionTest.java | 45 ++++++++++++++++++++++
3 files changed, 55 insertions(+), 7 deletions(-)
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 91b3d2d..6e1e8d2 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
@@ -1307,10 +1307,9 @@ public class Execution implements AccessExecution, Archiveable<ArchivedExecution
ComponentMainThreadExecutor jobMasterMainThreadExecutor =
getVertex().getExecutionGraph().getJobMasterMainThreadExecutor();
- FutureUtils.whenCompleteAsyncIfNotDone(
- slot.releaseSlot(cause),
- jobMasterMainThreadExecutor,
- (Object ignored, Throwable throwable) -> {
+ slot.releaseSlot(cause)
+ .whenComplete((Object ignored, Throwable throwable) -> {
+ jobMasterMainThreadExecutor.assertRunningInMainThread();
if (throwable != null) {
releaseFuture.completeExceptionally(throwable);
} else {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java
index a5477fe..6192426 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java
@@ -195,6 +195,8 @@ public class ExecutionGraphRestartTest extends TestLogger {
new InfiniteDelayRestartStrategy(),
scheduler);
+ executionGraph.start(TestingComponentMainThreadExecutorServiceAdapter.forMainThread());
+
JobVertex jobVertex = new JobVertex("NoOpInvokable");
jobVertex.setInvokableClass(NoOpInvokable.class);
jobVertex.setParallelism(NUM_TASKS);
@@ -212,8 +214,6 @@ public class ExecutionGraphRestartTest extends TestLogger {
// Kill the instance and wait for the job to restart
instance.markDead();
- Assert.assertEquals(JobStatus.RESTARTING, executionGraph.getState());
-
assertEquals(JobStatus.RESTARTING, executionGraph.getState());
// If we fail when being in RESTARTING, then we should try to restart again
@@ -797,7 +797,7 @@ public class ExecutionGraphRestartTest extends TestLogger {
}
private static ExecutionGraph newExecutionGraph(RestartStrategy restartStrategy, SlotProvider slotProvider) throws IOException {
- return new ExecutionGraph(
+ final ExecutionGraph executionGraph = new ExecutionGraph(
TestingUtils.defaultExecutor(),
TestingUtils.defaultExecutor(),
new JobID(),
@@ -807,6 +807,10 @@ public class ExecutionGraphRestartTest extends TestLogger {
AkkaUtils.getDefaultTimeout(),
restartStrategy,
slotProvider);
+
+ executionGraph.start(TestingComponentMainThreadExecutorServiceAdapter.forMainThread());
+
+ return executionGraph;
}
private void restartAfterFailure(ExecutionGraph eg, FiniteDuration timeout, boolean haltAfterRestart) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionTest.java
index 4881384..d69b0c3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionTest.java
@@ -39,6 +39,7 @@ import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.flink.runtime.testtasks.NoOpInvokable;
+import org.apache.flink.util.FlinkException;
import org.apache.flink.util.TestLogger;
import org.junit.ClassRule;
@@ -51,6 +52,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import static org.hamcrest.Matchers.containsInAnyOrder;
@@ -59,6 +61,7 @@ import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.sameInstance;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThat;
@@ -503,6 +506,48 @@ public class ExecutionTest extends TestLogger {
assertThat(returnedSlotFuture.get(), is(equalTo(slotRequestIdFuture.get())));
}
+ /**
+ * Tests that a slot release will atomically release the assigned {@link Execution}.
+ */
+ @Test
+ public void testSlotReleaseAtomicallyReleasesExecution() throws Exception {
+ final JobVertex jobVertex = createNoOpJobVertex();
+
+ final SingleSlotTestingSlotOwner slotOwner = new SingleSlotTestingSlotOwner();
+ final SingleLogicalSlot slot = ExecutionGraphSchedulingTest.createSingleLogicalSlot(
+ slotOwner,
+ new SimpleAckingTaskManagerGateway(),
+ new SlotRequestId());
+ final CompletableFuture<LogicalSlot> slotFuture = CompletableFuture.completedFuture(slot);
+
+ final CountDownLatch slotRequestLatch = new CountDownLatch(1);
+ final TestingSlotProvider slotProvider = new TestingSlotProvider(slotRequestId -> {
+ slotRequestLatch.countDown();
+ return slotFuture;
+ });
+ final ExecutionGraph executionGraph = ExecutionGraphTestUtils.createSimpleTestGraph(
+ new JobID(),
+ slotProvider,
+ new NoRestartStrategy(),
+ jobVertex);
+
+ final Execution execution = executionGraph.getJobVertex(jobVertex.getID()).getTaskVertices()[0].getCurrentExecutionAttempt();
+
+ executionGraph.start(testMainThreadUtil.getMainThreadExecutor());
+ testMainThreadUtil.execute(executionGraph::scheduleForExecution);
+
+ // wait until the slot has been requested
+ slotRequestLatch.await();
+
+ testMainThreadUtil.execute(() -> {
+ assertThat(execution.getAssignedResource(), is(sameInstance(slot)));
+
+ slot.release(new FlinkException("Test exception"));
+
+ assertThat(execution.getReleaseFuture().isDone(), is(true));
+ });
+ }
+
@Nonnull
private JobVertex createNoOpJobVertex() {
final JobVertex jobVertex = new JobVertex("Test vertex", new JobVertexID());