forked from apache/flink
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[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. This closes apache#5597.
- Loading branch information
1 parent
035257e
commit 45397fe
Showing
4 changed files
with
235 additions
and
53 deletions.
There are no files selected for viewing
102 changes: 102 additions & 0 deletions
102
...-yarn/src/main/java/org/apache/flink/yarn/RegisterApplicationMasterResponseReflector.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| /* | ||
| * 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 | ||
| * | ||
| * https://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()) { | ||
| return containers; | ||
| } | ||
| } catch (Exception t) { | ||
| logger.error("Error invoking 'getContainersFromPreviousAttempts()'", t); | ||
| } | ||
| } | ||
|
|
||
| return Collections.emptyList(); | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
| Method getMethod() { | ||
| return method; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
117 changes: 117 additions & 0 deletions
117
...n/src/test/java/org/apache/flink/yarn/RegisterApplicationMasterResponseReflectorTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,117 @@ | ||
| /* | ||
| * 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 | ||
| * | ||
| * https://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.util.TestLogger; | ||
|
|
||
| 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 extends TestLogger { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(RegisterApplicationMasterResponseReflectorTest.class); | ||
|
|
||
| @Mock | ||
| private Container mockContainer; | ||
|
|
||
| @Before | ||
| public void setUp() { | ||
| MockitoAnnotations.initMocks(this); | ||
| } | ||
|
|
||
| @Test | ||
| public void testCallsMethodIfPresent() { | ||
| final RegisterApplicationMasterResponseReflector registerApplicationMasterResponseReflector = | ||
| new RegisterApplicationMasterResponseReflector(LOG, HasMethod.class); | ||
|
|
||
| final List<Container> containersFromPreviousAttemptsUnsafe = | ||
| registerApplicationMasterResponseReflector.getContainersFromPreviousAttemptsUnsafe(new | ||
| HasMethod()); | ||
|
|
||
| assertThat(containersFromPreviousAttemptsUnsafe, hasSize(1)); | ||
| } | ||
|
|
||
| @Test | ||
| public void testDoesntCallMethodIfAbsent() { | ||
| final RegisterApplicationMasterResponseReflector registerApplicationMasterResponseReflector = | ||
| new RegisterApplicationMasterResponseReflector(LOG, HasMethod.class); | ||
|
|
||
| final List<Container> containersFromPreviousAttemptsUnsafe = | ||
| registerApplicationMasterResponseReflector.getContainersFromPreviousAttemptsUnsafe(new | ||
| Object()); | ||
|
|
||
| assertThat(containersFromPreviousAttemptsUnsafe, empty()); | ||
| } | ||
|
|
||
| @Test | ||
| public void testGetMethodReflectiveHadoop22() { | ||
| assumeTrue( | ||
| "Method getContainersFromPreviousAttempts is not supported by Hadoop: " + | ||
| VersionInfo.getVersion(), | ||
| isHadoopVersionGreaterThanOrEquals(2, 2)); | ||
|
|
||
| final RegisterApplicationMasterResponseReflector registerApplicationMasterResponseReflector = | ||
| new RegisterApplicationMasterResponseReflector(LOG); | ||
|
|
||
| final Method method = registerApplicationMasterResponseReflector.getMethod(); | ||
| assertThat(method, notNullValue()); | ||
| } | ||
|
|
||
| private static boolean isHadoopVersionGreaterThanOrEquals(final int major, final int minor) { | ||
| final String[] splitVersion = VersionInfo.getVersion().split("\\."); | ||
| final int[] versions = Arrays.stream(splitVersion).mapToInt(Integer::parseInt).toArray(); | ||
| return versions[0] >= major && versions[1] >= minor; | ||
| } | ||
|
|
||
| /** | ||
| * Class which has a method with the same signature as | ||
| * {@link RegisterApplicationMasterResponse#getContainersFromPreviousAttempts()}. | ||
| */ | ||
| private class HasMethod { | ||
|
|
||
| /** | ||
| * Called from {@link #testCallsMethodIfPresent()}. | ||
| */ | ||
| @SuppressWarnings("unused") | ||
| public List<Container> getContainersFromPreviousAttempts() { | ||
| return Collections.singletonList(mockContainer); | ||
| } | ||
| } | ||
| } |