openinx commented on code in PR #18439:
URL: https://github.com/apache/iceberg/pull/18439#discussion_r4254627825
##########
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:
Hi @developer-rpai , I'm catching the new flink integration, what is the
`cycle` meaning here ?
##########
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:
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.
--
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]
