[
https://issues.apache.org/jira/browse/FLINK-41456?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-41456:
-----------------------------------
Labels: pull-request-available unaligned-checkpoints (was:
unaligned-checkpoints)
> 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)
