amliuyong commented on issue #9236:
URL: https://github.com/apache/paimon/issues/9236#issuecomment-6486795099
We hit the runtime case described above ("The same completeness invariant
must also hold at runtime ...") in a normal run, without any failover. Adding a
reproduction here rather than opening a new issue.
**Observed** with Paimon 2.0.0 and Flink 1.20.4: a Flink job writing a
bounded `datagen` source (3000 rows) into an append table (`bucket = -1`, 3
partitions), streaming mode, checkpoint interval 3 s, sink parallelism 2. All
3000 rows were written to data files, but only 1506 were committed. The job
finished normally, with no exception or warning. The 1494 rows that one writer
flushed at end of input are in data files that no snapshot references.
Snapshots: identifier 1 (904 rows), then, both created when checkpoint 2
completed, identifier 2 (2 rows) and identifier `Long.MAX_VALUE` (600 rows,
only the other writer's end-of-input files).
**Sequence.** `CommitterOperator` is the same in 2.0.0, 2.1.0 and master
(071e9ee):
1. Writer A reaches end of input and emits its committables with checkpoint
id `Long.MAX_VALUE`.
2. Checkpoint N: the committer's `snapshotState()` polls A's END_INPUT
committables into state. Writer B is still running.
3. Writer B reaches end of input after barrier N and emits its END_INPUT
committables. EndOfData then reaches the committer, and `endInput()` sets
`endInput = true` and returns without polling, because streaming checkpointing
is enabled
([L180-L193](https://github.com/apache/paimon/blob/071e9eee2bb34f66afc0f293144f993109098667/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java#L180-L193)).
4. Only then does the committer handle `notifyCheckpointComplete(N)`,
because its previous commit was slow. As `endInput` is true, it commits up to
`END_INPUT_CHECKPOINT_ID`
([L198](https://github.com/apache/paimon/blob/071e9eee2bb34f66afc0f293144f993109098667/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java#L198)),
which creates the `Long.MAX_VALUE` snapshot with A's data only.
5. Checkpoint N+1 polls B's committables. On its completion they go through
`filterAndCommit`
([L211-L219](https://github.com/apache/paimon/blob/071e9eee2bb34f66afc0f293144f993109098667/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java#L211-L219)),
and `FileStoreCommitImpl#filterCommitted` drops them, because this commit user
already has a snapshot with identifier `Long.MAX_VALUE`
([L320](https://github.com/apache/paimon/blob/071e9eee2bb34f66afc0f293144f993109098667/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java#L320)).
**Harness test.** On master 071e9ee the test below fails with `expected:
["1, 10", "2, 20", "3, 30"] but was: ["1, 10", "2, 20"]`. With #9284 applied to
the same commit it passes, and so do the other tests in `CommitterOperatorTest`
and `AutoTagForSavepointCommitterOperatorTest`. #9284 has no test for this
no-failover path yet; feel free to use this one.
<details>
<summary><code>CommitterOperatorTest#testEndInputOfWritersFinishingInDifferentCheckpoints</code></summary>
```java
@Test
public void testEndInputOfWritersFinishingInDifferentCheckpoints()
throws Exception {
FileStoreTable table =
createFileStoreTable(
options -> {
options.setString("bucket", "-1");
options.remove("bucket-key");
});
OneInputStreamOperatorTestHarness<Committable, Committable>
testHarness =
createRecoverableTestHarness(table);
testHarness.open();
// two writer subtasks of a bounded streaming job
StreamWriteBuilder streamWriteBuilder =
table.newStreamWriteBuilder().withCommitUser(initialCommitUser);
StreamTableWrite write1 = streamWriteBuilder.newWrite();
StreamTableWrite write2 = streamWriteBuilder.newWrite();
long timestamp = 1;
// writer 1 reaches end of input before checkpoint 1
write1.write(GenericRow.of(1, 10L));
for (CommitMessage committable : write1.prepareCommit(true,
Long.MAX_VALUE)) {
testHarness.processElement(new Committable(Long.MAX_VALUE,
committable), timestamp++);
}
// writer 2 is still running at checkpoint 1
write2.write(GenericRow.of(2, 20L));
for (CommitMessage committable : write2.prepareCommit(false, 1)) {
testHarness.processElement(new Committable(1, committable),
timestamp++);
}
testHarness.snapshot(1, timestamp++);
// writer 2 reaches end of input after checkpoint 1 is snapshotted,
but before the
// committer is notified that checkpoint 1 is complete
write2.write(GenericRow.of(3, 30L));
for (CommitMessage committable : write2.prepareCommit(true,
Long.MAX_VALUE)) {
testHarness.processElement(new Committable(Long.MAX_VALUE,
committable), timestamp++);
}
testHarness.endInput();
testHarness.notifyOfCompletedCheckpoint(1);
// the final checkpoint, taken after end of input
testHarness.snapshot(2, timestamp++);
testHarness.notifyOfCompletedCheckpoint(2);
// the end-of-input data of both writers is committed in one
END_INPUT snapshot
assertResults(table, "1, 10", "2, 20", "3, 30");
Snapshot latest = table.snapshotManager().latestSnapshot();
assertThat(latest.commitIdentifier()).isEqualTo(Long.MAX_VALUE);
assertThat(latest.deltaRecordCount()).isEqualTo(2L);
write1.close();
write2.close();
testHarness.close();
}
```
</details>
We also reproduced it end to end on 2.0.0 with a Flink MiniCluster,
`FlinkSinkBuilder` and sink parallelism 2, with `write-only` both true and
false. We forced the timing: one source subtask finishes after its second
checkpoint, and a commit callback sleeps in the first commit. In 20 of 20 runs,
one writer's end-of-input data was lost, and the job still finished normally.
**Not affected:** batch mode, where `endInput()` polls and commits at once;
sink parallelism 1; unbounded input.
**Workarounds** until a fix is released: run bounded writes in batch mode
(`execution.runtime-mode = batch`), or set `sink.parallelism = 1` for streaming
writes of bounded input.
--
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]
