You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/08/02 10:48:27 UTC
[GitHub] [flink] Myasuka commented on a diff in pull request #20420: [FLINK-28626][tests] Refactor RescaleCheckpointManuallyITCase
Myasuka commented on code in PR #20420:
URL: https://github.com/apache/flink/pull/20420#discussion_r935407988
##########
flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescaleCheckpointManuallyITCase.java:
##########
@@ -322,10 +289,14 @@ public void run(SourceContext<Integer> ctx) throws Exception {
counter++;
}
} else {
- if (terminateAfterEmission) {
- running = false;
+ while (completedCheckpointNum.get() <= 1) {
+ Thread.sleep(50);
+ }
+ if (failAfterEmission) {
+ throw new FlinkRuntimeException(
+ "Make job fail artificially, to retain completed checkpoint.");
Review Comment:
Why we need to fail the job to get the retained checkpoint?
##########
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).
Review Comment:
Do we need to change the doc `refer to FLINK-26882 for more details` here?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org