mxm commented on code in PR #18439: URL: https://github.com/apache/iceberg/pull/18439#discussion_r4323465378
########## flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesSharedLock.java: ########## @@ -0,0 +1,151 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.flink.maintenance.api; + +import static org.apache.iceberg.flink.maintenance.api.TableMaintenance.LOCK_REMOVER_OPERATOR_NAME; +import static org.apache.iceberg.flink.maintenance.operator.TableMaintenanceMetrics.SUCCEEDED_TASK_COUNTER; +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.Serializable; +import java.time.Duration; +import java.util.Collections; +import java.util.List; +import org.apache.flink.api.common.functions.MapFunction; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.java.typeutils.ResultTypeQueryable; +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.SnapshotRef; +import org.apache.iceberg.Table; +import org.apache.iceberg.flink.maintenance.operator.ManualSource; +import org.apache.iceberg.flink.maintenance.operator.MetricsReporterFactoryForTests; +import org.apache.iceberg.flink.maintenance.operator.OperatorTestBase; +import org.apache.iceberg.flink.maintenance.operator.TableChange; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Test; + +/** + * ConvertEqualityDeletes shares the Flink-maintained maintenance lock with another task. The lock + * is released only when the unioned watermark reaches the lock remover, so a watermark from the + * other task has to pass through an idle converter. + */ +class TestConvertEqualityDeletesSharedLock extends OperatorTestBase { + private static final String PROBE_TASK_NAME = "SharedLockProbe"; + + @Test + void releasesSharedLockAfterEachCycle() throws Exception { Review Comment: I would move this to `TestConvertEqualityDeletesE2E`. ########## flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java: ########## @@ -133,6 +133,10 @@ public void processElement1(StreamRecord<DVPosition> record) { @Override public void processElement2(StreamRecord<EqualityConvertPlan> record) { + if (record.getValue().cycleStart()) { + return; + } + Review Comment: I don't think this is needed. ########## flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java: ########## @@ -143,19 +143,38 @@ public void processElement1(StreamRecord<DVWriteResult> record) { @Override public void processElement2(StreamRecord<EqualityConvertPlan> record) { + // Includes the cycle-start marker. Its maximal done timestamp holds watermarks until the real + // plan replaces it. planResult = record.getValue(); } @Override public void processWatermark(Watermark mark) throws Exception { - if (planResult == null || mark.getTimestamp() < planResult.doneTimestamp()) { + 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 // snapshot. return; } + if (planResult.cycleStart()) { + // The marker's done timestamp is maximal, so only an end-of-input watermark reaches here + // before the plan replaces the marker. Forward that watermark and drop the marker; there is + // nothing to commit. + bufferedResults.clear(); + planResult = null; + super.processWatermark(mark); + return; + } Review Comment: If we model this with a NOOP, I don't think we need this code. We won't commit anything and the following code will forward the Trigger for the Aggregator. ########## 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: Yes, that's what I had in mind. I don't think the end-of-input case can happen though. The planner emits the marker and the plan in the same processElement call on the same channel, so the plan always reaches the committer before any later watermark. I've pushed a fixup, which encodes the marker as a no-op plan. -- 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]
