[flink] branch master updated: [FLINK-11541] Run submitTask call in future executor

[sr...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

srichter 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 42d9df8  [FLINK-11541] Run submitTask call in future executor
42d9df8 is described below

commit 42d9df8fa20a907c16c1675b48ea4151188a042a
Author: Stefan Richter <s....@data-artisans.com>
AuthorDate: Fri Feb 22 18:17:16 2019 +0100

    [FLINK-11541] Run submitTask call in future executor
    
    This closes #7768.
---
 .../test-scripts/test_heavy_deployment.sh          |  4 +--
 .../flink/runtime/executiongraph/Execution.java    | 42 +++++++++++-----------
 .../ExecutionGraphDeploymentTest.java              |  5 +--
 .../ExecutionGraphSchedulingTest.java              |  4 +--
 .../executiongraph/ExecutionGraphTestUtils.java    |  3 +-
 5 files changed, 31 insertions(+), 27 deletions(-)

diff --git a/flink-end-to-end-tests/test-scripts/test_heavy_deployment.sh b/flink-end-to-end-tests/test-scripts/test_heavy_deployment.sh
index cc1a64d..7bcc0f0 100755
--- a/flink-end-to-end-tests/test-scripts/test_heavy_deployment.sh
+++ b/flink-end-to-end-tests/test-scripts/test_heavy_deployment.sh
@@ -38,10 +38,10 @@ set_conf "taskmanager.numberOfTaskSlots" "10" # 10 slots per TM
 start_cluster # this also starts 1TM
 start_taskmanagers 9 # 1TM + 9TM = 10TM a 10 slots = 100 slots
 
-# This call will result in a deployment with state meta data of 100 x 100 x 50 union states x each 50 entries.
+# This call will result in a deployment with state meta data of 100 x 100 x 40 union states x each 40 entries.
 # We can scale up the numbers to make the test even heavier.
 $FLINK_DIR/bin/flink run ${TEST_PROGRAM_JAR} \
 --environment.max_parallelism 1024 --environment.parallelism 100 \
 --environment.restart_strategy fixed_delay --environment.restart_strategy.fixed_delay.attempts 3 \
 --state_backend.checkpoint_directory ${CHECKPOINT_DIR} \
---heavy_deployment_test.num_list_states_per_op 50 --heavy_deployment_test.num_partitions_per_list_state 50
+--heavy_deployment_test.num_list_states_per_op 40 --heavy_deployment_test.num_partitions_per_list_state 40
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 7874f0b..d5f091f 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
@@ -76,6 +76,7 @@ import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import static org.apache.flink.runtime.execution.ExecutionState.CANCELED;
@@ -516,9 +517,6 @@ public class Execution implements AccessExecution, Archiveable<ArchivedExecution
 
 			final SlotRequestId slotRequestId = new SlotRequestId();
 
-			final ComponentMainThreadExecutor mainThreadExecutor =
-				vertex.getExecutionGraph().getJobMasterMainThreadExecutor();
-
 			final CompletableFuture<LogicalSlot> logicalSlotFuture =
 				preferredLocationsFuture.thenCompose(
 					(Collection<TaskManagerLocation> preferredLocations) ->
@@ -631,27 +629,31 @@ public class Execution implements AccessExecution, Archiveable<ArchivedExecution
 
 			final TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway();
 
-			final CompletableFuture<Acknowledge> submitResultFuture = taskManagerGateway.submitTask(deployment, rpcTimeout);
-			ComponentMainThreadExecutor jobMasterMainThreadExecutor =
+			final ComponentMainThreadExecutor jobMasterMainThreadExecutor =
 				vertex.getExecutionGraph().getJobMasterMainThreadExecutor();
 
-			FutureUtils.whenCompleteAsyncIfNotDone(
-				submitResultFuture,
-				jobMasterMainThreadExecutor,
-				(ack, failure) -> {
-					// only respond to the failure case
-					if (failure != null) {
-						if (failure instanceof TimeoutException) {
-							String taskname = vertex.getTaskNameWithSubtaskIndex() + " (" + attemptId + ')';
 
-							markFailed(new Exception(
-								"Cannot deploy task " + taskname + " - TaskManager (" + getAssignedResourceLocation()
-									+ ") not responding after a rpcTimeout of " + rpcTimeout, failure));
-						} else {
-							markFailed(failure);
+			// We run the submission in the future executor so that the serialization of large TDDs does not block
+			// the main thread and sync back to the main thread once submission is completed.
+			CompletableFuture.supplyAsync(() -> taskManagerGateway.submitTask(deployment, rpcTimeout), executor)
+				.thenCompose(Function.identity())
+				.whenCompleteAsync(
+					(ack, failure) -> {
+						// only respond to the failure case
+						if (failure != null) {
+							if (failure instanceof TimeoutException) {
+								String taskname = vertex.getTaskNameWithSubtaskIndex() + " (" + attemptId + ')';
+
+								markFailed(new Exception(
+									"Cannot deploy task " + taskname + " - TaskManager (" + getAssignedResourceLocation()
+										+ ") not responding after a rpcTimeout of " + rpcTimeout, failure));
+							} else {
+								markFailed(failure);
+							}
 						}
-					}
-				});
+					},
+					jobMasterMainThreadExecutor);
+
 		}
 		catch (Throwable t) {
 			markFailed(t);
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 adb313f..b4cccdc 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
@@ -157,10 +157,11 @@ public class ExecutionGraphDeploymentTest extends TestLogger {
 				jobId,
 				"some job");
 
+			DirectScheduledExecutorService executor = new DirectScheduledExecutorService();
 			ExecutionGraph eg = new ExecutionGraph(
 				expectedJobInformation,
-				TestingUtils.defaultExecutor(),
-				TestingUtils.defaultExecutor(),
+				executor,
+				executor,
 				AkkaUtils.getDefaultTimeout(),
 				new NoRestartStrategy(),
 				new RestartAllStrategy.Factory(),
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSchedulingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSchedulingTest.java
index a7c6947..bbbf65d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSchedulingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSchedulingTest.java
@@ -51,6 +51,7 @@ import org.apache.flink.runtime.jobmaster.slotpool.SlotProvider;
 import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation;
 import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
 import org.apache.flink.runtime.testtasks.NoOpInvokable;
+import org.apache.flink.runtime.testutils.DirectScheduledExecutorService;
 import org.apache.flink.util.FlinkException;
 import org.apache.flink.util.TestLogger;
 
@@ -67,7 +68,6 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -85,7 +85,7 @@ import static org.junit.Assert.assertTrue;
  */
 public class ExecutionGraphSchedulingTest extends TestLogger {
 
-	private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+	private final ScheduledExecutorService executor = new DirectScheduledExecutorService();
 
 	@After
 	public void shutdown() {
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 d865426..4ea8c62 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
@@ -57,6 +57,7 @@ import org.apache.flink.runtime.messages.TaskMessages.SubmitTask;
 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.runtime.testutils.DirectScheduledExecutorService;
 import org.apache.flink.util.SerializedValue;
 
 import org.slf4j.Logger;
@@ -545,7 +546,7 @@ public class ExecutionGraphTestUtils {
 	}
 	
 	public static ExecutionJobVertex getExecutionVertex(JobVertexID id) throws Exception {
-		return getExecutionVertex(id, TestingUtils.defaultExecutor());
+		return getExecutionVertex(id, new DirectScheduledExecutorService());
 	}
 
 	// ------------------------------------------------------------------------