You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GJL <gi...@git.apache.org> on 2018/02/28 12:29:20 UTC
[GitHub] flink pull request #5597: [FLINK-7805][flip6] Add HA capabilities to YarnRes...
GitHub user GJL opened a pull request:
https://github.com/apache/flink/pull/5597
[FLINK-7805][flip6] Add HA capabilities to YarnResourceManager
## What is the purpose of the change
*Recover previously running containers after a restart of the ApplicationMaster.
This is a port of a feature that was already implemented prior to FLIP-6.*
cc: @tillrohrmann
## Brief change log
- *Extract `RegisterApplicationMasterResponseReflector` class into separate file.*
- *Use `RegisterApplicationMasterResponseReflector` from within `YarnResourceManager`*
## Verifying this change
This change added tests and can be verified as follows:
- *Added unit tests for `RegisterApplicationMasterResponseReflector`*
- *Manually deployed a cluster on YARN with HA enabled. Submitted a job, and killed the master several times. Verified that the right log messages were generated ("Recovered X containers from previous attempts"), and that the taskmanager resource ids remained the same.*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (**yes** / no / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/GJL/flink FLINK-7805-2
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/5597.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #5597
----
commit 012f656705466cea4ba2e037f7e52f46c0a1bf9f
Author: gyao <ga...@...>
Date: 2018-02-27T15:58:53Z
[FLINK-8787][flip6] Do not copy flinkConfiguration in AbstractYarnClusterDescriptor
commit bf8211112c16a16bd29aba754c1c34f4de960af4
Author: gyao <ga...@...>
Date: 2018-02-28T12:04:19Z
[hotfix] Add missing space to log message in ZooKeeperLeaderElectionService
commit a435c6503dcd8e65e34b807d5dc1c7045911e788
Author: gyao <ga...@...>
Date: 2018-02-28T12:06:00Z
[hotfix][Javadoc] Fix typo in YARN Utils: teh -> the
commit 3e4c28c38cc04aa29bdb8928fa26daf3c4ab1e69
Author: gyao <ga...@...>
Date: 2018-02-28T12:07:04Z
[hotfix][Javadoc] Fix typo in YarnTestBase: teh -> the
commit 6b1efdb290c605d93f17fc0aadfb07205eaf60fd
Author: gyao <ga...@...>
Date: 2018-02-28T12:08:25Z
[hotfix][tests] Fix wrong assertEquals in YARNSessionCapacitySchedulerITCase
Test swapped actual and expected arguments.
Remove catching Throwable in test; instead propagate all exceptions.
commit 615c19acfc87bad021b3dce02b6dc5fac2aac784
Author: gyao <ga...@...>
Date: 2018-02-28T12:20:23Z
[FLINK-7805][flip6] Recover YARN containers after AM restart.
Recover previously running containers after a restart of the ApplicationMaster.
This is a port of a feature that was already implemented prior to FLIP-6.
Extract RegisterApplicationMasterResponseReflector class into separate file.
----
---
[GitHub] flink pull request #5597: [FLINK-7805][flip6] Add HA capabilities to YarnRes...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/5597
---
[GitHub] flink pull request #5597: [FLINK-7805][flip6] Add HA capabilities to YarnRes...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5597#discussion_r171323504
--- Diff: flink-yarn/src/test/java/org/apache/flink/yarn/RegisterApplicationMasterResponseReflectorTest.java ---
@@ -0,0 +1,115 @@
+/*
+ * 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.yarn;
+
+import org.apache.hadoop.util.VersionInfo;
+import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Method;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assume.assumeTrue;
+
+/**
+ * Tests for {@link RegisterApplicationMasterResponseReflector}.
+ */
+public class RegisterApplicationMasterResponseReflectorTest {
--- End diff --
`TestLogger` missing.
---
[GitHub] flink pull request #5597: [FLINK-7805][flip6] Add HA capabilities to YarnRes...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5597#discussion_r171323432
--- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/RegisterApplicationMasterResponseReflector.java ---
@@ -0,0 +1,103 @@
+/*
+ * 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.yarn;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.slf4j.Logger;
+
+import java.lang.reflect.Method;
+import java.util.Collections;
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Looks up the method {@link RegisterApplicationMasterResponse#getContainersFromPreviousAttempts()}
+ * once and saves the method. This saves computation time on subsequent calls.
+ */
+class RegisterApplicationMasterResponseReflector {
+
+ private final Logger logger;
+
+ /**
+ * Reflected method {@link RegisterApplicationMasterResponse#getContainersFromPreviousAttempts()}.
+ */
+ private Method method;
+
+ RegisterApplicationMasterResponseReflector(final Logger log) {
+ this(log, RegisterApplicationMasterResponse.class);
+ }
+
+ @VisibleForTesting
+ RegisterApplicationMasterResponseReflector(final Logger log, final Class<?> clazz) {
+ this.logger = requireNonNull(log);
+ requireNonNull(clazz);
+
+ try {
+ method = clazz.getMethod("getContainersFromPreviousAttempts");
+ } catch (NoSuchMethodException e) {
+ // that happens in earlier Hadoop versions (pre 2.2)
+ logger.info("Cannot reconnect to previously allocated containers. " +
+ "This YARN version does not support 'getContainersFromPreviousAttempts()'");
+ }
+ }
+
+ /**
+ * Checks if a YARN application still has registered containers. If the application master
+ * registered at the ResourceManager for the first time, this list will be empty. If the
+ * application master registered a repeated time (after a failure and recovery), this list
+ * will contain the containers that were previously allocated.
+ *
+ * @param response The response object from the registration at the ResourceManager.
+ * @return A list with containers from previous application attempt.
+ */
+ List<Container> getContainersFromPreviousAttempts(final RegisterApplicationMasterResponse response) {
+ return getContainersFromPreviousAttemptsUnsafe(response);
+ }
+
+ /**
+ * Same as {@link #getContainersFromPreviousAttempts(RegisterApplicationMasterResponse)} but
+ * allows to pass objects that are not of type {@link RegisterApplicationMasterResponse}.
+ */
+ @VisibleForTesting
+ List<Container> getContainersFromPreviousAttemptsUnsafe(final Object response) {
+ if (method != null && response != null) {
+ try {
+ @SuppressWarnings("unchecked")
+ final List<Container> containers = (List<Container>) method.invoke(response);
+ if (containers != null && !containers.isEmpty()) {
+ logger.info("Recovered {} containers from previous attempts ({}).", containers.size(), containers);
--- End diff --
Let's log this in the `YarnResourceManager`.
---