Guotao Ding created FLINK-41456:
-----------------------------------
Summary: 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
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)
