Flink: Forward watermarks in EqualityConvertCommitter with no active cycle by developer-rpai · Pull Request #18439 · apache/iceberg

developer-rpai · 2026-10-09T20:49:49Z

When ConvertEqualityDeletes shares the Flink-maintained lock with another maintenance task, the lock is never released: EqualityConvertCommitter.processWatermark drops watermarks while planResult == null, stalling the unioned watermark that LockRemoverOperator needs to emit the lock release. The planner now emits a cycle-start marker on the metadata stream before its first phase watermark; the committer forwards watermarks while no cycle is active and holds them back from the marker until the commit, so the pre-plan phase watermarks cannot release the lock early (the #17038 race). Fixes #18435.

Test plan: new holdsBackPhaseWatermarkBeforePlanArrives asserts a pre-plan phase watermark is held back and the cycle still commits; forwardsWatermarkWithoutActiveCycle covers idle forwarding; existing holdsBackWatermarkUntilCommit covers mid-cycle hold-back; planner tests assert the marker precedes the plan. Note: the same pattern exists in flink/v1.20 and flink/v2.2 — left for a follow-up to keep this minimal. Could not compile locally (no JDK on this machine); relying on CI.

…s active Fixes apache#18435.

pvary · 2026-10-11T06:27:15Z

  }

