[
https://issues.apache.org/jira/browse/FLINK-41456?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18143640#comment-18143640
]
Guotao Ding commented on FLINK-41456:
-------------------------------------
Hi [~fanrui] , could you take a look at this issue when you have time? I'd also
like to work on it, could you assign it to me?
I have opened two PRs:
* master: https://github.com/apache/flink/pull/29892
* release-1.20: https://github.com/apache/flink/pull/29893
The fix moves \{{checkForUnsupportedToplogyChanges}} before the FLINK-31963
short-circuit in \{{reDistributeResultSubpartitionStates}} and
\{{reDistributeInputChannelStates}}. A restore whose in-flight data belongs to
an exchange that is now chained fails with the existing error instead of
dropping the data silently.
We have already applied the same fix in our production environment based on
Flink 1.20.
Both PRs add unit tests in \{{StateAssignmentOperationTest}}. The ITCase in the
description reproduces the issue on release-1.20.
> In-flight data of an unaligned checkpoint is silently dropped when its
> exchange becomes chained after a parallelism change
> --------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41456
> URL: https://issues.apache.org/jira/browse/FLINK-41456
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Checkpointing
> Affects Versions: 2.3.0, 1.20.5
> Reporter: Guotao Ding
> Priority: Critical
> Labels: pull-request-available, unaligned-checkpoints
> Attachments: uc-chained-data-loss.drawio.png
>
>
> h2. Description
> A job restored from an unaligned checkpoint with a different parallelism
> silently loses the in-flight data of a non-pointwise exchange (e.g.
> rebalance) when that exchange becomes a chained forward connection in the
> restored graph.
> The edge in the example below is declared without an explicit partitioner.
> While the two parallelisms differ, Flink materializes it as a rebalance,
> which is why an unaligned checkpoint persists in-flight buffers for it. If
> the restore changes the parallelism so that both operators end up with the
> same parallelism, the same edge is materialized as a forward connection, the
> two operators are chained into a single {{JobVertex}}, and the exchange is
> folded into the chain: no network edge is left between them.
> The in-flight data of that exchange is still in the checkpoint, but there is
> no exchange left to restore it to:
> * The output state ({{ResultSubpartitionStateHandle}}) belongs to the former
> upstream tail, which is no longer the tail (output operator) of its vertex.
> * The input state ({{InputChannelStateHandle}}) belongs to the former
> downstream head, which is no longer the head (input operator) of its vertex.
> *Expected:* the restore is either rejected, or every record still reaches the
> sink.
> *Actual:* the restore completes without an error and without a warning. The
> persisted records are dropped, and nothing in the logs, metrics or the job
> status indicates it.
> Example: {{source(3) \-rebalance-> map(2) \-forward-> sink(2)}} is restored
> with {{source(2)}}. All three operators now have parallelism 2 and are
> chained into one vertex:
> !uc-chained-data-loss.drawio.png!
> h2. Root cause
> {{StateAssignmentOperation}} already has a check for exactly this case.
> {{checkForUnsupportedToplogyChanges}} (sic) fails with:
> {noformat}
> Cannot recover from unaligned checkpoint when topology changes, such that
> data exchanges with
> persisted data are now chained.{noformat}
> However, the check is never reached. In
> {{reDistributeResultSubpartitionStates}} and
> {{reDistributeInputChannelStates}} it runs after an early return:
> {code:java}
> public void reDistributeResultSubpartitionStates(TaskStateAssignment
> assignment) {
> // FLINK-31963: We can skip this phase if there is no output state AND
> downstream has no
> // input states
> if (!assignment.hasOutputState() &&
> !assignment.hasDownstreamInputStates()) {
> return;
> }
> checkForUnsupportedToplogyChanges(
> assignment.oldState,
> OperatorSubtaskState::getResultSubpartitionState,
> assignment.outputOperatorID);
> ...
> }
> {code}
> {{hasOutputState()}} and {{hasDownstreamInputStates()}} only look at the
> output (tail) and input (head) operator of the vertex. When the in-flight
> data is owned by an operator that is chained in the new graph, that operator
> is neither the tail nor the head, so both conditions are false, the method
> returns early, and the check that would have caught the case is skipped.
> The early return was introduced by FLINK-19801 and later extended by
> FLINK-31963. In both versions it only looks at the output (tail) and input
> (head) operator, so the check placed after it is skipped for this case.
> h2. How to reproduce
> The following ITCase runs {{source(3) \-> map(2) \-> sink(2)}} with unaligned
> checkpoints and a slow map, waits for a checkpoint with in-flight buffers,
> cancels the job and restores it as {{source(2) \-> map(2) \-> sink(2)}}. The
> edge is declared without a partitioner, so phase 1 sees a rebalance
> (parallelisms differ) and phase 2 sees a forward connection (parallelisms
> match). The sink keeps all received records in operator state, so after the
> restored job finishes, every record of the bounded source must be present.
> On {{release-1.20}} ({{9957486cbfc}}) the test fails, e.g.:
> {noformat}
> [2947 of 50000 records were lost after recovering from the unaligned
> checkpoint]{noformat}
> The number of lost records varies with the amount of in-flight data in the
> checkpoint.
> The test is written against {{release-1.20}}. It uses the legacy
> {{SinkFunction}} API, which no longer exists on master; the affected code
> path in {{StateAssignmentOperation}} is unchanged on master.
> {code:java}
> package org.apache.flink.test.checkpointing;
> // imports omitted
> /**
> * Phase 1 runs {@code source(3) -> map(2) -> sink(2)}: the source to map
> exchange is a rebalance,
> * so the checkpoint persists in-flight buffers for it. Phase 2 restores with
> {@code source(2) ->
> * map(2) -> sink(2)}, where the same exchange becomes a forward connection
> inside a single chained
> * task. The recovery must either be rejected, or every record must still
> reach the sink.
> */
> public class UnalignedCheckpointChainedExchangeITCase extends TestLogger {
> private static final long NUM_RECORDS = 50_000;
> private static final Set<Long> RESULTS = ConcurrentHashMap.newKeySet();
> private static MiniClusterWithClientResource cluster;
> @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
> @Before
> public void setup() throws Exception {
> RESULTS.clear();
> cluster =
> new MiniClusterWithClientResource(
> new MiniClusterResourceConfiguration.Builder()
> .setNumberTaskManagers(1)
> .setNumberSlotsPerTaskManager(10)
> .build());
> cluster.before();
> }
> @After
> public void shutDownExistingCluster() {
> if (cluster != null) {
> cluster.after();
> cluster = null;
> }
> }
> @Test
> public void testNoInFlightDataLossWhenExchangeBecomesChained() throws
> Exception {
> final MiniCluster miniCluster = cluster.getMiniCluster();
> // Phase 1: source(3) -> map(2) is a rebalance exchange with
> in-flight data.
> final JobClient jobClient1 =
> executeJob(getEnv(null), 3,
> RateLimiterStrategy.perSecond(5000), true);
> CommonTestUtils.waitForJobStatus(jobClient1,
> Collections.singletonList(JobStatus.RUNNING));
> CommonTestUtils.waitForAllTaskRunning(miniCluster,
> jobClient1.getJobID(), false);
> final String checkpointPath =
> CommonTestUtils.waitForCheckpointWithInflightBuffers(
> jobClient1.getJobID(), miniCluster);
> jobClient1.cancel().get();
> // Phase 2: source(2) -> map(2) becomes a forward exchange chained
> into one task.
> final JobClient jobClient2 =
> executeJob(getEnv(checkpointPath), 2,
> RateLimiterStrategy.noOp(), false);
> try {
> jobClient2.getJobExecutionResult().get(5, TimeUnit.MINUTES);
> } catch (ExecutionException e) {
> assertThat(
> ExceptionUtils.findThrowableWithMessage(
> e, "data exchanges with persisted data
> are now chained"))
> .as("The recovery failed for an unexpected reason: %s", e)
> .isPresent();
> return;
> }
> final List<Long> missing = new ArrayList<>();
> for (long i = 0; i < NUM_RECORDS; i++) {
> if (!RESULTS.contains(i)) {
> missing.add(i);
> }
> }
> assertThat(missing)
> .as(
> "%d of %d records were lost after recovering from the
> unaligned checkpoint",
> missing.size(), NUM_RECORDS)
> .isEmpty();
> }
> private StreamExecutionEnvironment getEnv(@Nullable String recoveryPath)
> throws IOException {
> final Configuration conf = new Configuration();
> conf.set(CheckpointingOptions.CHECKPOINTING_INTERVAL,
> Duration.ofSeconds(1));
> conf.set(CheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT,
> Duration.ofSeconds(0));
> conf.set(
> CheckpointingOptions.EXTERNALIZED_CHECKPOINT_RETENTION,
> ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION);
> conf.set(
> CheckpointingOptions.CHECKPOINTS_DIRECTORY,
> temporaryFolder.newFolder().toURI().toString());
> conf.set(CheckpointingOptions.ENABLE_UNALIGNED, true);
> conf.set(TaskManagerOptions.MEMORY_SEGMENT_SIZE, MemorySize.parse("1
> kb"));
> conf.set(CheckpointingOptions.MAX_RETAINED_CHECKPOINTS, 50);
> conf.set(RestartStrategyOptions.RESTART_STRATEGY, "none");
> if (recoveryPath != null) {
> conf.set(StateRecoveryOptions.SAVEPOINT_PATH, recoveryPath);
> }
> // Chaining stays enabled: the forward exchange of phase 2 must be
> chained.
> return StreamExecutionEnvironment.getExecutionEnvironment(conf);
> }
> private static JobClient executeJob(
> StreamExecutionEnvironment env,
> int sourceParallelism,
> RateLimiterStrategy rateLimiter,
> boolean slowMap)
> throws Exception {
> env.fromSource(
> new DataGeneratorSource<>(
> index -> index, NUM_RECORDS, rateLimiter,
> Types.LONG),
> WatermarkStrategy.noWatermarks(),
> "Source")
> .uid("source")
> .setParallelism(sourceParallelism)
> .map(
> x -> {
> if (slowMap) {
> Thread.sleep(1);
> }
> return x;
> })
> .uid("map")
> .setParallelism(2)
> .addSink(new CheckpointedCollectingSink())
> .uid("sink")
> .setParallelism(2);
> return env.executeAsync();
> }
> /** Keeps every received record in operator state and publishes them on
> finish. */
> private static class CheckpointedCollectingSink extends
> RichSinkFunction<Long>
> implements CheckpointedFunction {
> private final List<Long> received = new ArrayList<>();
> private transient ListState<Long> state;
> @Override
> public void invoke(Long value, Context context) {
> received.add(value);
> }
> @Override
> public void snapshotState(FunctionSnapshotContext context) throws
> Exception {
> state.update(received);
> }
> @Override
> public void initializeState(FunctionInitializationContext context)
> throws Exception {
> state =
> context.getOperatorStateStore()
> .getListState(new
> ListStateDescriptor<>("received", Types.LONG));
> if (context.isRestored()) {
> state.get().forEach(received::add);
> }
> }
> @Override
> public void finish() {
> RESULTS.addAll(received);
> }
> }
> }
> {code}
> h2. Proposed fix
> Run {{checkForUnsupportedToplogyChanges}} before the early return in both
> {{reDistributeResultSubpartitionStates}} and
> {{reDistributeInputChannelStates}}, and keep the short-circuit condition
> itself unchanged.
> Why moving it is safe:
> * The check only iterates over the checkpoint metadata on the JobManager. It
> does not read any state files, so it costs nothing on the restore path; the
> state files are still only touched for the operators that actually receive
> state.
> * It only counts handles with offsets as data ({{getOffsets().isEmpty()}}),
> so a non-pointwise exchange that becomes forward without in-flight data is
> still restored as before. The short circuit keeps doing its job in that case.
> * When the topology does not change, a chained operator carries no channel
> state of its own, so the check stays a no-op for the jobs that were already
> fine.
> This makes the forward case consistent with a closely related one. If the
> same exchange becomes a rescale instead of a forward connection (e.g.
> {{source(4).rescale() \-> map(2)}}), the recovery already fails today,
> because the pointwise partitioner maps to {{SubtaskStateMapper.UNSUPPORTED}}:
> {noformat}
> Cannot rescale the given pointwise partitioner.{noformat}
> In both cases the persisted in-flight data cannot be restored, so the
> recovery should fail explicitly instead of continuing without the data.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
