Jackeyzhe commented on code in PR #29161:
URL: https://github.com/apache/flink/pull/29161#discussion_r3996641035
##########
flink-tests/src/test/java/org/apache/flink/test/checkpointing/FileMergingChannelStateITCase.java:
##########
@@ -216,37 +228,135 @@ private StreamExecutionEnvironment createEnvironment(
return env;
}
+ /**
+ * Returns the path of a completed checkpoint that carries in-flight input
channel state for the
+ * slow mapper.
+ *
+ * <p>Whether a particular checkpoint contains in-flight data for a
particular subtask depends
+ * on where the barriers happen to be when the checkpoint is triggered, so
waiting for a fixed
+ * number of checkpoints - or for the latest checkpoint that persisted
<em>any</em> in-flight
+ * data anywhere in the job - also accepts checkpoints that do not
exercise channel state
+ * recovery for the mapper at all. Inspect the metadata of every completed
checkpoint instead
+ * and return the first one that really contains the state this test is
about.
+ */
+ private static String waitForCheckpointWithSlowMapperChannelState(
+ JobID jobID, MiniCluster miniCluster) throws Exception {
+ final Set<Long> inspectedCheckpoints = new HashSet<>();
+ final AtomicReference<String> restorePath = new AtomicReference<>();
+ CommonTestUtils.waitUntilCondition(
+ () -> {
+ final AccessExecutionGraph graph =
miniCluster.getExecutionGraph(jobID).get();
+ for (CompletedCheckpointStats checkpoint :
+ checkpointsNotInspectedYet(graph,
inspectedCheckpoints)) {
+ if (carriesSlowMapperChannelState(checkpoint)) {
+ restorePath.set(checkpoint.getExternalPath());
+ return true;
+ }
+ }
+ failIfJobStoppedCheckpointing(graph, inspectedCheckpoints);
+ return false;
+ });
+ return restorePath.get();
+ }
+
+ /**
+ * Returns the retained checkpoints that persisted in-flight data and have
not been looked at by
+ * an earlier call, oldest first: the earliest usable checkpoint is the
one that leaves the most
+ * records for the restored job to replay.
+ */
+ private static List<CompletedCheckpointStats> checkpointsNotInspectedYet(
+ AccessExecutionGraph graph, Set<Long> inspectedCheckpoints) {
+ final CheckpointStatsSnapshot snapshot =
graph.getCheckpointStatsSnapshot();
+ if (snapshot == null) {
+ return Collections.emptyList();
+ }
+ // The history is ordered from the newest to the oldest checkpoint.
+ final List<AbstractCheckpointStats> history =
+ new ArrayList<>(snapshot.getHistory().getCheckpoints());
+ Collections.reverse(history);
+ return history.stream()
+ .filter(CompletedCheckpointStats.class::isInstance)
+ .map(CompletedCheckpointStats.class::cast)
+ .filter(checkpoint -> checkpoint.getPersistedData() > 0L)
+ .filter(checkpoint -> checkpoint.getExternalPath() != null)
+ .filter(checkpoint ->
inspectedCheckpoints.add(checkpoint.getCheckpointId()))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Returns whether restoring from the given checkpoint would exercise
file-merged channel state
+ * recovery, i.e. whether it holds in-flight input channel state for the
slow mapper.
+ */
+ private static boolean
carriesSlowMapperChannelState(CompletedCheckpointStats checkpoint) {
+ try {
+ final CheckpointMetadata metadata =
+
TestUtils.loadCheckpointMetadata(checkpoint.getExternalPath());
+ return
!collectChannelStateDelegates(metadata).slowMapperInputChannelState.isEmpty();
+ } catch (IOException e) {
Review Comment:
This catches more than the subsumption race described by the
comment.`loadCheckpointMetadata` also reports corrupt headers, unsupported
versions, truncated input, and deserialization failures as `IOException`. Since
this checkpoint was already added to `inspectedCheckpoints`, those failures are
silently discarded and a later checkpoint can make the test pass. Could we
suppress only the missing/cleaned-up checkpoint case and rethrow other metadata
read failures?
--
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]
