[GitHub] tillrohrmann commented on a change in pull request #7768: [FLINK-11541][tests] Reduce scale of heavy deployment e2e test for hi…

[GitBox <gi...@apache.org>]

tillrohrmann commented on a change in pull request #7768: [FLINK-11541][tests] Reduce scale of heavy deployment e2e test for hi…
URL: https://github.com/apache/flink/pull/7768#discussion_r259392385
 
 

 ##########
 File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
 ##########
 @@ -631,27 +628,29 @@ public void deploy() throws JobException {
 
 			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);
+			executor.execute(() -> {
+				// 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.
+				taskManagerGateway
+					.submitTask(deployment, rpcTimeout)
+					.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);
 
 Review comment:
   I think it would be better to wrap the `taskManagerGateway.submitTask` into a future because we would also catch exceptions originating from this call:
   ```
   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);
   ```

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services