fmorillo7694 opened a new issue, #18435:
URL: https://github.com/apache/iceberg/issues/18435
### Apache Iceberg version
1.12.0 (latest release)
### Query engine
Flink
### Please describe the bug 🐞
`ConvertEqualityDeletes` (new in 1.12.0) holds the Flink-maintained
maintenance lock forever once any other maintenance task on the same table has
run. After that, neither task runs again.
**Environment**
- Iceberg 1.12.0, `iceberg-flink-runtime-2.3-1.12.0.jar`, Flink 2.3.0
- `IcebergSink` (`table.exec.iceberg.use-v2-sink = true`), maintenance
configured through `flink-maintenance.*` table options
- Default Flink-maintained lock (no `flink-maintenance.lock.type`)
- format-version 3, upsert table with a primary key, parallelism 1, REST
catalog
- The code involved is unchanged on `main`
**Reproduce**
```sql
SET 'table.exec.iceberg.use-v2-sink' = 'true';
SET 'execution.checkpointing.interval' = '15s';
CREATE TABLE cat.db.t (
id BIGINT, v STRING, PRIMARY KEY (id) NOT ENFORCED
) WITH (
'format-version' = '3',
'write.upsert.enabled' = 'true',
'flink-maintenance.convert-equality-deletes.enabled' = 'true',
'flink-maintenance.convert-equality-deletes.schedule.commit-count' = '1',
'flink-maintenance.rewrite.enabled' = 'true',
'flink-maintenance.rewrite.schedule.commit-count' = '2',
'flink-maintenance.rate-limit-seconds' = '5',
'flink-maintenance.lock-check-delay-seconds' = '5'
);
CREATE TEMPORARY TABLE src (id BIGINT, v STRING) WITH (
'connector' = 'datagen', 'rows-per-second' = '1000',
'fields.id.min' = '1', 'fields.id.max' = '10000'
);
INSERT INTO cat.db.t SELECT id, v FROM src;
```
**Expected**
Both tasks keep running, one at a time under the shared lock. The
ConvertEqualityDeletes section of `docs/docs/flink-maintenance.md` says the
shared maintenance lock keeps it safe alongside compaction.
**Actual**
- The converter runs one cycle and the lock is released (`Received lock
released event`).
- `RewriteDataFiles [0]` then runs and finishes: `LockRemoverOperator` logs
`Processing result ... RewriteDataFiles [0]`. No `Received lock released event`
follows.
- From then on, `TriggerManagerOperator` logs `Failed to acquire lock.
Delaying task to ...` on every check (21 times in a 120s run, 33 in a 180s
run), and `concurrentRunThrottled` keeps growing.
- Neither task runs again. `processedStagingSnapshotNum` stays at 0, and
equality-delete files pile up: 48 after 120s, 96 after 180s.
- With the converter as the only maintenance task, the same table and load
ran normally in five runs: 31 conversion commits, no errors.
**Cause, as far as we can tell**
1. Every task's pipeline sees every trigger's watermark. `Filter <i>`
(`TableMaintenance.java:329-337`) drops other tasks' records but not their
watermarks. All task outputs are unioned into `Lock remover` (`:349-365`).
2. With the Flink-maintained lock, `LockRemoverOperator` releases only from
`processWatermark`, by sending `LockReleaseEvent(mark)`
(`LockRemoverOperator.java:106-114`). `processElement` only updates metrics.
`TriggerManagerOperator.handleLockRelease` clears the lock only when `timestamp
>= lockTime` (`:254-261`).
3. `EqualityConvertCommitter.processWatermark` returns without forwarding
when `planResult == null` (`EqualityConvertCommitter.java:151-157`).
`planResult` is set only from the converter's own planner, so a watermark from
another task's trigger is dropped.
4. The union's minimum watermark therefore stays at the converter's last
forwarded value, and the lock taken for the other task is never released. Only
a new converter cycle would advance that watermark, and starting one needs the
lock.
The hold-back was added in #17038 to fix a race where the planner's
mid-cycle phase watermarks released the lock early. In-flight cycles are still
held back correctly. The problem is the `planResult == null` case.
**Probably also affected (from reading the code, not reproduced)**
- The converter combined with `expire-snapshots` or `delete-orphan-files`.
Both of those branches forward every watermark, like rewrite.
- A converter-only table restored from a checkpoint or savepoint.
`TriggerManagerOperator.initializeState` takes the lock and emits
`Trigger.recovery(...)`, which has no task id. Only its watermark reaches the
converter committer, which drops it, so the recovery lock would never be
released.
**Possible fix**
- Make the committer forward watermarks whenever no conversion cycle is in
flight. A plain `if (planResult == null) forward` would bring back the #17038
race: the planner emits phase watermarks (`advancePhase`) before it emits the
plan record. One option is for the planner to emit a cycle-start marker on the
metadata stream before its first `advancePhase`. The committer would then hold
watermarks only between that marker and the commit.
- As an extra safeguard, `LockRemoverOperator.processElement` could also
send `LockReleaseEvent(taskResult.startEpoch())`, as the legacy `LockRemover`
does on `TaskResult`. That isn't enough on its own. `TaskResultAggregator`
emits the result before it forwards the watermark, so on the normal path the
watermark's release event would then reach `handleLockRelease` with `lockTime
== null` and fail its `Preconditions.checkArgument`
(`TriggerManagerOperator.java:255`). That check would have to become a no-op
for an already-released lock. It also wouldn't help the restore case, which is
released by watermark only.
Neither `TestMaintenanceE2E` nor `TestIcebergSinkTableMaintenance` runs
`ConvertEqualityDeletes` together with another task. A test that runs both and
checks that each completes at least twice would catch this.
### Willingness to contribute
- [ ] 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
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
