AHeise commented on a change in pull request #15589:
URL: https://github.com/apache/flink/pull/15589#discussion_r614609306
##########
File path:
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/recovery/DemultiplexingRecordDeserializerTest.java
##########
@@ -123,7 +122,8 @@ public void testUpscale() throws IOException {
assertEquals(
Arrays.asList(start + 1L, start + 2L, start + 3L),
readLongs(deserializer));
- assertTrue(bufferBuilder.getMemorySegment().isFreed());
+
+ bufferBuilder.recycle();
Review comment:
What is this revert about? I can't connect it to any other change in
this PR.
##########
File path:
flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterStopWithSavepointITCase.java
##########
@@ -313,20 +315,21 @@ private void setUpJobGraph(
}
private void waitForJob() throws Exception {
- for (int i = 0; i < 60; i++) {
- try {
- final JobStatus jobStatus =
-
clusterClient.getJobStatus(jobGraph.getJobID()).get(60, TimeUnit.SECONDS);
- assertThat(jobStatus.isGloballyTerminalState(),
equalTo(false));
- if (jobStatus == JobStatus.RUNNING) {
- return;
- }
- } catch (ExecutionException ignored) {
- // JobManagerRunner is not yet registered in Dispatcher
- }
- Thread.sleep(1000);
- }
- throw new AssertionError("Job did not become running within timeout.");
+ Deadline deadline = Deadline.fromNow(Duration.ofMinutes(5));
+ JobID jobID = jobGraph.getJobID();
+ CommonTestUtils.waitUntilCondition(
+ () ->
+ clusterClient.getJobStatus(jobID).get(60,
TimeUnit.SECONDS)
+ == JobStatus.RUNNING,
+ deadline,
+ 20L);
Review comment:
If this is necessary, it should be done inside `waitForAllTaskRunning`
--
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]
