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 2021/03/19 12:30:37 UTC
[flink] 01/02: [FLINK-21654][tests] Adds retry loop to
YarnClient.getApplications
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9e29a221440169bdd264c6114091580ad09592f8
Author: Matthias Pohl <ma...@ververica.com>
AuthorDate: Wed Mar 10 08:50:44 2021 +0100
[FLINK-21654][tests] Adds retry loop to YarnClient.getApplications
This fix should cover the issue with YARN-7007. The corresponding discussion
is happening in FLINK-15534.
This closes #15134.
---
.../flink/yarn/YARNHighAvailabilityITCase.java | 4 +-
.../yarn/YARNSessionCapacitySchedulerITCase.java | 10 +++--
.../apache/flink/yarn/YARNSessionFIFOITCase.java | 11 +++--
.../java/org/apache/flink/yarn/YarnTestBase.java | 48 ++++++++++++++++++++--
4 files changed, 61 insertions(+), 12 deletions(-)
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
index a0ba2d4..36a32e3 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
@@ -306,8 +306,8 @@ public class YARNHighAvailabilityITCase extends YarnTestBase {
CommonTestUtils.waitUntilCondition(
() ->
- !yarnClient
- .getApplications(
+ !getApplicationReportWithRetryOnNPE(
+ yarnClient,
EnumSet.of(
YarnApplicationState.KILLED,
YarnApplicationState.FINISHED))
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
index 02dd411..96a77ba 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
@@ -628,7 +628,8 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
// find out the application id and wait until it has finished.
try {
List<ApplicationReport> apps =
- yc.getApplications(EnumSet.of(YarnApplicationState.RUNNING));
+ getApplicationReportWithRetryOnNPE(
+ yc, EnumSet.of(YarnApplicationState.RUNNING));
ApplicationId tmpAppId;
if (apps.size() == 1) {
@@ -639,12 +640,15 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
LOG.info("waiting for the job with appId {} to finish", tmpAppId);
// wait until the app has finished
- while (yc.getApplications(EnumSet.of(YarnApplicationState.RUNNING)).size() > 0) {
+ while (getApplicationReportWithRetryOnNPE(
+ yc, EnumSet.of(YarnApplicationState.RUNNING))
+ .size()
+ > 0) {
sleep(500);
}
} else {
// get appId by finding the latest finished appid
- apps = yc.getApplications();
+ apps = getApplicationReportWithRetryOnNPE(yc);
Collections.sort(
apps,
new Comparator<ApplicationReport>() {
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
index 3bafc91..603f508 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
@@ -181,7 +181,8 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
yc.init(YARN_CONFIGURATION);
yc.start();
List<ApplicationReport> apps =
- yc.getApplications(EnumSet.of(YarnApplicationState.RUNNING));
+ getApplicationReportWithRetryOnNPE(
+ yc, EnumSet.of(YarnApplicationState.RUNNING));
Assert.assertEquals(1, apps.size()); // Only one running
ApplicationReport app = apps.get(0);
@@ -190,9 +191,13 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
ApplicationId id = app.getApplicationId();
yc.killApplication(id);
- while (yc.getApplications(EnumSet.of(YarnApplicationState.KILLED)).size()
+ while (getApplicationReportWithRetryOnNPE(
+ yc, EnumSet.of(YarnApplicationState.KILLED))
+ .size()
== 0
- && yc.getApplications(EnumSet.of(YarnApplicationState.FINISHED))
+ && getApplicationReportWithRetryOnNPE(
+ yc,
+ EnumSet.of(YarnApplicationState.FINISHED))
.size()
== 0) {
sleep(500);
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
index b9fd51e..8858b6b 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
@@ -24,6 +24,7 @@ import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.runtime.clusterframework.BootstrapTools;
import org.apache.flink.test.util.TestBaseUtils;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.TestLogger;
import org.apache.flink.util.function.RunnableWithException;
@@ -278,7 +279,7 @@ public abstract class YarnTestBase extends TestLogger {
Deadline deadline = Deadline.now().plus(Duration.ofSeconds(10));
boolean isAnyJobRunning =
- yarnClient.getApplications().stream()
+ getApplicationReportWithRetryOnNPE(yarnClient).stream()
.anyMatch(YarnTestBase::isApplicationRunning);
while (deadline.hasTimeLeft() && isAnyJobRunning) {
@@ -288,13 +289,13 @@ public abstract class YarnTestBase extends TestLogger {
Assert.fail("Should not happen");
}
isAnyJobRunning =
- yarnClient.getApplications().stream()
+ getApplicationReportWithRetryOnNPE(yarnClient).stream()
.anyMatch(YarnTestBase::isApplicationRunning);
}
if (isAnyJobRunning) {
final List<String> runningApps =
- yarnClient.getApplications().stream()
+ getApplicationReportWithRetryOnNPE(yarnClient).stream()
.filter(YarnTestBase::isApplicationRunning)
.map(
app ->
@@ -313,6 +314,44 @@ public abstract class YarnTestBase extends TestLogger {
}
}
+ static List<ApplicationReport> getApplicationReportWithRetryOnNPE(final YarnClient yarnClient)
+ throws IOException, YarnException {
+ return getApplicationReportWithRetryOnNPE(yarnClient, null);
+ }
+
+ static List<ApplicationReport> getApplicationReportWithRetryOnNPE(
+ final YarnClient yarnClient, @Nullable EnumSet<YarnApplicationState> states)
+ throws IOException, YarnException {
+ final int maxRetryCount = 10;
+ NullPointerException mostRecentNPE = null;
+ for (int i = 0; i < maxRetryCount; i++) {
+ try {
+ return yarnClient.getApplications(states);
+ } catch (NullPointerException e) {
+ String npeStr = ExceptionUtils.stringifyException(e);
+ if (!npeStr.contains("RMAppAttemptMetrics.getAggregateAppResourceUsage")) {
+ // unrelated NullPointerExceptions should be forwarded to the calling method
+ throw e;
+ }
+
+ mostRecentNPE = e;
+ final String logMessage =
+ "NullPointerException was caught most likely being related to YARN-7007. The related discussion is happening in FLINK-15534. The exception is going to be ignored.";
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(logMessage, mostRecentNPE);
+ } else {
+ LOG.warn(logMessage);
+ }
+ }
+ }
+
+ throw new IllegalStateException(
+ "YarnClient.getApplications command failed "
+ + maxRetryCount
+ + " times to gather the application report. Check FLINK-15534 for further details.",
+ mostRecentNPE);
+ }
+
private static boolean isApplicationRunning(ApplicationReport app) {
final YarnApplicationState yarnApplicationState = app.getYarnApplicationState();
return yarnApplicationState != YarnApplicationState.FINISHED
@@ -676,7 +715,8 @@ public abstract class YarnTestBase extends TestLogger {
checkState(yarnClient != null);
final List<ApplicationReport> apps =
- yarnClient.getApplications(EnumSet.of(YarnApplicationState.RUNNING));
+ getApplicationReportWithRetryOnNPE(
+ yarnClient, EnumSet.of(YarnApplicationState.RUNNING));
assertEquals(1, apps.size()); // Only one running
return apps.get(0);
}