apache / apache/iceberg

Iceberg Flink sinks duplicate data during recovery when used with the REST catalog

Open
#14,425 17 comments 0 reactions 0 assignees View on GitHub
bug
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:

Image

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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.