+  @Test
+  void forwardsWatermarkWithoutActiveCycle() throws Exception {


We are testing our internal workings, not that the change actually solves the original problem of watermarks created by the task is escaping the task scope

Agreed. I added TestConvertEqualityDeletesSharedLock, which runs ConvertEqualityDeletes next to another task sharing the same lock and checks the lock is released after each cycle.

The planner emits phase watermarks before the plan reaches the committer. Forwarding watermarks while no plan has been received would release the maintenance lock early, reintroducing the race fixed in apache#17038. The planner now emits a cycle-start marker on the metadata stream before its first phase watermark; the committer forwards watermarks only while no cycle is active and holds them back from the marker until the commit.

Reflow Javadoc paragraphs in EqualityConvertPlan and EqualityConvertCommitter to match palantir-java-format.

openinx · 2026-10-13T04:06:15Z


  @Override
  public void processElement2(StreamRecord<EqualityConvertPlan> record) {
+    if (record.getValue().cycleStart()) {


Hi @developer-rpai , I'm catching the new flink integration, what is the cycle meaning here ?

A cycle is one run of the convert task for a single trigger: the planner plans the conversion, the DV writers write, and the committer commits at the plan's doneTimestamp. With the separate flag gone, the committer tracks this through planResult now.

openinx · 2026-10-13T04:10:10Z

+    if (planResult == null) {
+      // The cycle started but the plan has not arrived yet. The planner emits phase watermarks
+      // before the plan; forwarding them would release the maintenance lock before the commit,
+      // letting the next trigger run a concurrent cycle on the same staging snapshot.
+      return;
+    }
+
+    if (mark.getTimestamp() < planResult.doneTimestamp()) {
+      // Hold back watermarks until the cycle commits so the LockRemover keeps the maintenance
+      // lock for the whole cycle.
      return;
    }


Those lines are the same as the left side, let's revert it back to the original version so that we won't introduce any unneeded change.

Good catch, that block only reworded the existing comment. Restored the original in cef37c0.

openinx · 2026-10-13T04:25:47Z

- * #METADATA_STREAM} side output after the read commands.
+ * #METADATA_STREAM} side output after the read commands. A {@link EqualityConvertPlan#cycleStart}
+ * marker is emitted first, before any phase watermark, so the committer can hold back watermarks
+ * until the plan arrives.


what does you mean so the committer can hold back watermarks until the plan arrives. here ?

Reworded it in cef37c0. It now says the planner emits phase watermarks before the plan, so the marker tells the committer a cycle has started and it holds watermarks until the commit.

mxm

Thanks for the PR @developer-rpai!

mxm · 2026-10-19T14:33:00Z

  private transient Table table;
  private transient List<DVWriteResult> bufferedResults;
  private transient EqualityConvertPlan planResult;
+  private transient boolean cycleInFlight;


I think we can remove this flag, see the change below.

mxm · 2026-10-19T14:38:48Z

-    if (planResult == null || mark.getTimestamp() < planResult.doneTimestamp()) {
-      // Hold back watermarks until the cycle commits so the LockRemover keeps the maintenance lock
-      // for the whole cycle. Forwarding the planner's mid-cycle phase watermarks will release the
-      // lock early and could let the next trigger run a concurrent cycle on the same staging
-      // snapshot.
+    if (!cycleInFlight) {
+      // No conversion cycle is active: forward the watermark. It may originate from another
+      // task's trigger sharing the maintenance lock, and holding it back would stall lock release
+      // for that task.
+      super.processWatermark(mark);
+      return;
+    }
+
+    if (planResult == null) {
+      // The cycle started but the plan has not arrived yet. The planner emits phase watermarks
+      // before the plan; forwarding them would release the maintenance lock before the commit,
+      // letting the next trigger run a concurrent cycle on the same staging snapshot.
+      return;
+    }
+
+    if (mark.getTimestamp() < planResult.doneTimestamp()) {
+      // Hold back watermarks until the cycle commits so the LockRemover keeps the maintenance
+      // lock for the whole cycle.
      return;
    }


This should work without using a flag:

@Override public void processWatermark(Watermark mark) throws Exception { if (planResult == null) { // No cycle in flight: the watermark comes from another task's trigger or a recovery trigger // sharing the maintenance lock. Holding it back would keep that lock forever. super.processWatermark(mark); return; } if (mark.getTimestamp() < planResult.doneTimestamp()) { // Hold back watermarks until the cycle commits so the LockRemover keeps the maintenance lock // for the whole cycle. Forwarding the planner's mid-cycle phase watermarks will release the // lock early and could let the next trigger run a concurrent cycle on the same staging return; }

Thanks for the review and the sketch — you're right, the cycleInFlight flag is redundant. I'll drop it and restructure processWatermark along the lines you suggested.

Pushed in cef37c0. One wrinkle with only the null check: the planner emits its phase watermarks before the plan, so those would get forwarded and bring back the early release from #17038. So the cycle-start marker now sets planResult itself. Its doneTimestamp is Long.MAX_VALUE, so your mark < doneTimestamp check holds everything back until the real plan replaces it, and the flag is gone. If a MAX watermark arrives at end of input while the marker is still current, it's forwarded without a commit. Does that work for you?

Drop the cycleInFlight flag. The committer stores the planner's cycle-start marker as the plan result so its maximal done timestamp holds phase watermarks until the real plan arrives, and forwards watermarks when no cycle is active. An end-of-input watermark while the marker is current is forwarded without a commit. Co-authored-by: Rakesh Pai <developer-rpai@users.noreply.github.com> Generated-by: Cursor

github-actions Bot added the flink label Oct 9, 2026

Flink: Forward watermarks in EqualityConvertCommitter when no cycle i…

68bc0b0

…s active Fixes apache#18435.

developer-rpai force-pushed the fix-18435-flink-lock-watermark branch from 2784848 to 68bc0b0 Compare October 9, 2026 23:05

pvary reviewed Oct 11, 2026

View reviewed changes

developer-rpai added 4 commits October 11, 2026 20:50

Fix spotless Javadoc formatting violations

0492605

Reflow Javadoc paragraphs in EqualityConvertPlan and EqualityConvertCommitter to match palantir-java-format.

Fix remaining spotless Javadoc wrapping in EqualityConvertCommitter

9dcc38d

Flink: Advance watermark past marker phase in DVWriter marker test

bb7cc3f

openinx reviewed Oct 13, 2026

View reviewed changes

mxm reviewed Oct 19, 2026

View reviewed changes

Conversation

developer-rpai commented Oct 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mxm left a comment

Choose a reason for hiding this comment

Uh oh!

mxm Oct 19, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

developer-rpai commented Oct 9, 2026 •

edited

Loading

mxm Oct 19, 2026 •

edited

Loading