XuQianJin-Stars commented on code in PR #3509: URL: https://github.com/apache/fluss/pull/3509#discussion_r3458455258
########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeCommitter.java: ########## @@ -0,0 +1,238 @@ +/* + * 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.io.UncheckedIOException; +import java.util.HashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +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 { + ensureNoCompactionWriteStatuses(committable); + + 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); Review Comment: restarting an expired heartbeat right before commit is unsafe ```java if (writeClient.getHeartbeatClient().isHeartbeatExpired(instant)) { writeClient.getHeartbeatClient().start(instant); } ``` In Hudi an expired heartbeat means the instant is considered abandoned by other writers; restarting it and immediately calling `commit()` is undefined in Hudi 1.x and on some versions Hudi will silently roll the instant back during commit. If you only want to tolerate a short heartbeat stall, use `HoodieHeartbeatClient#extendHeartbeat`; otherwise this fallback should be removed and the caller should retry the whole instant from scratch. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeCommitter.java: ########## @@ -0,0 +1,238 @@ +/* + * 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.io.UncheckedIOException; +import java.util.HashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +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 { + ensureNoCompactionWriteStatuses(committable); + + 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)); Review Comment: failed commit only aborts ckpMetadata, no `rollback`, and parses instant as `long` without protection ```java boolean committed = writeClient.commit(instant, statuses, Option.of(commitMetadata)); if (!committed) { ckpMetadata.abortInstant(instant); throw new IOException("Failed to commit Hudi instant " + instant + "."); } return LakeCommitResult.committedIsReadable(Long.parseLong(instant)); ``` Two issues: - When `commit` returns `false`, Hudi has already produced inflight files. Calling only `ckpMetadata.abortInstant(instant)` leaves the timeline / inflight / requested metadata behind. This is inconsistent with `abort()` which performs both `writeClient.rollback(instant)` and `ckpMetadata.abortInstant(instant)`. Reuse that path or at least call `writeClient.rollback(instant)` here. - `Long.parseLong(instant)` can throw `NumberFormatException` for non-numeric Hudi instant types (e.g. compaction instants `.compaction` suffixed). Wrap in `IOException` or assert numeric in `validateWriteStatuses`. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeCommitter.java: ########## @@ -0,0 +1,238 @@ +/* + * 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.io.UncheckedIOException; +import java.util.HashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +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 { + ensureNoCompactionWriteStatuses(committable); + + 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 { + Set<String> instants = new LinkedHashSet<>(committable.getWriteStatuses().keySet()); + instants.addAll(committable.getCompactionWriteStatuses().keySet()); + for (String instant : instants) { + 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); + } + } + } + + private static void ensureNoCompactionWriteStatuses(HudiCommittable committable) + throws IOException { + Map<String, List<WriteStatus>> compactionWriteStatuses = + committable.getCompactionWriteStatuses(); + if (!compactionWriteStatuses.isEmpty()) { + throw new IOException( + "Hudi compaction write statuses are not supported yet, but got instants " + + compactionWriteStatuses.keySet() + + "."); + } + } + + @Nullable + @Override + public CommittedLakeSnapshot getMissingLakeSnapshot(@Nullable Long latestLakeSnapshotIdOfFluss) + throws IOException { + HoodieTimeline latestLakeTimeline = + getCompletedTimelineCommittedBy(FLUSS_LAKE_TIERING_COMMIT_USER); + Optional<HoodieInstant> latestLakeInstant = + latestLakeTimeline.getReverseOrderedInstantsByCompletionTime().findFirst(); + if (!latestLakeInstant.isPresent()) { + return null; + } + + long latestLakeSnapshotId = Long.parseLong(latestLakeInstant.get().requestedTime()); Review Comment: `getMissingLakeSnapshot` uses `requestedTime()` which is not guaranteed to align with the value returned in `commit()` ```java // commit return LakeCommitResult.committedIsReadable(Long.parseLong(instant)); ``` ```java // getMissingLakeSnapshot Optional<HoodieInstant> latestLakeInstant = latestLakeTimeline.getReverseOrderedInstantsByCompletionTime().findFirst(); ... long latestLakeSnapshotId = Long.parseLong(latestLakeInstant.get().requestedTime()); ``` The snapshot id returned in `commit()` is the Hudi `instantTime` produced by the writer (via `writeClient.startCommit()`). In `getMissingLakeSnapshot()` you order by **completion time** but read `requestedTime()`. When two commits complete out of order (C1 requested first, completed last; C2 requested last, completed first), the "latest by completion time" is C2 — but its `requestedTime` is **not** monotonic with the snapshot id returned by `commit()`. As a result the `latestLakeSnapshotId <= latestLakeSnapshotIdOfFluss` check can miss a real missing snapshot. **Fix:** order by `requestedTime()` (i.e. `getReverseOrderedInstants()`), or compare on the same axis as `commit()`'s return value. ########## fluss-common/src/main/java/org/apache/fluss/row/decode/KeyDecoder.java: ########## @@ -60,6 +60,9 @@ static KeyDecoder ofPrimaryKeyDecoder( if (lakeFormat == null || lakeFormat == DataLakeFormat.LANCE) { return CompactedKeyDecoder.createKeyDecoder(rowType, keyFields); } + if (lakeFormat == DataLakeFormat.HUDI) { + return CompactedKeyDecoder.createKeyDecoder(rowType, keyFields); Review Comment: asymmetric placement of the HUDI branch hurts readability and is bug-prone Encoder: ```java // KeyEncoder.java L85-L88 (short-circuits at the top) if (dataLakeFormat == DataLakeFormat.HUDI) { return CompactedKeyEncoder.createKeyEncoder(rowType, keyFields); } ``` Decoder: ```java // KeyDecoder.java L60-L65 (nested inside the v1 / v2-defaultBucketKey branch) if (lakeFormat == null || lakeFormat == DataLakeFormat.LANCE) { return CompactedKeyDecoder.createKeyDecoder(rowType, keyFields); } if (lakeFormat == DataLakeFormat.HUDI) { return CompactedKeyDecoder.createKeyDecoder(rowType, keyFields); } ``` On the only combination that matters (`kvFormatVersion == 1 && !isDefaultBucketKey && lakeFormat == HUDI`) both sides happen to agree today, but the asymmetry will make future changes (e.g. introducing a HudiKeyDecoder for lossy encoding) silently diverge encoder and decoder. Please: - Either move the HUDI short-circuit to the **same** position on both sides. - Or remove the short-circuit on `KeyEncoder` and let the explicit decision tree handle it. Add a comment on both sides explaining "Hudi's lake-format encoder is lossy, therefore the PK encoder must always use `CompactedKeyEncoder` regardless of `isDefaultBucketKey`". Also: L60 and L63 in `KeyDecoder` have identical bodies — collapse them: ```java if (lakeFormat == null || lakeFormat == DataLakeFormat.LANCE || lakeFormat == DataLakeFormat.HUDI) { return CompactedKeyDecoder.createKeyDecoder(rowType, keyFields); } ``` ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeCommitter.java: ########## @@ -0,0 +1,238 @@ +/* + * 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.io.UncheckedIOException; +import java.util.HashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +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 { + ensureNoCompactionWriteStatuses(committable); + + 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 { + Set<String> instants = new LinkedHashSet<>(committable.getWriteStatuses().keySet()); + instants.addAll(committable.getCompactionWriteStatuses().keySet()); + for (String instant : instants) { + 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: A single failing instant prevents later instants from being rolled back. Collect exceptions (`use Throwable#addSuppressed)` and throw a combined IOException after the loop. Cleanup of "good" instants should never depend on the success of "bad" ones. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiCommittableSerializer.java: ########## @@ -0,0 +1,106 @@ +/* + * 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.serializer.SimpleVersionedSerializer; +import org.apache.fluss.utils.InstantiationUtils; + +import org.apache.hudi.client.WriteStatus; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.util.List; +import java.util.Map; + +/** Serializer for {@link HudiCommittable}. */ +public class HudiCommittableSerializer implements SimpleVersionedSerializer<HudiCommittable> { + + private static final int CURRENT_VERSION = 1; + + @Override + public int getVersion() { + return CURRENT_VERSION; + } + + @Override + public byte[] serialize(HudiCommittable hudiCommittable) throws IOException { + byte[] writeStatusesBytes = + InstantiationUtils.serializeObject(hudiCommittable.getWriteStatuses()); + byte[] compactionWriteStatusesBytes = + InstantiationUtils.serializeObject(hudiCommittable.getCompactionWriteStatuses()); Review Comment: serializing `WriteStatus` via JDK Java serialization is not safe across Hudi versions `WriteStatus` and its transitive types (`HoodieRecordLocation`, `HoodieWriteStat`, …) are `Serializable` but Hudi does not contract `serialVersionUID` stability across versions. After even a Hudi patch upgrade, Fluss checkpoints recovered from the old format may fail with `InvalidClassException`. Compare with `IcebergLakeCommitter` / `PaimonLakeCommitter` which serialize the **fields they actually need** rather than Hudi/Paimon/Iceberg internal classes. **Recommendation:** extract the minimal fields needed for `commit()` (fileId, partitionPath, totalErrorRecords, the bytes of `HoodieWriteStat` written by Hudi's own JSON `HoodieCommitMetadata`, etc.) and serialize them explicitly. This also keeps the committable size predictable. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeCommitter.java: ########## @@ -0,0 +1,238 @@ +/* + * 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.io.UncheckedIOException; +import java.util.HashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +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 { + ensureNoCompactionWriteStatuses(committable); + + 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 { + Set<String> instants = new LinkedHashSet<>(committable.getWriteStatuses().keySet()); + instants.addAll(committable.getCompactionWriteStatuses().keySet()); + for (String instant : instants) { + 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); + } + } + } + + private static void ensureNoCompactionWriteStatuses(HudiCommittable committable) + throws IOException { + Map<String, List<WriteStatus>> compactionWriteStatuses = + committable.getCompactionWriteStatuses(); + if (!compactionWriteStatuses.isEmpty()) { + throw new IOException( + "Hudi compaction write statuses are not supported yet, but got instants " + + compactionWriteStatuses.keySet() + + "."); + } + } + + @Nullable + @Override + public CommittedLakeSnapshot getMissingLakeSnapshot(@Nullable Long latestLakeSnapshotIdOfFluss) + throws IOException { + HoodieTimeline latestLakeTimeline = + getCompletedTimelineCommittedBy(FLUSS_LAKE_TIERING_COMMIT_USER); + Optional<HoodieInstant> latestLakeInstant = + latestLakeTimeline.getReverseOrderedInstantsByCompletionTime().findFirst(); + if (!latestLakeInstant.isPresent()) { + return null; + } + + long latestLakeSnapshotId = Long.parseLong(latestLakeInstant.get().requestedTime()); + if (latestLakeSnapshotIdOfFluss != null + && latestLakeSnapshotId <= latestLakeSnapshotIdOfFluss) { + return null; + } + + HoodieCommitMetadata metadata = + latestLakeTimeline.readCommitMetadata(latestLakeInstant.get()); + Map<String, String> extraMetadata = metadata.getExtraMetadata(); + if (extraMetadata == null) { + throw new IOException("Failed to load committed Hudi instant extra metadata."); + } + return new CommittedLakeSnapshot(latestLakeSnapshotId, extraMetadata); + } + + @Override + public void close() throws Exception { + IOUtils.closeQuietly(ckpMetadata, "hudi checkpoint metadata"); + IOUtils.closeQuietly(hudiTableInfo, "hudi table info"); Review Comment: does not close the write client，`HoodieFlinkWriteClient` is `AutoCloseable` and owns a heartbeat thread pool, executor, embedded timeline server and FS view. If `hudiTableInfo.close()` does not transitively close `writeClient`, this is a real resource leak (visible as orphan threads in long-running TM JVMs). Please either close `writeClient` explicitly here, or document the ownership chain in `HudiWriteTableInfo`. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeCommitter.java: ########## @@ -0,0 +1,238 @@ +/* + * 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.io.UncheckedIOException; +import java.util.HashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +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 { + ensureNoCompactionWriteStatuses(committable); + + 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 { + Set<String> instants = new LinkedHashSet<>(committable.getWriteStatuses().keySet()); + instants.addAll(committable.getCompactionWriteStatuses().keySet()); + for (String instant : instants) { + 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); + } + } + } + + private static void ensureNoCompactionWriteStatuses(HudiCommittable committable) + throws IOException { + Map<String, List<WriteStatus>> compactionWriteStatuses = + committable.getCompactionWriteStatuses(); + if (!compactionWriteStatuses.isEmpty()) { + throw new IOException( + "Hudi compaction write statuses are not supported yet, but got instants " + + compactionWriteStatuses.keySet() + + "."); + } + } + + @Nullable + @Override + public CommittedLakeSnapshot getMissingLakeSnapshot(@Nullable Long latestLakeSnapshotIdOfFluss) + throws IOException { + HoodieTimeline latestLakeTimeline = + getCompletedTimelineCommittedBy(FLUSS_LAKE_TIERING_COMMIT_USER); + Optional<HoodieInstant> latestLakeInstant = + latestLakeTimeline.getReverseOrderedInstantsByCompletionTime().findFirst(); + if (!latestLakeInstant.isPresent()) { + return null; + } + + long latestLakeSnapshotId = Long.parseLong(latestLakeInstant.get().requestedTime()); + if (latestLakeSnapshotIdOfFluss != null + && latestLakeSnapshotId <= latestLakeSnapshotIdOfFluss) { + return null; + } + + HoodieCommitMetadata metadata = Review Comment: Add a null check on metadata itself (Hudi 1.x can return a sentinel empty `HoodieCommitMetadata` on some failure paths, but a future version may not), or wrap the call in `try/catch `with `IOException`. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiCommittable.java: ########## @@ -0,0 +1,136 @@ +/* + * 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.hudi.client.WriteStatus; + +import javax.annotation.Nullable; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +/** The committable aggregated from Hudi write results for one tiering round. */ +public class HudiCommittable implements Serializable { + + private static final long serialVersionUID = 1L; + + private final Map<String, List<WriteStatus>> writeStatuses; + private final Map<String, List<WriteStatus>> compactionWriteStatuses; + + public HudiCommittable( + Map<String, List<WriteStatus>> writeStatuses, + @Nullable Map<String, List<WriteStatus>> compactionWriteStatuses) { + this.writeStatuses = copyWriteStatuses(writeStatuses); + this.compactionWriteStatuses = copyWriteStatuses(compactionWriteStatuses); + } + + public Map<String, List<WriteStatus>> getWriteStatuses() { + return writeStatuses; + } + + public Map<String, List<WriteStatus>> getCompactionWriteStatuses() { + return compactionWriteStatuses; + } + + public static Builder builder() { + return new Builder(); + } + + private static Map<String, List<WriteStatus>> copyWriteStatuses( + @Nullable Map<String, List<WriteStatus>> statusesByInstant) { + if (statusesByInstant == null || statusesByInstant.isEmpty()) { + return Collections.emptyMap(); + } + + Map<String, List<WriteStatus>> copiedStatuses = new HashMap<>(); + for (Map.Entry<String, List<WriteStatus>> entry : statusesByInstant.entrySet()) { + copiedStatuses.put( + entry.getKey(), + Collections.unmodifiableList(new ArrayList<>(entry.getValue()))); Review Comment: `WriteStatus` lists in `Hudi` can be large (one entry per file group). Copying them twice (once in the Builder, once here) doubles peak heap during checkpoint. Since the Builder is private and not exposed, wrap with `Collections.unmodifiableMap(...)` without the deep copy. -- 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]
