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 2020/10/20 21:03:09 UTC
[flink] 04/09: [hotfix] Correct TestingDispatcherGatewayService
closeAsync behavior
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
commit d5f8a7c719eee152b8342bcba26f62903ee3d16d
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Tue Oct 20 11:34:52 2020 +0200
[hotfix] Correct TestingDispatcherGatewayService closeAsync behavior
Let TestingDispatcherGatewayService complete the termination future when calling closeAsync.
---
.../runner/AbstractDispatcherLeaderProcess.java | 4 +-
.../runner/SessionDispatcherLeaderProcessTest.java | 3 +-
.../runner/TestingDispatcherGatewayService.java | 45 +++++++++++++---------
3 files changed, 30 insertions(+), 22 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/AbstractDispatcherLeaderProcess.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/AbstractDispatcherLeaderProcess.java
index 66d7d6c..0ecdb8b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/AbstractDispatcherLeaderProcess.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/AbstractDispatcherLeaderProcess.java
@@ -139,6 +139,8 @@ public abstract class AbstractDispatcherLeaderProcess implements DispatcherLeade
private void closeInternal() {
log.info("Stopping {}.", getClass().getSimpleName());
+ state = State.STOPPED;
+
final CompletableFuture<Void> dispatcherServiceTerminationFuture = closeDispatcherService();
final CompletableFuture<Void> onCloseTerminationFuture = FutureUtils.composeAfterwards(
@@ -148,8 +150,6 @@ public abstract class AbstractDispatcherLeaderProcess implements DispatcherLeade
FutureUtils.forward(
onCloseTerminationFuture,
this.terminationFuture);
-
- state = State.STOPPED;
}
private CompletableFuture<Void> closeDispatcherService() {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessTest.java
index 0a34399..799b91b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessTest.java
@@ -147,7 +147,8 @@ public class SessionDispatcherLeaderProcessTest extends TestLogger {
final CompletableFuture<Void> dispatcherServiceTerminationFuture = new CompletableFuture<>();
dispatcherServiceFactory = TestingDispatcherServiceFactory.newBuilder()
.setCreateFunction((ignoredA, ignoredB, ignoredC) -> TestingDispatcherGatewayService.newBuilder()
- .setTerminationFutureSupplier(() -> dispatcherServiceTerminationFuture)
+ .setTerminationFuture(dispatcherServiceTerminationFuture)
+ .withManualTerminationFutureCompletion()
.build())
.build();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/TestingDispatcherGatewayService.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/TestingDispatcherGatewayService.java
index ea999d8..b27c835 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/TestingDispatcherGatewayService.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/TestingDispatcherGatewayService.java
@@ -26,31 +26,29 @@ import org.apache.flink.runtime.webmonitor.TestingDispatcherGateway;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
-import java.util.function.Supplier;
class TestingDispatcherGatewayService implements AbstractDispatcherLeaderProcess.DispatcherGatewayService {
- private final Object lock = new Object();
-
- private final Supplier<CompletableFuture<Void>> terminationFutureSupplier;
-
private final Function<JobID, CompletableFuture<Void>> onRemovedJobGraphFunction;
private final DispatcherGateway dispatcherGateway;
private final CompletableFuture<ApplicationStatus> shutDownFuture;
- private CompletableFuture<Void> terminationFuture;
+ private final CompletableFuture<Void> terminationFuture;
+ private final boolean completeTerminationFutureOnClose;
private TestingDispatcherGatewayService(
- Supplier<CompletableFuture<Void>> terminationFutureSupplier,
+ CompletableFuture<Void> terminationFuture,
Function<JobID, CompletableFuture<Void>> onRemovedJobGraphFunction,
DispatcherGateway dispatcherGateway,
- CompletableFuture<ApplicationStatus> shutDownFuture) {
- this.terminationFutureSupplier = terminationFutureSupplier;
+ CompletableFuture<ApplicationStatus> shutDownFuture,
+ boolean completeTerminationFutureOnClose) {
+ this.terminationFuture = terminationFuture;
this.onRemovedJobGraphFunction = onRemovedJobGraphFunction;
this.dispatcherGateway = dispatcherGateway;
this.shutDownFuture = shutDownFuture;
+ this.completeTerminationFutureOnClose = completeTerminationFutureOnClose;
}
@Override
@@ -68,15 +66,17 @@ class TestingDispatcherGatewayService implements AbstractDispatcherLeaderProcess
return shutDownFuture;
}
+ public CompletableFuture<Void> getTerminationFuture() {
+ return terminationFuture;
+ }
+
@Override
public CompletableFuture<Void> closeAsync() {
- synchronized (lock) {
- if (terminationFuture == null) {
- terminationFuture = terminationFutureSupplier.get();
- }
-
- return terminationFuture;
+ if (completeTerminationFutureOnClose) {
+ terminationFuture.complete(null);
}
+
+ return terminationFuture;
}
public static Builder newBuilder() {
@@ -85,7 +85,7 @@ class TestingDispatcherGatewayService implements AbstractDispatcherLeaderProcess
public static class Builder {
- private Supplier<CompletableFuture<Void>> terminationFutureSupplier = FutureUtils::completedVoidFuture;
+ private CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
private Function<JobID, CompletableFuture<Void>> onRemovedJobGraphFunction = ignored -> FutureUtils.completedVoidFuture();
@@ -93,10 +93,12 @@ class TestingDispatcherGatewayService implements AbstractDispatcherLeaderProcess
private CompletableFuture<ApplicationStatus> shutDownFuture = new CompletableFuture<>();
+ private boolean completeTerminationFutureOnClose = true;
+
private Builder() {}
- public Builder setTerminationFutureSupplier(Supplier<CompletableFuture<Void>> terminationFutureSupplier) {
- this.terminationFutureSupplier = terminationFutureSupplier;
+ public Builder setTerminationFuture(CompletableFuture<Void> terminationFuture) {
+ this.terminationFuture = terminationFuture;
return this;
}
@@ -115,8 +117,13 @@ class TestingDispatcherGatewayService implements AbstractDispatcherLeaderProcess
return this;
}
+ public Builder withManualTerminationFutureCompletion() {
+ completeTerminationFutureOnClose = false;
+ return this;
+ }
+
public TestingDispatcherGatewayService build() {
- return new TestingDispatcherGatewayService(terminationFutureSupplier, onRemovedJobGraphFunction, dispatcherGateway, shutDownFuture);
+ return new TestingDispatcherGatewayService(terminationFuture, onRemovedJobGraphFunction, dispatcherGateway, shutDownFuture, completeTerminationFutureOnClose);
}
}
}