[FLINK-21518][tests] Fix testMinCheckpointPause instability

The instability is caused by last assertion depending on timing.
Solve by waiting on condition.

Another issue is that failure can cause RejectionExecutionException.
This is caused by shutting down executor (in tests finally block)
without shutting down CheckpointCoordinator.
Solve by shutting down CheckpointCoordinator first.

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java

@@ -276,6 +276,7 @@ public void testMinCheckpointPause() throws Exception {
         // will use a different thread to allow checkpoint triggering before exiting from
         // receiveAcknowledgeMessage
         ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
+        CheckpointCoordinator coordinator = null;
         try {
             int pause = 1000;
             JobVertexID jobVertexId = new JobVertexID();
@@ -291,7 +292,7 @@ public void testMinCheckpointPause() throws Exception {
             ExecutionVertex vertex = graph.getJobVertex(jobVertexId).getTaskVertices()[0];
             ExecutionAttemptID attemptId = vertex.getCurrentExecutionAttempt().getAttemptId();
 
-            CheckpointCoordinator coordinator =
+            coordinator =
                     new CheckpointCoordinatorBuilder()
                             .setTimer(new ScheduledExecutorServiceAdapter(executorService))
                             .setCheckpointCoordinatorConfiguration(
@@ -319,9 +320,14 @@ public void testMinCheckpointPause() throws Exception {
                     TASK_MANAGER_LOCATION_INFO);
             Thread.sleep(pause / 2);
             assertEquals(0, coordinator.getNumberOfPendingCheckpoints());
-            Thread.sleep(pause);
-            assertEquals(1, coordinator.getNumberOfPendingCheckpoints());
+            // make sure that the 2nd request is eventually processed
+            while (coordinator.getNumberOfPendingCheckpoints() == 0) {
+                Thread.sleep(1);
+            }
         } finally {
+            if (coordinator != null) {
+                coordinator.shutdown();
+            }
             executorService.shutdownNow();
         }
     }