Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6352#discussion_r203293425
--- Diff:
flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java ---
@@ -186,39 +189,56 @@ public static void
populateYarnSecureConfigurations(Configuration conf, String p
conf.set("hadoop.security.auth_to_local", "RULE:[1:$1]
RULE:[2:$1]");
}
- /**
- * Sleep a bit between the tests (we are re-using the YARN cluster for
the tests).
- */
- @After
- public void sleep() {
- try {
- Thread.sleep(500);
- } catch (InterruptedException e) {
- Assert.fail("Should not happen");
- }
- }
-
@Before
- public void checkClusterEmpty() throws IOException, YarnException {
+ public void checkClusterEmpty() {
if (yarnClient == null) {
yarnClient = YarnClient.createYarnClient();
yarnClient.init(getYarnConfiguration());
yarnClient.start();
}
- List<ApplicationReport> apps = yarnClient.getApplications();
- for (ApplicationReport app : apps) {
- if (app.getYarnApplicationState() !=
YarnApplicationState.FINISHED
- && app.getYarnApplicationState() !=
YarnApplicationState.KILLED
- && app.getYarnApplicationState() !=
YarnApplicationState.FAILED) {
- Assert.fail("There is at least one application
on the cluster is not finished." +
- "App " + app.getApplicationId()
+ " is in state " + app.getYarnApplicationState());
+ flinkConfiguration = new
org.apache.flink.configuration.Configuration(globalConfiguration);
+
+ isNewMode = Objects.equals(TestBaseUtils.CodebaseType.NEW,
TestBaseUtils.getCodebaseType());
+ }
+
+ /**
+ * Sleep a bit between the tests (we are re-using the YARN cluster for
the tests).
+ */
+ @After
+ public void sleep() throws IOException, YarnException {
+ Deadline deadline = Deadline.now().plus(Duration.ofSeconds(10));
+
+ boolean isAnyJobRunning = yarnClient.getApplications().stream()
+ .anyMatch(YarnTestBase::isApplicationRunning);
+
+ while (deadline.hasTimeLeft() && isAnyJobRunning) {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ Assert.fail("Should not happen");
}
+ isAnyJobRunning = yarnClient.getApplications().stream()
+ .anyMatch(YarnTestBase::isApplicationRunning);
}
- flinkConfiguration = new
org.apache.flink.configuration.Configuration(globalConfiguration);
+ if (isAnyJobRunning) {
+ final Optional<ApplicationReport> runningApp =
yarnClient.getApplications().stream()
+ .filter(YarnTestBase::isApplicationRunning)
+ .findAny();
+ if (runningApp.isPresent()) {
+ final ApplicationReport app = runningApp.get();
+ Assert.fail("There is at least one application
on the cluster that is not finished." +
+ "App " + app.getApplicationId() + " is
in state " + app.getYarnApplicationState());
--- End diff --
Could we log all running applications instead of any?
---
