developer-rpai commented on code in PR #18439:
URL: https://github.com/apache/iceberg/pull/18439#discussion_r4320972830
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java:
##########
@@ -143,16 +142,38 @@ public void processElement1(StreamRecord<DVWriteResult>
record) {
@Override
public void processElement2(StreamRecord<EqualityConvertPlan> record) {
+ if (record.getValue().cycleStart()) {
+ // The planner started a conversion cycle. Hold back watermarks until
the plan arrives; the
+ // planner emits phase watermarks before the plan, and forwarding those
would release the
+ // maintenance lock before the commit.
+ cycleInFlight = true;
+ return;
+ }
+
planResult = record.getValue();
+ cycleInFlight = true;
}
@Override
public void processWatermark(Watermark mark) throws Exception {
- 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;
}
Review Comment:
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?
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java:
##########
@@ -143,16 +142,38 @@ public void processElement1(StreamRecord<DVWriteResult>
record) {
@Override
public void processElement2(StreamRecord<EqualityConvertPlan> record) {
+ if (record.getValue().cycleStart()) {
Review Comment:
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.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java:
##########
@@ -143,16 +142,38 @@ public void processElement1(StreamRecord<DVWriteResult>
record) {
@Override
public void processElement2(StreamRecord<EqualityConvertPlan> record) {
+ if (record.getValue().cycleStart()) {
+ // The planner started a conversion cycle. Hold back watermarks until
the plan arrives; the
+ // planner emits phase watermarks before the plan, and forwarding those
would release the
+ // maintenance lock before the commit.
+ cycleInFlight = true;
+ return;
+ }
+
planResult = record.getValue();
+ cycleInFlight = true;
}
@Override
public void processWatermark(Watermark mark) throws Exception {
- 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;
}
Review Comment:
Good catch, that block only reworded the existing comment. Restored the
original in cef37c0.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java:
##########
@@ -82,7 +82,9 @@
* {@link #advancePhase()}.
*
* <p>An {@link EqualityConvertPlan} with the current cycle's metadata is
emitted via the {@link
- * #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.
Review Comment:
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.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java:
##########
@@ -133,6 +133,24 @@ void holdsBackWatermarkUntilCommit() throws Exception {
}
}
+ @Test
+ void forwardsWatermarkWithoutActiveCycle() throws Exception {
Review Comment:
Agreed. I added TestConvertEqualityDeletesSharedLock, which runs
ConvertEqualityDeletes next to another task sharing the same lock and checks
the lock is released after each cycle.
--
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]
