Iceberg Flink sinks duplicate data during recovery when used with the REST catalog
- Dominant language
- Java
- Stars
- 9.2k
- Forks
- 3.5k
- Avg merge
- 2d 11h
- Merged PRs (30d)
- 132
Description
### Solution
The PR, which resolves the issue is in https://github.com/apache/iceberg/pull/14517.
### Apache Iceberg version
1.10.0 (latest release)
### Query engine
Flink
### Please describe the bug 🐞
The `DynamicIcebergSink` used with the Iceberg REST catalog duplicates data after recovering from a failed POST commit request. The `getMaxCommittedCheckpointId()` check in `DynamicCommitter` leads to data duplication if the first commit request, which is still being processed in the REST catalog, is committed after the `getMaxCommittedCheckpointId()`, but before the refresh of the base snapshot in `SnapshotProducer.apply()`. This situation is illustrated in the diagram below.
The commit duplication happens in the following sequence of events:
1. `DynamicIcebergSink` starts the checkpoint 1 (`chk-1`).
2. `DynamicIcebergSink` completes `chk-1` and saves the commit request 1 (`CR-1`) with data/delete files to the checkpoint state.
3. `DynamicCommitter` finds no commits for `chk-1` (`maxCheckpointId` is `-1`) and sends the POST table update request (`Req-1`) to the REST catalog.
4. REST catalog receives the request and triggers the table update in the underlying catalog (eg JDBC).
5. `DynamicCommitter` restarts due to a Flink task manager process failure:
- This can be any failure, even a shutdown of the Flink client JVM; one example is a `SocketTimeoutException` in the REST client, which happens when the REST catalog was too slow to commit to the underlying catalog, for example, due to retrying for too long.
6. `DynamicCommitter` restarts and initiates the commit sequence:
- Loads the table from the REST catalog;
- Checks the max committed checkpoint ID in the table snapshot history and decides to commit because `Req-1` has not completed yet, so the `maxCheckpointId` is still `-1`.
7. `Req-1` completes in the REST catalog and sets the current table snapshot ID to `1`.
8. `DynamicCommitter` gets into the `SnapshotProducer.apply()` call and triggers the `SnapshotProducer.refresh()`:
- This fetches the latest table snapshot `1`, which has been just set by the `Req-1`, and is different to the one before the `getMaxCommittedCheckpointId()` check (🔥🔥 RACE CONDITION 🔥🔥);
- `SnapshotProducer.apply()` sets the parent snapshot ID to `1` and sends a duplicate table update request (POST) to the REST catalog.
9. REST catalog receives the duplicate request and commits to the underlying catalog:
- The commit lands in the catalog without any retries because the parent snapshot of the `Req-2` is equal to the latest table snapshot, set by `Req-1` in the previous step.
This data duplication scenario is depicted in the following sequence diagram:
The `IcebergSink` will have the same issue due to identical code paths in `IcebergCommitter` and `DynamicCommitter`.
### Willingness to contribute
- [x] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time
Contributor guide
Research direction
Start with DynamicCommitter and IcebergCommitter, then trace getMaxCommittedCheckpointId() through SnapshotProducer.apply() and refresh() for the REST-catalog recovery sequence described here. Compare the behavior with PR #14517 and verify that recovery no longer permits the duplicate commit scenario for both sinks.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 25/100