mxm commented on code in PR #18439:
URL: https://github.com/apache/iceberg/pull/18439#discussion_r4315133853
##########
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:
This should work without using a flag:
```java
@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;
}
```
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java:
##########
@@ -93,6 +91,7 @@ public class EqualityConvertCommitter extends
AbstractStreamOperator<Trigger>
private transient Table table;
private transient List<DVWriteResult> bufferedResults;
private transient EqualityConvertPlan planResult;
+ private transient boolean cycleInFlight;
Review Comment:
We can remove this flag with the above change.
--
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]
