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:13 UTC
[flink] 08/09: [FLINK-19022] Fail
DispatcherResourceManagerComponent fatally if ResourceManager terminates
unexpectedly
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);
+ }
+ }
+}