[FLINK-14246][runtime] Annotate MiniClusterITCase with AlsoRunWithSch…

…edulerNG and fix broken tests

The tests broke because the error messages of NoResourceAvailableExceptions from
LegacyScheduler and DefaultScheduler are different.

flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java

@@ -46,9 +46,11 @@
 import org.apache.flink.runtime.testtasks.BlockingNoOpInvokable;
 import org.apache.flink.runtime.testtasks.NoOpInvokable;
 import org.apache.flink.runtime.testtasks.WaitingNoOpInvokable;
+import org.apache.flink.testutils.junit.category.AlsoRunWithSchedulerNG;
 import org.apache.flink.util.TestLogger;
 
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 
 import java.io.IOException;
 import java.util.concurrent.CompletableFuture;
@@ -62,6 +64,7 @@
 /**
  * Integration test cases for the {@link MiniCluster}.
  */
+@Category(AlsoRunWithSchedulerNG.class)
 public class MiniClusterITCase extends TestLogger {
 
  @Test
@@ -110,7 +113,12 @@ public void testHandleStreamingJobsWhenNotEnoughSlot() throws Exception {
  } catch (JobExecutionException e) {
  assertTrue(findThrowableWithMessage(e, "Job execution failed.").isPresent());
  assertTrue(findThrowable(e, NoResourceAvailableException.class).isPresent());
- assertTrue(findThrowableWithMessage(e, "Slots required: 2, slots allocated: 1").isPresent());
+
+ //TODO: remove the legacy scheduler message check once legacy scheduler is removed
+ final String legacySchedulerErrorMessage = "Slots required: 2, slots allocated: 1";
+ final String ngSchedulerErrorMessage = "Could not allocate the required slot within slot request timeout";
+ assertTrue(findThrowableWithMessage(e, legacySchedulerErrorMessage).isPresent() ||
+ findThrowableWithMessage(e, ngSchedulerErrorMessage).isPresent());
  }
  }
 
@@ -122,7 +130,12 @@ public void testHandleBatchJobsWhenNotEnoughSlot() throws Exception {
  } catch (JobExecutionException e) {
  assertTrue(findThrowableWithMessage(e, "Job execution failed.").isPresent());
  assertTrue(findThrowable(e, NoResourceAvailableException.class).isPresent());
- assertTrue(findThrowableWithMessage(e, "Could not allocate enough slots").isPresent());
+
+ //TODO: remove the legacy scheduler message check once legacy scheduler is removed
+ final String legacySchedulerErrorMessage = "Could not allocate enough slots";
+ final String ngSchedulerErrorMessage = "Could not allocate the required slot within slot request timeout";
+ assertTrue(findThrowableWithMessage(e, legacySchedulerErrorMessage).isPresent() ||
+ findThrowableWithMessage(e, ngSchedulerErrorMessage).isPresent());
  }
  }