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:05 UTC
[flink] branch master updated (8fee135 -> 0ff7d7a)
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 8fee135 [FLINK-19732] Disable checkpointing in BATCH mode
new 75d15bd [hotfix] Add FlinkMatchers.willNotComplete to assert that a future times out
new 643d79b [hotfix] Let DispatcherResourceManagerComponent accept AutoCloseableAsync instead of WebMonitorEndpoint
new 08fbf82 [hotfix] Introduce ResourceManagerService to make DispatcherResourceManagerComponent easier to test
new d5f8a7c [hotfix] Correct TestingDispatcherGatewayService closeAsync behavior
new 806b9f3 [FLINK-19022][rpc] Log if RpcEndpoint failed its termination
new 9e31510 [FLINK-19022] Let TaskExecutor, Dispatcher, RM fail fatally if onStart throws a Throwable
new 11a4232 [FLINK-19022] Fail fatally if DispatcherGatewayService terminates unexpectedly in AbstractDispatcherLeaderProcess
new b813c99 [FLINK-19022] Fail DispatcherResourceManagerComponent fatally if ResourceManager terminates unexpectedly
new 0ff7d7a [FLINK-19022] Fail TaskManagerRunner fatally if TaskExecutor terminates unexpectedly
The 9 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../flink/runtime/dispatcher/Dispatcher.java | 4 +-
.../runner/AbstractDispatcherLeaderProcess.java | 25 +++++-
.../runner/DefaultDispatcherGatewayService.java | 5 ++
...tDispatcherResourceManagerComponentFactory.java | 5 +-
.../component/DefaultResourceManagerService.java | 56 +++++++++++++
.../DispatcherResourceManagerComponent.java | 40 +++++++--
.../runtime/resourcemanager/ResourceManager.java | 4 +-
.../flink/runtime/rpc/akka/AkkaRpcActor.java | 2 +
.../flink/runtime/taskexecutor/TaskExecutor.java | 4 +-
.../TaskExecutorToServiceAdapter.java} | 35 +++++---
.../runtime/taskexecutor/TaskManagerRunner.java | 77 ++++++++++++++++--
.../runner/SessionDispatcherLeaderProcessTest.java | 46 ++++++++++-
.../runner/TestingDispatcherGatewayService.java | 45 +++++-----
.../dispatcher/runner/TestingDispatcherRunner.java | 78 ++++++++++++++++++
.../DispatcherResourceManagerComponentTest.java | 95 ++++++++++++++++++++++
.../component/TestingResourceManagerService.java | 93 +++++++++++++++++++++
.../taskexecutor/TaskManagerRunnerTest.java | 62 +++++++++++++-
.../taskexecutor/TestingTaskExecutorService.java | 92 +++++++++++++++++++++
.../apache/flink/core/testutils/FlinkMatchers.java | 44 ++++++++++
.../JobManagerHAProcessFailureRecoveryITCase.java | 2 +-
20 files changed, 756 insertions(+), 58 deletions(-)
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultResourceManagerService.java
copy flink-runtime/src/{test/java/org/apache/flink/runtime/taskexecutor/NoOpPartitionProducerStateChecker.java => main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorToServiceAdapter.java} (52%)
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/TestingDispatcherRunner.java
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponentTest.java
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/component/TestingResourceManagerService.java
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorService.java
[flink] 02/09: [hotfix] Let DispatcherResourceManagerComponent
accept AutoCloseableAsync instead of WebMonitorEndpoint
Posted by tr...@apache.org.
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 643d79bf55bc35548f4d2b506fb01af3b53c4f95
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Mon Oct 19 16:50:14 2020 +0200
[hotfix] Let DispatcherResourceManagerComponent accept AutoCloseableAsync instead of WebMonitorEndpoint
This commit improves the testability of DispatcherResourceManagerComponent by allowing to pass
in a AutoCloseableAsync implementation.
---
.../entrypoint/component/DispatcherResourceManagerComponent.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
index 4d1b1f9..b19e709 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
@@ -61,7 +61,7 @@ public class DispatcherResourceManagerComponent implements AutoCloseableAsync {
private final LeaderRetrievalService resourceManagerRetrievalService;
@Nonnull
- private final WebMonitorEndpoint<?> webMonitorEndpoint;
+ private final AutoCloseableAsync webMonitorEndpoint;
private final CompletableFuture<Void> terminationFuture;
@@ -74,7 +74,7 @@ public class DispatcherResourceManagerComponent implements AutoCloseableAsync {
@Nonnull ResourceManager<?> resourceManager,
@Nonnull LeaderRetrievalService dispatcherLeaderRetrievalService,
@Nonnull LeaderRetrievalService resourceManagerRetrievalService,
- @Nonnull WebMonitorEndpoint<?> webMonitorEndpoint) {
+ @Nonnull AutoCloseableAsync webMonitorEndpoint) {
this.dispatcherRunner = dispatcherRunner;
this.resourceManager = resourceManager;
this.dispatcherLeaderRetrievalService = dispatcherLeaderRetrievalService;
[flink] 04/09: [hotfix] Correct TestingDispatcherGatewayService
closeAsync behavior
Posted by tr...@apache.org.
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);
}
}
}
[flink] 03/09: [hotfix] Introduce ResourceManagerService to make
DispatcherResourceManagerComponent easier to test
Posted by tr...@apache.org.
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 08fbf828959714e11438218a14fdc032c9c95032
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Mon Oct 19 16:55:48 2020 +0200
[hotfix] Introduce ResourceManagerService to make DispatcherResourceManagerComponent easier to test
By passing an interface instead of ResourceManager, it is easier to test the DispatcherResourceManagerComponent.
---
...tDispatcherResourceManagerComponentFactory.java | 2 +-
.../component/DefaultResourceManagerService.java | 51 ++++++++++++++++++++++
.../DispatcherResourceManagerComponent.java | 18 +++++---
3 files changed, 65 insertions(+), 6 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultDispatcherResourceManagerComponentFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultDispatcherResourceManagerComponentFactory.java
index 1386162..51bef98 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultDispatcherResourceManagerComponentFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultDispatcherResourceManagerComponentFactory.java
@@ -209,7 +209,7 @@ public class DefaultDispatcherResourceManagerComponentFactory implements Dispatc
return new DispatcherResourceManagerComponent(
dispatcherRunner,
- resourceManager,
+ DefaultResourceManagerService.createFor(resourceManager),
dispatcherLeaderRetrievalService,
resourceManagerRetrievalService,
webMonitorEndpoint);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultResourceManagerService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultResourceManagerService.java
new file mode 100644
index 0000000..e0e2246
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultResourceManagerService.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.entrypoint.component;
+
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Default {@link DispatcherResourceManagerComponent.ResourceManagerService} implementation which
+ * uses a {@link ResourceManager} instance.
+ */
+public class DefaultResourceManagerService implements DispatcherResourceManagerComponent.ResourceManagerService {
+
+ private final ResourceManager<?> resourceManager;
+
+ private DefaultResourceManagerService(ResourceManager<?> resourceManager) {
+ this.resourceManager = resourceManager;
+ }
+
+ @Override
+ public ResourceManagerGateway getGateway() {
+ return resourceManager.getSelfGateway(ResourceManagerGateway.class);
+ }
+
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ return resourceManager.closeAsync();
+ }
+
+ public static DefaultResourceManagerService createFor(ResourceManager<?> resourceManager) {
+ return new DefaultResourceManagerService(resourceManager);
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
index b19e709..ffd3139 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
@@ -52,7 +52,7 @@ public class DispatcherResourceManagerComponent implements AutoCloseableAsync {
private final DispatcherRunner dispatcherRunner;
@Nonnull
- private final ResourceManager<?> resourceManager;
+ private final ResourceManagerService resourceManagerService;
@Nonnull
private final LeaderRetrievalService dispatcherLeaderRetrievalService;
@@ -71,12 +71,12 @@ public class DispatcherResourceManagerComponent implements AutoCloseableAsync {
DispatcherResourceManagerComponent(
@Nonnull DispatcherRunner dispatcherRunner,
- @Nonnull ResourceManager<?> resourceManager,
+ @Nonnull ResourceManagerService resourceManagerService,
@Nonnull LeaderRetrievalService dispatcherLeaderRetrievalService,
@Nonnull LeaderRetrievalService resourceManagerRetrievalService,
@Nonnull AutoCloseableAsync webMonitorEndpoint) {
this.dispatcherRunner = dispatcherRunner;
- this.resourceManager = resourceManager;
+ this.resourceManagerService = resourceManagerService;
this.dispatcherLeaderRetrievalService = dispatcherLeaderRetrievalService;
this.resourceManagerRetrievalService = resourceManagerRetrievalService;
this.webMonitorEndpoint = webMonitorEndpoint;
@@ -120,7 +120,7 @@ public class DispatcherResourceManagerComponent implements AutoCloseableAsync {
final ApplicationStatus applicationStatus,
final @Nullable String diagnostics) {
- final ResourceManagerGateway selfGateway = resourceManager.getSelfGateway(ResourceManagerGateway.class);
+ final ResourceManagerGateway selfGateway = resourceManagerService.getGateway();
return selfGateway.deregisterApplication(applicationStatus, diagnostics).thenApply(ack -> null);
}
@@ -145,7 +145,7 @@ public class DispatcherResourceManagerComponent implements AutoCloseableAsync {
terminationFutures.add(dispatcherRunner.closeAsync());
- terminationFutures.add(resourceManager.closeAsync());
+ terminationFutures.add(resourceManagerService.closeAsync());
if (exception != null) {
terminationFutures.add(FutureUtils.completedExceptionally(exception));
@@ -168,4 +168,12 @@ public class DispatcherResourceManagerComponent implements AutoCloseableAsync {
public CompletableFuture<Void> closeAsync() {
return deregisterApplicationAndClose(ApplicationStatus.CANCELED, "DispatcherResourceManagerComponent has been closed.");
}
+
+ /**
+ * Service which gives access to a {@link ResourceManagerGateway}.
+ */
+ interface ResourceManagerService extends AutoCloseableAsync {
+
+ ResourceManagerGateway getGateway();
+ }
}
[flink] 09/09: [FLINK-19022] Fail TaskManagerRunner fatally if
TaskExecutor terminates unexpectedly
Posted by tr...@apache.org.
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 0ff7d7a6ddf28237797ff1c9ed6fc525a9245d90
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Tue Oct 20 13:47:38 2020 +0200
[FLINK-19022] Fail TaskManagerRunner fatally if TaskExecutor terminates unexpectedly
An unexpected termination of the TaskExecutor indicates that the system has reached an
invalid state. Therefore, the runner will fail fatally if this happens.
This closes #13319.
---
.../taskexecutor/TaskExecutorToServiceAdapter.java | 52 ++++++++++++
.../runtime/taskexecutor/TaskManagerRunner.java | 77 ++++++++++++++++--
.../taskexecutor/TaskManagerRunnerTest.java | 62 ++++++++++++++-
.../taskexecutor/TestingTaskExecutorService.java | 92 ++++++++++++++++++++++
.../JobManagerHAProcessFailureRecoveryITCase.java | 2 +-
5 files changed, 277 insertions(+), 8 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorToServiceAdapter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorToServiceAdapter.java
new file mode 100644
index 0000000..14512e3
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorToServiceAdapter.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.taskexecutor;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Simple adapter for {@link TaskExecutor} to adapt to {@link TaskManagerRunner.TaskExecutorService}.
+ */
+public class TaskExecutorToServiceAdapter implements TaskManagerRunner.TaskExecutorService {
+
+ private final TaskExecutor taskExecutor;
+
+ private TaskExecutorToServiceAdapter(TaskExecutor taskExecutor) {
+ this.taskExecutor = taskExecutor;
+ }
+
+ @Override
+ public void start() {
+ taskExecutor.start();
+ }
+
+ @Override
+ public CompletableFuture<Void> getTerminationFuture() {
+ return taskExecutor.getTerminationFuture();
+ }
+
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ return taskExecutor.closeAsync();
+ }
+
+ public static TaskExecutorToServiceAdapter createFor(TaskExecutor taskExecutor) {
+ return new TaskExecutorToServiceAdapter(taskExecutor);
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
index b296baf..1c5ce10 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
@@ -67,6 +67,7 @@ import org.apache.flink.util.AbstractID;
import org.apache.flink.util.AutoCloseableAsync;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.ExecutorUtils;
+import org.apache.flink.util.FlinkException;
import org.apache.flink.util.StringUtils;
import org.apache.flink.util.TaskManagerExceptionUtils;
@@ -121,13 +122,16 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
/** Executor used to run future callbacks. */
private final ExecutorService executor;
- private final TaskExecutor taskManager;
+ private final TaskExecutorService taskExecutorService;
private final CompletableFuture<Void> terminationFuture;
private boolean shutdown;
- public TaskManagerRunner(Configuration configuration, PluginManager pluginManager) throws Exception {
+ public TaskManagerRunner(
+ Configuration configuration,
+ PluginManager pluginManager,
+ TaskExecutorServiceFactory taskExecutorServiceFactory) throws Exception {
this.configuration = checkNotNull(configuration);
timeout = AkkaUtils.getTimeoutAsTime(configuration);
@@ -165,7 +169,7 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
ExternalResourceUtils.getExternalResourceAmountMap(configuration),
ExternalResourceUtils.externalResourceDriversFromConfig(configuration, pluginManager));
- taskManager = startTaskManager(
+ taskExecutorService = taskExecutorServiceFactory.createTaskExecutor(
this.configuration,
this.resourceId,
rpcService,
@@ -179,16 +183,27 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
this.terminationFuture = new CompletableFuture<>();
this.shutdown = false;
+ handleUnexpectedTaskExecutorServiceTermination();
MemoryLogger.startIfConfigured(LOG, configuration, terminationFuture);
}
+ private void handleUnexpectedTaskExecutorServiceTermination() {
+ taskExecutorService.getTerminationFuture().whenComplete((unused, throwable) -> {
+ synchronized (lock) {
+ if (!shutdown) {
+ onFatalError(new FlinkException("Unexpected termination of the TaskExecutor.", throwable));
+ }
+ }
+ });
+ }
+
// --------------------------------------------------------------------------------------------
// Lifecycle management
// --------------------------------------------------------------------------------------------
public void start() throws Exception {
- taskManager.start();
+ taskExecutorService.start();
}
@Override
@@ -197,7 +212,7 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
if (!shutdown) {
shutdown = true;
- final CompletableFuture<Void> taskManagerTerminationFuture = taskManager.closeAsync();
+ final CompletableFuture<Void> taskManagerTerminationFuture = taskExecutorService.closeAsync();
final CompletableFuture<Void> serviceTerminationFuture = FutureUtils.composeAfterwards(
taskManagerTerminationFuture,
@@ -317,7 +332,7 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
}
public static void runTaskManager(Configuration configuration, PluginManager pluginManager) throws Exception {
- final TaskManagerRunner taskManagerRunner = new TaskManagerRunner(configuration, pluginManager);
+ final TaskManagerRunner taskManagerRunner = new TaskManagerRunner(configuration, pluginManager, TaskManagerRunner::createTaskExecutorService);
taskManagerRunner.start();
}
@@ -350,6 +365,33 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
// Static utilities
// --------------------------------------------------------------------------------------------
+ public static TaskExecutorService createTaskExecutorService(
+ Configuration configuration,
+ ResourceID resourceID,
+ RpcService rpcService,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ MetricRegistry metricRegistry,
+ BlobCacheService blobCacheService,
+ boolean localCommunicationOnly,
+ ExternalResourceInfoProvider externalResourceInfoProvider,
+ FatalErrorHandler fatalErrorHandler) throws Exception {
+
+ final TaskExecutor taskExecutor = startTaskManager(
+ configuration,
+ resourceID,
+ rpcService,
+ highAvailabilityServices,
+ heartbeatServices,
+ metricRegistry,
+ blobCacheService,
+ localCommunicationOnly,
+ externalResourceInfoProvider,
+ fatalErrorHandler);
+
+ return TaskExecutorToServiceAdapter.createFor(taskExecutor);
+ }
+
public static TaskExecutor startTaskManager(
Configuration configuration,
ResourceID resourceID,
@@ -489,4 +531,27 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
: rpcAddress + ":" + rpcPort + "-" + new AbstractID().toString().substring(0, 6)),
config.getString(TaskManagerOptionsInternal.TASK_MANAGER_RESOURCE_ID_METADATA, ""));
}
+
+ /**
+ * Factory for {@link TaskExecutor}.
+ */
+ public interface TaskExecutorServiceFactory {
+ TaskExecutorService createTaskExecutor(
+ Configuration configuration,
+ ResourceID resourceID,
+ RpcService rpcService,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ MetricRegistry metricRegistry,
+ BlobCacheService blobCacheService,
+ boolean localCommunicationOnly,
+ ExternalResourceInfoProvider externalResourceInfoProvider,
+ FatalErrorHandler fatalErrorHandler) throws Exception;
+ }
+
+ public interface TaskExecutorService extends AutoCloseableAsync {
+ void start();
+
+ CompletableFuture<Void> getTerminationFuture();
+ }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerTest.java
index ca8f91d..667b88c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerTest.java
@@ -26,6 +26,7 @@ import org.apache.flink.core.plugin.PluginManager;
import org.apache.flink.core.plugin.PluginUtils;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.testutils.SystemExitTrackingSecurityManager;
+import org.apache.flink.util.FlinkException;
import org.apache.flink.util.TestLogger;
import org.apache.flink.util.TimeUtils;
@@ -35,8 +36,13 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.Timeout;
+import javax.annotation.Nonnull;
+
import java.net.InetAddress;
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import static org.apache.flink.core.testutils.FlinkMatchers.willNotComplete;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
@@ -144,6 +150,56 @@ public class TaskManagerRunnerTest extends TestLogger {
assertThat(taskManagerResourceID.getResourceIdString(), containsString(InetAddress.getLocalHost().getHostName()));
}
+ @Test
+ public void testUnexpectedTaskManagerTerminationFailsRunnerFatally() throws Exception {
+ final CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
+ final TestingTaskExecutorService taskExecutorService = TestingTaskExecutorService.newBuilder()
+ .setTerminationFuture(terminationFuture)
+ .build();
+ final TaskManagerRunner taskManagerRunner = createTaskManagerRunner(
+ createConfiguration(),
+ createTaskExecutorServiceFactory(taskExecutorService));
+
+ terminationFuture.completeExceptionally(new FlinkException("Test exception."));
+
+ Integer statusCode = systemExitTrackingSecurityManager.getSystemExitFuture().get();
+ assertThat(statusCode, is(equalTo(TaskManagerRunner.RUNTIME_FAILURE_RETURN_CODE)));
+ }
+
+ @Test
+ public void testUnexpectedTaskManagerTerminationAfterRunnerCloseWillBeIgnored() throws Exception {
+ final CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
+ final TestingTaskExecutorService taskExecutorService = TestingTaskExecutorService.newBuilder()
+ .setTerminationFuture(terminationFuture)
+ .withManualTerminationFutureCompletion()
+ .build();
+ final TaskManagerRunner taskManagerRunner = createTaskManagerRunner(
+ createConfiguration(),
+ createTaskExecutorServiceFactory(taskExecutorService));
+
+ taskManagerRunner.closeAsync();
+
+ terminationFuture.completeExceptionally(new FlinkException("Test exception."));
+
+ assertThat(systemExitTrackingSecurityManager.getSystemExitFuture(), willNotComplete(Duration.ofMillis(10L)));
+ }
+
+ @Nonnull
+ private TaskManagerRunner.TaskExecutorServiceFactory createTaskExecutorServiceFactory(
+ TestingTaskExecutorService taskExecutorService) {
+ return (
+ configuration,
+ resourceID,
+ rpcService,
+ highAvailabilityServices,
+ heartbeatServices,
+ metricRegistry,
+ blobCacheService,
+ localCommunicationOnly,
+ externalResourceInfoProvider,
+ fatalErrorHandler) -> taskExecutorService;
+ }
+
private static Configuration createConfiguration() {
final Configuration configuration = new Configuration();
configuration.setString(JobManagerOptions.ADDRESS, "localhost");
@@ -152,8 +208,12 @@ public class TaskManagerRunnerTest extends TestLogger {
}
private static TaskManagerRunner createTaskManagerRunner(final Configuration configuration) throws Exception {
+ return createTaskManagerRunner(configuration, TaskManagerRunner::createTaskExecutorService);
+ }
+
+ private static TaskManagerRunner createTaskManagerRunner(final Configuration configuration, TaskManagerRunner.TaskExecutorServiceFactory taskExecutorServiceFactory) throws Exception {
final PluginManager pluginManager = PluginUtils.createPluginManagerFromRootFolder(configuration);
- TaskManagerRunner taskManagerRunner = new TaskManagerRunner(configuration, pluginManager);
+ TaskManagerRunner taskManagerRunner = new TaskManagerRunner(configuration, pluginManager, taskExecutorServiceFactory);
taskManagerRunner.start();
return taskManagerRunner;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorService.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorService.java
new file mode 100644
index 0000000..b9bd56c
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorService.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.taskexecutor;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Testing implementation of {@link TaskManagerRunner.TaskExecutorService}.
+ */
+public class TestingTaskExecutorService implements TaskManagerRunner.TaskExecutorService {
+ private final Runnable startRunnable;
+ private final CompletableFuture<Void> terminationFuture;
+ private final boolean completeTerminationFutureOnClose;
+
+ private TestingTaskExecutorService(
+ Runnable startRunnable,
+ CompletableFuture<Void> terminationFuture,
+ boolean completeTerminationFutureOnClose) {
+ this.startRunnable = startRunnable;
+ this.terminationFuture = terminationFuture;
+ this.completeTerminationFutureOnClose = completeTerminationFutureOnClose;
+ }
+
+ @Override
+ public void start() {
+ startRunnable.run();
+ }
+
+ @Override
+ public CompletableFuture<Void> getTerminationFuture() {
+ return terminationFuture;
+ }
+
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ if (completeTerminationFutureOnClose) {
+ terminationFuture.complete(null);
+ }
+ return terminationFuture;
+ }
+
+ public static Builder newBuilder() {
+ return new Builder();
+ }
+
+ /**
+ * Builder for {@link TestingTaskExecutorService}.
+ */
+ public static final class Builder {
+ private Runnable startRunnable = () -> {};
+ private CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
+ private boolean completeTerminationFutureOnClose = true;
+
+ public Builder setStartRunnable(Runnable startRunnable) {
+ this.startRunnable = startRunnable;
+ return this;
+ }
+
+ public Builder setTerminationFuture(CompletableFuture<Void> terminationFuture) {
+ this.terminationFuture = terminationFuture;
+ return this;
+ }
+
+ public Builder withManualTerminationFutureCompletion() {
+ completeTerminationFutureOnClose = false;
+ return this;
+ }
+
+ TestingTaskExecutorService build() {
+ return new TestingTaskExecutorService(
+ startRunnable,
+ terminationFuture,
+ completeTerminationFutureOnClose);
+ }
+ }
+}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java
index cdbb7d8..a4bd8fc 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java
@@ -273,7 +273,7 @@ public class JobManagerHAProcessFailureRecoveryITCase extends TestLogger {
final PluginManager pluginManager = PluginUtils.createPluginManagerFromRootFolder(config);
// Start the task manager process
for (int i = 0; i < numberOfTaskManagers; i++) {
- taskManagerRunners[i] = new TaskManagerRunner(config, pluginManager);
+ taskManagerRunners[i] = new TaskManagerRunner(config, pluginManager, TaskManagerRunner::createTaskExecutorService);
taskManagerRunners[i].start();
}
[flink] 08/09: [FLINK-19022] Fail
DispatcherResourceManagerComponent fatally if ResourceManager terminates
unexpectedly
Posted by tr...@apache.org.
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 b813c99afa0fb1130af0a48c4879c9c3f0d6dfb4
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Tue Oct 20 10:45:32 2020 +0200
[FLINK-19022] Fail DispatcherResourceManagerComponent fatally if ResourceManager terminates unexpectedly
An unexpected termination of the ResourceManager indicates an invalid state. Hence, the
DispatcherResourceManagerComponent should react to it by calling the FatalErrorHandler.
---
...tDispatcherResourceManagerComponentFactory.java | 3 +-
.../component/DefaultResourceManagerService.java | 5 ++
.../DispatcherResourceManagerComponent.java | 20 ++++-
.../dispatcher/runner/TestingDispatcherRunner.java | 78 ++++++++++++++++++
.../DispatcherResourceManagerComponentTest.java | 95 ++++++++++++++++++++++
.../component/TestingResourceManagerService.java | 93 +++++++++++++++++++++
6 files changed, 292 insertions(+), 2 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultDispatcherResourceManagerComponentFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultDispatcherResourceManagerComponentFactory.java
index 51bef98..d8a00b6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultDispatcherResourceManagerComponentFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultDispatcherResourceManagerComponentFactory.java
@@ -212,7 +212,8 @@ public class DefaultDispatcherResourceManagerComponentFactory implements Dispatc
DefaultResourceManagerService.createFor(resourceManager),
dispatcherLeaderRetrievalService,
resourceManagerRetrievalService,
- webMonitorEndpoint);
+ webMonitorEndpoint,
+ fatalErrorHandler);
} catch (Exception exception) {
// clean up all started components
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultResourceManagerService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultResourceManagerService.java
index e0e2246..2e6f99f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultResourceManagerService.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultResourceManagerService.java
@@ -41,6 +41,11 @@ public class DefaultResourceManagerService implements DispatcherResourceManagerC
}
@Override
+ public CompletableFuture<Void> getTerminationFuture() {
+ return resourceManager.getTerminationFuture();
+ }
+
+ @Override
public CompletableFuture<Void> closeAsync() {
return resourceManager.closeAsync();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
index ffd3139..5b68995 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
@@ -25,9 +25,11 @@ import org.apache.flink.runtime.dispatcher.runner.DispatcherRunner;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.resourcemanager.ResourceManager;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.webmonitor.WebMonitorEndpoint;
import org.apache.flink.util.AutoCloseableAsync;
import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.FlinkException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -69,21 +71,35 @@ public class DispatcherResourceManagerComponent implements AutoCloseableAsync {
private final AtomicBoolean isRunning = new AtomicBoolean(true);
+ private final FatalErrorHandler fatalErrorHandler;
+
DispatcherResourceManagerComponent(
@Nonnull DispatcherRunner dispatcherRunner,
@Nonnull ResourceManagerService resourceManagerService,
@Nonnull LeaderRetrievalService dispatcherLeaderRetrievalService,
@Nonnull LeaderRetrievalService resourceManagerRetrievalService,
- @Nonnull AutoCloseableAsync webMonitorEndpoint) {
+ @Nonnull AutoCloseableAsync webMonitorEndpoint,
+ @Nonnull FatalErrorHandler fatalErrorHandler) {
this.dispatcherRunner = dispatcherRunner;
this.resourceManagerService = resourceManagerService;
this.dispatcherLeaderRetrievalService = dispatcherLeaderRetrievalService;
this.resourceManagerRetrievalService = resourceManagerRetrievalService;
this.webMonitorEndpoint = webMonitorEndpoint;
+ this.fatalErrorHandler = fatalErrorHandler;
this.terminationFuture = new CompletableFuture<>();
this.shutDownFuture = new CompletableFuture<>();
registerShutDownFuture();
+ handleUnexpectedResourceManagerTermination();
+ }
+
+ private void handleUnexpectedResourceManagerTermination() {
+ resourceManagerService.getTerminationFuture().whenComplete(
+ (ignored, throwable) -> {
+ if (isRunning.get()) {
+ fatalErrorHandler.onFatalError(new FlinkException("Unexpected termination of ResourceManager.", throwable));
+ }
+ });
}
private void registerShutDownFuture() {
@@ -175,5 +191,7 @@ public class DispatcherResourceManagerComponent implements AutoCloseableAsync {
interface ResourceManagerService extends AutoCloseableAsync {
ResourceManagerGateway getGateway();
+
+ CompletableFuture<Void> getTerminationFuture();
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/TestingDispatcherRunner.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/TestingDispatcherRunner.java
new file mode 100644
index 0000000..aa1d73c
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/TestingDispatcherRunner.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.dispatcher.runner;
+
+import org.apache.flink.runtime.clusterframework.ApplicationStatus;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
+
+/**
+ * Testing implementation of {@link DispatcherRunner}.
+ */
+public class TestingDispatcherRunner implements DispatcherRunner {
+ private final CompletableFuture<ApplicationStatus> shutDownFuture;
+ private final Supplier<CompletableFuture<Void>> closeAsyncSupplier;
+
+ private TestingDispatcherRunner(
+ CompletableFuture<ApplicationStatus> shutDownFuture,
+ Supplier<CompletableFuture<Void>> closeAsyncSupplier) {
+ this.shutDownFuture = shutDownFuture;
+ this.closeAsyncSupplier = closeAsyncSupplier;
+ }
+
+ @Override
+ public CompletableFuture<ApplicationStatus> getShutDownFuture() {
+ return shutDownFuture;
+ }
+
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ return closeAsyncSupplier.get();
+ }
+
+ public static Builder newBuilder() {
+ return new Builder();
+ }
+
+ /**
+ * Builder for {@link TestingDispatcherRunner}.
+ */
+ public static final class Builder {
+ private CompletableFuture<ApplicationStatus> shutDownFuture = new CompletableFuture<>();
+ private Supplier<CompletableFuture<Void>> closeAsyncSupplier = FutureUtils::completedVoidFuture;
+
+ public Builder setCloseAsyncSupplier(Supplier<CompletableFuture<Void>> closeAsyncSupplier) {
+ this.closeAsyncSupplier = closeAsyncSupplier;
+ return this;
+ }
+
+ public Builder setShutDownFuture(CompletableFuture<ApplicationStatus> shutDownFuture) {
+ this.shutDownFuture = shutDownFuture;
+ return this;
+ }
+
+ public TestingDispatcherRunner build() {
+ return new TestingDispatcherRunner(
+ shutDownFuture,
+ closeAsyncSupplier);
+ }
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponentTest.java
new file mode 100644
index 0000000..4a094b0
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponentTest.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.entrypoint.component;
+
+import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.runtime.dispatcher.runner.TestingDispatcherRunner;
+import org.apache.flink.runtime.leaderretrieval.SettableLeaderRetrievalService;
+import org.apache.flink.runtime.util.TestingFatalErrorHandler;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.flink.core.testutils.FlinkMatchers.containsCause;
+import static org.apache.flink.core.testutils.FlinkMatchers.willNotComplete;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for the {@link DispatcherResourceManagerComponent}.
+ */
+public class DispatcherResourceManagerComponentTest extends TestLogger {
+
+ @Test
+ public void unexpectedResourceManagerTermination_failsFatally() {
+ final CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
+ final TestingFatalErrorHandler fatalErrorHandler = new TestingFatalErrorHandler();
+ final TestingResourceManagerService resourceManagerService = TestingResourceManagerService
+ .newBuilder()
+ .setTerminationFuture(terminationFuture)
+ .build();
+
+ createDispatcherResourceManagerComponent(fatalErrorHandler, resourceManagerService);
+
+ final FlinkException expectedException = new FlinkException("Expected test exception.");
+
+ terminationFuture.completeExceptionally(expectedException);
+
+ final Throwable error = fatalErrorHandler.getException();
+ assertThat(error, containsCause(expectedException));
+ }
+
+ private DispatcherResourceManagerComponent createDispatcherResourceManagerComponent(
+ TestingFatalErrorHandler fatalErrorHandler,
+ TestingResourceManagerService resourceManagerService) {
+ return new DispatcherResourceManagerComponent(
+ TestingDispatcherRunner.newBuilder().build(),
+ resourceManagerService,
+ new SettableLeaderRetrievalService(),
+ new SettableLeaderRetrievalService(),
+ FutureUtils::completedVoidFuture,
+ fatalErrorHandler);
+ }
+
+ @Test
+ public void unexpectedResourceManagerTermination_ifNotRunning_doesNotFailFatally() {
+ final TestingFatalErrorHandler fatalErrorHandler = new TestingFatalErrorHandler();
+ final CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
+ final TestingResourceManagerService resourceManagerService = TestingResourceManagerService
+ .newBuilder()
+ .setTerminationFuture(terminationFuture)
+ .withManualTerminationFutureCompletion()
+ .build();
+
+ final DispatcherResourceManagerComponent dispatcherResourceManagerComponent = createDispatcherResourceManagerComponent(
+ fatalErrorHandler,
+ resourceManagerService);
+
+ dispatcherResourceManagerComponent.closeAsync();
+
+ final FlinkException expectedException = new FlinkException("Expected test exception.");
+ terminationFuture.completeExceptionally(expectedException);
+
+ final CompletableFuture<Throwable> errorFuture = fatalErrorHandler.getErrorFuture();
+ assertThat(errorFuture, willNotComplete(Duration.ofMillis(10L)));
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/component/TestingResourceManagerService.java b/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/component/TestingResourceManagerService.java
new file mode 100644
index 0000000..ed133a4
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/component/TestingResourceManagerService.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.entrypoint.component;
+
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import org.apache.flink.runtime.resourcemanager.utils.TestingResourceManagerGateway;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Testing implementation of {@link DispatcherResourceManagerComponent.ResourceManagerService}.
+ */
+public class TestingResourceManagerService implements DispatcherResourceManagerComponent.ResourceManagerService {
+ private final ResourceManagerGateway resourceManagerGateway;
+
+ private final CompletableFuture<Void> terminationFuture;
+ private final boolean completeTerminationFutureOnClose;
+
+ private TestingResourceManagerService(
+ ResourceManagerGateway resourceManagerGateway,
+ CompletableFuture<Void> terminationFuture,
+ boolean completeTerminationFutureOnClose) {
+ this.resourceManagerGateway = resourceManagerGateway;
+ this.terminationFuture = terminationFuture;
+ this.completeTerminationFutureOnClose = completeTerminationFutureOnClose;
+ }
+
+ @Override
+ public ResourceManagerGateway getGateway() {
+ return resourceManagerGateway;
+ }
+
+ @Override
+ public CompletableFuture<Void> getTerminationFuture() {
+ return terminationFuture;
+ }
+
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ if (completeTerminationFutureOnClose) {
+ terminationFuture.complete(null);
+ }
+ return getTerminationFuture();
+ }
+
+ public static Builder newBuilder() {
+ return new Builder();
+ }
+
+ /**
+ * Builder for {@link TestingResourceManagerService}.
+ */
+ public static final class Builder {
+ private ResourceManagerGateway resourceManagerGateway = new TestingResourceManagerGateway();
+ private CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
+ private boolean completeTerminationFutureOnClose = true;
+
+ public Builder setResourceManagerGateway(ResourceManagerGateway resourceManagerGateway) {
+ this.resourceManagerGateway = resourceManagerGateway;
+ return this;
+ }
+
+ public Builder setTerminationFuture(CompletableFuture<Void> terminationFuture) {
+ this.terminationFuture = terminationFuture;
+ return this;
+ }
+
+ public Builder withManualTerminationFutureCompletion() {
+ completeTerminationFutureOnClose = false;
+ return this;
+ }
+
+ public TestingResourceManagerService build() {
+ return new TestingResourceManagerService(resourceManagerGateway, terminationFuture, completeTerminationFutureOnClose);
+ }
+ }
+}
[flink] 07/09: [FLINK-19022] Fail fatally if
DispatcherGatewayService terminates unexpectedly in
AbstractDispatcherLeaderProcess
Posted by tr...@apache.org.
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 11a4232a9218841c2b01bc3a53cadfd318c59813
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Tue Oct 20 11:36:02 2020 +0200
[FLINK-19022] Fail fatally if DispatcherGatewayService terminates unexpectedly in AbstractDispatcherLeaderProcess
This commit ensures that we fail fatally if the DispatcherGatewayService termiantes unexpectedly in the
AbstractDispatcherLeaderProcess. This ensures that the system won't stay in an invalid state.
---
.../runner/AbstractDispatcherLeaderProcess.java | 21 ++++++++++-
.../runner/DefaultDispatcherGatewayService.java | 5 +++
.../runner/SessionDispatcherLeaderProcessTest.java | 43 ++++++++++++++++++++++
3 files changed, 67 insertions(+), 2 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 0ecdb8b..d98fce4 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
@@ -30,6 +30,7 @@ import org.apache.flink.runtime.jobmanager.JobGraphWriter;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.webmonitor.RestfulGateway;
import org.apache.flink.util.AutoCloseableAsync;
+import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
@@ -177,6 +178,16 @@ public abstract class AbstractDispatcherLeaderProcess implements DispatcherLeade
dispatcherService = createdDispatcherService;
dispatcherGatewayFuture.complete(createdDispatcherService.getGateway());
FutureUtils.forward(createdDispatcherService.getShutDownFuture(), shutDownFuture);
+ handleUnexpectedDispatcherServiceTermination(createdDispatcherService);
+ }
+
+ private void handleUnexpectedDispatcherServiceTermination(DispatcherGatewayService createdDispatcherService) {
+ createdDispatcherService.getTerminationFuture().whenComplete(
+ (ignored, throwable) -> {
+ runIfStateIs(State.RUNNING, () -> {
+ handleError(new FlinkException("Unexpected termination of DispatcherService.", throwable));
+ });
+ });
}
final <V> Optional<V> supplyUnsynchronizedIfRunning(Supplier<V> supplier) {
@@ -225,13 +236,17 @@ public abstract class AbstractDispatcherLeaderProcess implements DispatcherLeade
}
if (throwable != null) {
- closeAsync();
- fatalErrorHandler.onFatalError(throwable);
+ handleError(throwable);
}
return null;
}
+ private void handleError(Throwable throwable) {
+ closeAsync();
+ fatalErrorHandler.onFatalError(throwable);
+ }
+
/**
* The state of the {@link DispatcherLeaderProcess}.
*/
@@ -264,5 +279,7 @@ public abstract class AbstractDispatcherLeaderProcess implements DispatcherLeade
CompletableFuture<Void> onRemovedJobGraph(JobID jobId);
CompletableFuture<ApplicationStatus> getShutDownFuture();
+
+ CompletableFuture<Void> getTerminationFuture();
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherGatewayService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherGatewayService.java
index cc668cc..908e1b3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherGatewayService.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherGatewayService.java
@@ -56,6 +56,11 @@ public class DefaultDispatcherGatewayService implements AbstractDispatcherLeader
}
@Override
+ public CompletableFuture<Void> getTerminationFuture() {
+ return dispatcher.getTerminationFuture();
+ }
+
+ @Override
public CompletableFuture<Void> closeAsync() {
return dispatcher.closeAsync();
}
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 799b91b..7c0462a 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
@@ -42,6 +42,7 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
+import java.time.Duration;
import java.util.Collection;
import java.util.Collections;
import java.util.UUID;
@@ -51,6 +52,8 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import static org.apache.flink.core.testutils.FlinkMatchers.containsCause;
+import static org.apache.flink.core.testutils.FlinkMatchers.willNotComplete;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsInAnyOrder;
@@ -174,6 +177,46 @@ public class SessionDispatcherLeaderProcessTest extends TestLogger {
}
@Test
+ public void unexpectedDispatcherServiceTerminationWhileRunning_callsFatalErrorHandler() {
+ final CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
+ dispatcherServiceFactory = TestingDispatcherServiceFactory.newBuilder()
+ .setCreateFunction((ignoredA, ignoredB, ignoredC) -> TestingDispatcherGatewayService.newBuilder()
+ .setTerminationFuture(terminationFuture)
+ .build())
+ .build();
+ final SessionDispatcherLeaderProcess dispatcherLeaderProcess = createDispatcherLeaderProcess();
+ dispatcherLeaderProcess.start();
+
+ final FlinkException expectedFailure = new FlinkException("Expected test failure.");
+ terminationFuture.completeExceptionally(expectedFailure);
+
+ final Throwable error = fatalErrorHandler.getErrorFuture().join();
+ assertThat(error, containsCause(expectedFailure));
+
+ fatalErrorHandler.clearError();
+ }
+
+ @Test
+ public void unexpectedDispatcherServiceTerminationWhileNotRunning_doesNotCallFatalErrorHandler() {
+ final CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
+ dispatcherServiceFactory = TestingDispatcherServiceFactory.newBuilder()
+ .setCreateFunction((ignoredA, ignoredB, ignoredC) -> TestingDispatcherGatewayService.newBuilder()
+ .setTerminationFuture(terminationFuture)
+ .withManualTerminationFutureCompletion()
+ .build())
+ .build();
+ final SessionDispatcherLeaderProcess dispatcherLeaderProcess = createDispatcherLeaderProcess();
+ dispatcherLeaderProcess.start();
+
+ dispatcherLeaderProcess.closeAsync();
+
+ final FlinkException expectedFailure = new FlinkException("Expected test failure.");
+ terminationFuture.completeExceptionally(expectedFailure);
+
+ assertThat(fatalErrorHandler.getErrorFuture(), willNotComplete(Duration.ofMillis(10)));
+ }
+
+ @Test
public void confirmLeaderSessionFuture_completesAfterDispatcherServiceHasBeenStarted() throws Exception {
final OneShotLatch createDispatcherServiceLatch = new OneShotLatch();
final String dispatcherAddress = "myAddress";
[flink] 06/09: [FLINK-19022] Let TaskExecutor, Dispatcher,
RM fail fatally if onStart throws a Throwable
Posted by tr...@apache.org.
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 9e31510df66b508e28ab19ff8cf024c02ec423b1
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Mon Oct 19 15:33:27 2020 +0200
[FLINK-19022] Let TaskExecutor, Dispatcher, RM fail fatally if onStart throws a Throwable
If TaskExecutor., Dispatcher. or ResourceManager.onStart throws a Throwable, the system is
no longer in a valid state. Hence, this commit ensures that the components will properly
call FatalErrorHandler.onFatalError in such a case.
---
.../src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java | 4 ++--
.../org/apache/flink/runtime/resourcemanager/ResourceManager.java | 4 ++--
.../main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java | 4 ++--
3 files changed, 6 insertions(+), 6 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
index 67a827f..0d400bb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
@@ -202,8 +202,8 @@ public abstract class Dispatcher extends PermanentlyFencedRpcEndpoint<Dispatcher
public void onStart() throws Exception {
try {
startDispatcherServices();
- } catch (Exception e) {
- final DispatcherException exception = new DispatcherException(String.format("Could not start the Dispatcher %s", getAddress()), e);
+ } catch (Throwable t) {
+ final DispatcherException exception = new DispatcherException(String.format("Could not start the Dispatcher %s", getAddress()), t);
onFatalError(exception);
throw exception;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
index 915e141..df51454 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
@@ -216,8 +216,8 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
public final void onStart() throws Exception {
try {
startResourceManagerServices();
- } catch (Exception e) {
- final ResourceManagerException exception = new ResourceManagerException(String.format("Could not start the ResourceManager %s", getAddress()), e);
+ } catch (Throwable t) {
+ final ResourceManagerException exception = new ResourceManagerException(String.format("Could not start the ResourceManager %s", getAddress()), t);
onFatalError(exception);
throw exception;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
index e2be76e..b93f9fa 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
@@ -360,8 +360,8 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
public void onStart() throws Exception {
try {
startTaskExecutorServices();
- } catch (Exception e) {
- final TaskManagerException exception = new TaskManagerException(String.format("Could not start the TaskExecutor %s", getAddress()), e);
+ } catch (Throwable t) {
+ final TaskManagerException exception = new TaskManagerException(String.format("Could not start the TaskExecutor %s", getAddress()), t);
onFatalError(exception);
throw exception;
}
[flink] 01/09: [hotfix] Add FlinkMatchers.willNotComplete to assert
that a future times out
Posted by tr...@apache.org.
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 75d15bd66f43e5a6b37da03b14fffc301b9942a5
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Tue Oct 20 11:06:43 2020 +0200
[hotfix] Add FlinkMatchers.willNotComplete to assert that a future times out
---
.../apache/flink/core/testutils/FlinkMatchers.java | 44 ++++++++++++++++++++++
1 file changed, 44 insertions(+)
diff --git a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkMatchers.java b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkMatchers.java
index 5867ef6..a1dc6c6 100644
--- a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkMatchers.java
+++ b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkMatchers.java
@@ -89,6 +89,13 @@ public class FlinkMatchers {
return new ContainsCauseMatcher(failureCause);
}
+ /**
+ * Checks that a {@link CompletableFuture} won't complete within the given timeout.
+ */
+ public static Matcher<CompletableFuture<?>> willNotComplete(Duration timeout) {
+ return new WillNotCompleteMatcher(timeout);
+ }
+
// ------------------------------------------------------------------------
/** This class should not be instantiated. */
@@ -267,4 +274,41 @@ public class FlinkMatchers {
return Optional.empty();
}
}
+
+ private static final class WillNotCompleteMatcher extends TypeSafeDiagnosingMatcher<CompletableFuture<?>> {
+
+ private final Duration timeout;
+
+ private WillNotCompleteMatcher(Duration timeout) {
+ this.timeout = timeout;
+ }
+
+ @Override
+ protected boolean matchesSafely(
+ CompletableFuture<?> item,
+ Description mismatchDescription) {
+
+ try {
+ final Object value = item.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
+ mismatchDescription.appendText("The given future completed with ")
+ .appendValue(value);
+ } catch (TimeoutException timeoutException) {
+ return true;
+ } catch (InterruptedException e) {
+ mismatchDescription.appendText("The waiting thread was interrupted.");
+ } catch (ExecutionException e) {
+ mismatchDescription.appendText("The given future was completed exceptionally: ")
+ .appendValue(e);
+ }
+
+ return false;
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText("The given future should not complete within ")
+ .appendValue(timeout.toMillis())
+ .appendText(" ms.");
+ }
+ }
}
[flink] 05/09: [FLINK-19022][rpc] Log if RpcEndpoint failed its
termination
Posted by tr...@apache.org.
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 806b9f332efe86926d26861e64b528e4c6ce1cd6
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Mon Oct 19 15:29:20 2020 +0200
[FLINK-19022][rpc] Log if RpcEndpoint failed its termination
In order to not swallow the failure reason of failed RpcEndpoints, we are now
logging the cause of a RpcEndpoint failure.
---
.../src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java | 2 ++
1 file changed, 2 insertions(+)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java
index 7ad3776..34d4690 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java
@@ -127,8 +127,10 @@ class AkkaRpcActor<T extends RpcEndpoint & RpcGateway> extends AbstractActor {
super.postStop();
if (rpcEndpointTerminationResult.isSuccess()) {
+ log.debug("The RpcEndpoint {} terminated successfully.", rpcEndpoint.getEndpointId());
terminationFuture.complete(null);
} else {
+ log.info("The RpcEndpoint {} failed.", rpcEndpoint.getEndpointId(), rpcEndpointTerminationResult.getFailureCause());
terminationFuture.completeExceptionally(rpcEndpointTerminationResult.getFailureCause());
}