fredia commented on code in PR #20420:
URL: https://github.com/apache/flink/pull/20420#discussion_r939922525
##########
flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescaleCheckpointManuallyITCase.java:
##########
@@ -150,54 +137,33 @@ public void testCheckpointRescalingKeyedState(boolean
scaleOut) throws Exception
numberKeys,
numberElements2,
numberElements + numberElements2,
- client,
+ miniCluster,
checkpointPath);
}
- private static String runJobAndGetCheckpoint(
+ private String runJobAndGetCheckpoint(
int numberKeys,
int numberElements,
int parallelism,
int maxParallelism,
- ClusterClient<?> client,
- File checkpointDir)
+ MiniCluster miniCluster)
throws Exception {
try {
- Duration timeout = Duration.ofMinutes(5);
- Deadline deadline = Deadline.now().plus(timeout);
-
JobGraph jobGraph =
createJobGraphWithKeyedState(
- parallelism, maxParallelism, numberKeys,
numberElements, false, 100);
- client.submitJob(jobGraph).get();
-
- assertTrue(
- SubtaskIndexFlatMapper.workCompletedLatch.await(
- deadline.timeLeft().toMillis(),
TimeUnit.MILLISECONDS));
-
- // verify the current state
- Set<Tuple2<Integer, Integer>> actualResult =
CollectionSink.getElementsSet();
-
- Set<Tuple2<Integer, Integer>> expectedResult = new HashSet<>();
-
- for (int key = 0; key < numberKeys; key++) {
- int keyGroupIndex =
KeyGroupRangeAssignment.assignToKeyGroup(key, maxParallelism);
- expectedResult.add(
- Tuple2.of(
-
KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
- maxParallelism, parallelism,
keyGroupIndex),
- numberElements * key));
- }
-
- assertEquals(expectedResult, actualResult);
-
- // ensure contents of state within SubtaskIndexFlatMapper are all
included
- // in the last checkpoint (refer to FLINK-26882 for more details).
-
cluster.getMiniCluster().triggerCheckpoint(jobGraph.getJobID()).get();
-
- client.cancel(jobGraph.getJobID()).get();
- TestUtils.waitUntilJobCanceled(jobGraph.getJobID(), client);
- return
TestUtils.getMostRecentCompletedCheckpoint(checkpointDir).getAbsolutePath();
+ parallelism,
+ maxParallelism,
+ numberKeys,
+ numberElements,
+ numberElements,
+ true,
+ 100);
+ miniCluster.submitJob(jobGraph).get();
+ miniCluster.requestJobResult(jobGraph.getJobID()).get();
+ // The elements may not all be sent to sink when unaligned
checkpoints enabled(refer to
+ // FLINK-26882 for more details).
+ // Don't verify current state here.
+ return getLatestCompletedCheckpointPath(jobGraph.getJobID(),
miniCluster).get();
Review Comment:
Thanks for pointing this out, it happens because checkpoint in
`NotifyingDefiniteKeySource` completed does not mean the checkpoint of the
whole job is completed, I updated the wait condition in
`NotifyingDefiniteKeySource` to fix this.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
