AHeise commented on a change in pull request #13604:
URL: https://github.com/apache/flink/pull/13604#discussion_r504913074
##########
File path:
flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointCompatibilityITCase.java
##########
@@ -107,28 +119,55 @@ public void test() throws Exception {
}
private Tuple2<String, Map<String, Object>> runAndTakeSavepoint()
throws Exception {
- JobClient jobClient = submitJobInitially(env(startAligned, 0,
emptyMap()));
- Thread.sleep(FIRST_RUN_EL_COUNT * FIRST_RUN_BACKPRESSURE_MS);
// wait for all tasks to run and some backpressure from sink
- Future<Map<String, Object>> accFuture =
jobClient.getAccumulators();
- Future<String> savepointFuture =
jobClient.stopWithSavepoint(false, tempFolder().toURI().toString());
- return new Tuple2<>(savepointFuture.get(), accFuture.get());
+ return withCluster(new Configuration(), miniCluster -> {
+ JobClient jobClient =
submitJobInitially(env(startAligned, 0));
+ waitForAllTaskRunning(miniCluster,
jobClient.getJobID(), Deadline.fromNow(Duration.of(30, ChronoUnit.SECONDS)));
+ Thread.sleep(FIRST_RUN_BACKPRESSURE_MS); // wait for
some backpressure from sink
Review comment:
The original test used sleeping for waiting until all jobs are running
(hence this bug). However, whether backpressure build up or not is quite
irrelevant for the semantics of the test. It's just testing that some outputs
are present. I could probably wait on the output metric.
However, it feels like LOTS of such building blocks are missing and should
probably be at a more general place. (I tried to do that with
`waitForAllTaskRunning` but it's easily overlooked. Ideally it should be in
MiniCluster).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
