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:
[email protected]
With regards,
Apache Git Services
