Copilot commented on code in PR #3509: URL: https://github.com/apache/fluss/pull/3509#discussion_r3456156960
########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeCommitter.java: ########## @@ -0,0 +1,214 @@ +/* + * 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.fluss.lake.hudi.tiering; + +import org.apache.fluss.lake.committer.CommittedLakeSnapshot; +import org.apache.fluss.lake.committer.LakeCommitResult; +import org.apache.fluss.lake.committer.LakeCommitter; +import org.apache.fluss.lake.hudi.utils.meta.CkpMetadata; +import org.apache.fluss.lake.hudi.utils.meta.CkpMetadataProvider; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.utils.IOUtils; + +import org.apache.hudi.client.HoodieFlinkWriteClient; +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.exception.HoodieException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static org.apache.fluss.lake.writer.LakeTieringFactory.FLUSS_LAKE_TIERING_COMMIT_USER; + +/** Hudi implementation of {@link LakeCommitter}. */ +public class HudiLakeCommitter implements LakeCommitter<HudiWriteResult, HudiCommittable> { + + private static final Logger LOG = LoggerFactory.getLogger(HudiLakeCommitter.class); + + private static final String COMMITTER_USER = "commit-user"; + + private final HudiWriteTableInfo hudiTableInfo; + private final HoodieFlinkWriteClient<?> writeClient; + private final CkpMetadata ckpMetadata; + + public HudiLakeCommitter( + HudiCatalogProvider hudiCatalogProvider, + CkpMetadataProvider ckpMetadataProvider, + TablePath tablePath) + throws IOException { + this.hudiTableInfo = HudiWriteTableInfo.create(hudiCatalogProvider, tablePath); + this.writeClient = hudiTableInfo.getWriteClient(); + this.ckpMetadata = ckpMetadataProvider.get(tablePath, hudiTableInfo); + LOG.info( + "Created HudiLakeCommitter with configuration {}.", hudiTableInfo.getFlinkConfig()); + } + + @Override + public HudiCommittable toCommittable(List<HudiWriteResult> hudiWriteResults) { + HudiCommittable.Builder committableBuilder = HudiCommittable.builder(); + for (HudiWriteResult hudiWriteResult : hudiWriteResults) { + committableBuilder.addWriteStatuses(hudiWriteResult.getWriteStatuses()); + committableBuilder.addCompactionWriteStatuses( + hudiWriteResult.getCompactionWriteStatuses()); + } + return committableBuilder.build(); + } + + @Override + public LakeCommitResult commit( + HudiCommittable committable, Map<String, String> snapshotProperties) + throws IOException { + Map<String, List<WriteStatus>> writeStatuses = committable.getWriteStatuses(); + if (writeStatuses.size() != 1) { + throw new IOException( + "Hudi write statuses must contain exactly one instant, but got " + + writeStatuses.keySet() + + "."); + } Review Comment: `HudiCommittable` carries both regular and compaction write statuses, but `commit()` currently ignores `getCompactionWriteStatuses()` entirely. That makes it easy to silently drop compaction work (or leave compaction instants unhandled) if they ever get produced. If compaction is not supported yet, it’s safer to fail fast when compaction statuses are present. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeCommitter.java: ########## @@ -0,0 +1,214 @@ +/* + * 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.fluss.lake.hudi.tiering; + +import org.apache.fluss.lake.committer.CommittedLakeSnapshot; +import org.apache.fluss.lake.committer.LakeCommitResult; +import org.apache.fluss.lake.committer.LakeCommitter; +import org.apache.fluss.lake.hudi.utils.meta.CkpMetadata; +import org.apache.fluss.lake.hudi.utils.meta.CkpMetadataProvider; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.utils.IOUtils; + +import org.apache.hudi.client.HoodieFlinkWriteClient; +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.exception.HoodieException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static org.apache.fluss.lake.writer.LakeTieringFactory.FLUSS_LAKE_TIERING_COMMIT_USER; + +/** Hudi implementation of {@link LakeCommitter}. */ +public class HudiLakeCommitter implements LakeCommitter<HudiWriteResult, HudiCommittable> { + + private static final Logger LOG = LoggerFactory.getLogger(HudiLakeCommitter.class); + + private static final String COMMITTER_USER = "commit-user"; + + private final HudiWriteTableInfo hudiTableInfo; + private final HoodieFlinkWriteClient<?> writeClient; + private final CkpMetadata ckpMetadata; + + public HudiLakeCommitter( + HudiCatalogProvider hudiCatalogProvider, + CkpMetadataProvider ckpMetadataProvider, + TablePath tablePath) + throws IOException { + this.hudiTableInfo = HudiWriteTableInfo.create(hudiCatalogProvider, tablePath); + this.writeClient = hudiTableInfo.getWriteClient(); + this.ckpMetadata = ckpMetadataProvider.get(tablePath, hudiTableInfo); + LOG.info( + "Created HudiLakeCommitter with configuration {}.", hudiTableInfo.getFlinkConfig()); + } + + @Override + public HudiCommittable toCommittable(List<HudiWriteResult> hudiWriteResults) { + HudiCommittable.Builder committableBuilder = HudiCommittable.builder(); + for (HudiWriteResult hudiWriteResult : hudiWriteResults) { + committableBuilder.addWriteStatuses(hudiWriteResult.getWriteStatuses()); + committableBuilder.addCompactionWriteStatuses( + hudiWriteResult.getCompactionWriteStatuses()); + } + return committableBuilder.build(); + } + + @Override + public LakeCommitResult commit( + HudiCommittable committable, Map<String, String> snapshotProperties) + throws IOException { + Map<String, List<WriteStatus>> writeStatuses = committable.getWriteStatuses(); + if (writeStatuses.size() != 1) { + throw new IOException( + "Hudi write statuses must contain exactly one instant, but got " + + writeStatuses.keySet() + + "."); + } + + Map.Entry<String, List<WriteStatus>> entry = writeStatuses.entrySet().iterator().next(); + String instant = entry.getKey(); + List<WriteStatus> statuses = entry.getValue(); + + Map<String, String> commitMetadata = new HashMap<>(snapshotProperties); + commitMetadata.put(COMMITTER_USER, FLUSS_LAKE_TIERING_COMMIT_USER); + + try { + validateWriteStatuses(instant, statuses); + if (writeClient.getHeartbeatClient().isHeartbeatExpired(instant)) { + writeClient.getHeartbeatClient().start(instant); + } + + LOG.info( + "Committing Hudi instant {} with {} write status entries and metadata {}.", + instant, + statuses.size(), + commitMetadata); + boolean committed = writeClient.commit(instant, statuses, Option.of(commitMetadata)); + if (!committed) { + ckpMetadata.abortInstant(instant); + throw new IOException("Failed to commit Hudi instant " + instant + "."); + } + + ckpMetadata.commitInstant(instant); + LOG.info("Committed Hudi instant {} successfully.", instant); + return LakeCommitResult.committedIsReadable(Long.parseLong(instant)); + } catch (Exception e) { + if (e instanceof IOException) { + throw (IOException) e; + } + throw new IOException("Failed to commit Hudi instant " + instant + ".", e); + } + } + + @Override + public void abort(HudiCommittable committable) throws IOException { + Map<String, List<WriteStatus>> writeStatuses = committable.getWriteStatuses(); + for (String instant : writeStatuses.keySet()) { + try { + writeClient.rollback(instant); + ckpMetadata.abortInstant(instant); + LOG.info("Aborted Hudi instant {}.", instant); + } catch (Exception e) { + throw new IOException("Failed to abort Hudi instant " + instant + ".", e); + } + } + } Review Comment: `abort()` rolls back only instants from `getWriteStatuses()`, but ignores `getCompactionWriteStatuses()`. If compaction instants are present in the committable, they will be left behind (no rollback, no checkpoint metadata update). -- 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]
