You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2019/04/04 13:38:07 UTC
[flink] branch master updated: [FLINK-11897][tests] Wait until all
tasks are submitted
This is an automated email from the ASF dual-hosted git repository.
chesnay 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 8a12e0f [FLINK-11897][tests] Wait until all tasks are submitted
8a12e0f is described below
commit 8a12e0f63de90375653a030304522ae0600cb3f9
Author: chunpinghe <ch...@gmail.com>
AuthorDate: Thu Apr 4 21:37:56 2019 +0800
[FLINK-11897][tests] Wait until all tasks are submitted
---
.../executiongraph/ExecutionGraphSuspendTest.java | 9 +++++----
.../InteractionsCountingTaskManagerGateway.java | 20 ++++++++++++++++++++
2 files changed, 25 insertions(+), 4 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSuspendTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSuspendTest.java
index d167a64..2d5dd67 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSuspendTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSuspendTest.java
@@ -69,8 +69,8 @@ public class ExecutionGraphSuspendTest extends TestLogger {
*/
@Test
public void testSuspendedOutOfDeploying() throws Exception {
- final InteractionsCountingTaskManagerGateway gateway = new InteractionsCountingTaskManagerGateway();
final int parallelism = 10;
+ final InteractionsCountingTaskManagerGateway gateway = new InteractionsCountingTaskManagerGateway(parallelism);
final ExecutionGraph eg = createExecutionGraph(gateway, parallelism);
eg.scheduleForExecution();
@@ -91,8 +91,8 @@ public class ExecutionGraphSuspendTest extends TestLogger {
*/
@Test
public void testSuspendedOutOfRunning() throws Exception {
- final InteractionsCountingTaskManagerGateway gateway = new InteractionsCountingTaskManagerGateway();
final int parallelism = 10;
+ final InteractionsCountingTaskManagerGateway gateway = new InteractionsCountingTaskManagerGateway(parallelism);
final ExecutionGraph eg = createExecutionGraph(gateway, parallelism);
eg.scheduleForExecution();
@@ -115,8 +115,8 @@ public class ExecutionGraphSuspendTest extends TestLogger {
*/
@Test
public void testSuspendedOutOfFailing() throws Exception {
- final InteractionsCountingTaskManagerGateway gateway = new InteractionsCountingTaskManagerGateway();
final int parallelism = 10;
+ final InteractionsCountingTaskManagerGateway gateway = new InteractionsCountingTaskManagerGateway(parallelism);
final ExecutionGraph eg = createExecutionGraph(gateway, parallelism);
eg.scheduleForExecution();
@@ -167,8 +167,8 @@ public class ExecutionGraphSuspendTest extends TestLogger {
*/
@Test
public void testSuspendedOutOfCanceling() throws Exception {
- final InteractionsCountingTaskManagerGateway gateway = new InteractionsCountingTaskManagerGateway();
final int parallelism = 10;
+ final InteractionsCountingTaskManagerGateway gateway = new InteractionsCountingTaskManagerGateway(parallelism);
final ExecutionGraph eg = createExecutionGraph(gateway, parallelism);
eg.scheduleForExecution();
@@ -247,6 +247,7 @@ public class ExecutionGraphSuspendTest extends TestLogger {
// ------------------------------------------------------------------------
private static void ensureCannotLeaveSuspendedState(ExecutionGraph eg, InteractionsCountingTaskManagerGateway gateway) {
+ gateway.waitUntilAllTasksAreSubmitted();
assertEquals(JobStatus.SUSPENDED, eg.getState());
gateway.resetCounts();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/InteractionsCountingTaskManagerGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/InteractionsCountingTaskManagerGateway.java
index feab301..b8f4a73 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/InteractionsCountingTaskManagerGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/InteractionsCountingTaskManagerGateway.java
@@ -24,6 +24,7 @@ import org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGate
import org.apache.flink.runtime.messages.Acknowledge;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
class InteractionsCountingTaskManagerGateway extends SimpleAckingTaskManagerGateway {
@@ -32,6 +33,16 @@ class InteractionsCountingTaskManagerGateway extends SimpleAckingTaskManagerGate
private final AtomicInteger submitTaskCount = new AtomicInteger(0);
+ private CountDownLatch submitLatch;
+
+ public InteractionsCountingTaskManagerGateway() {
+ submitLatch = new CountDownLatch(0);
+ }
+
+ public InteractionsCountingTaskManagerGateway(final int expectedSubmitCount) {
+ this.submitLatch = new CountDownLatch(expectedSubmitCount);
+ }
+
@Override
public CompletableFuture<Acknowledge> cancelTask(ExecutionAttemptID executionAttemptID, Time timeout) {
cancelTaskCount.incrementAndGet();
@@ -41,6 +52,7 @@ class InteractionsCountingTaskManagerGateway extends SimpleAckingTaskManagerGate
@Override
public CompletableFuture<Acknowledge> submitTask(TaskDeploymentDescriptor tdd, Time timeout) {
submitTaskCount.incrementAndGet();
+ submitLatch.countDown();
return CompletableFuture.completedFuture(Acknowledge.get());
}
@@ -60,4 +72,12 @@ class InteractionsCountingTaskManagerGateway extends SimpleAckingTaskManagerGate
int getInteractionsCount() {
return cancelTaskCount.get() + submitTaskCount.get();
}
+
+ void waitUntilAllTasksAreSubmitted() {
+ try {
+ submitLatch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
}