Copilot commented on code in PR #3507: URL: https://github.com/apache/fluss/pull/3507#discussion_r3450089398
########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResultSerializer.java: ########## @@ -0,0 +1,100 @@ +/* + * 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 HudiWriteResult}. */ +public class HudiWriteResultSerializer implements SimpleVersionedSerializer<HudiWriteResult> { + + private static final int CURRENT_VERSION = 1; + + @Override + public int getVersion() { + return CURRENT_VERSION; + } + + @Override + public byte[] serialize(HudiWriteResult hudiWriteResult) throws IOException { + byte[] writeResultBytes = + InstantiationUtils.serializeObject(hudiWriteResult.getWriteStatuses()); + byte[] compactionWriteResultBytes = + InstantiationUtils.serializeObject(hudiWriteResult.getCompactionWriteStatuses()); + + try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); + DataOutputStream dos = new DataOutputStream(baos)) { + dos.writeInt(writeResultBytes.length); + dos.write(writeResultBytes); + dos.writeInt(compactionWriteResultBytes.length); + dos.write(compactionWriteResultBytes); + return baos.toByteArray(); + } + } + + @Override + public HudiWriteResult deserialize(int version, byte[] serialized) throws IOException { + if (version != CURRENT_VERSION) { + throw new IOException( + "Unsupported HudiWriteResult version " + + version + + ", expected " + + CURRENT_VERSION); + } + + Map<String, List<WriteStatus>> writeResult; + Map<String, List<WriteStatus>> compactionWriteResult; + try (DataInputStream dis = new DataInputStream(new ByteArrayInputStream(serialized))) { + int writeResultLength = dis.readInt(); + validateLength(writeResultLength, serialized.length, "WriteResult"); + byte[] writeResultBytes = new byte[writeResultLength]; + dis.readFully(writeResultBytes); + writeResult = + InstantiationUtils.deserializeObject( + writeResultBytes, getClass().getClassLoader()); + + int compactionWriteResultLength = dis.readInt(); + validateLength(compactionWriteResultLength, serialized.length, "CompactionWriteResult"); + byte[] compactionWriteResultBytes = new byte[compactionWriteResultLength]; + dis.readFully(compactionWriteResultBytes); + compactionWriteResult = + InstantiationUtils.deserializeObject( + compactionWriteResultBytes, getClass().getClassLoader()); + } catch (ClassNotFoundException e) { + throw new IOException("Couldn't deserialize HudiWriteResult.", e); + } Review Comment: `validateLength(..., serialized.length, ...)` validates each field length against the *total* byte array size, not the remaining bytes in the stream. This allows corrupted payloads (e.g., first length consumes most bytes, second length still <= total) to pass validation and then fail with EOF during `readFully`, producing a less actionable error and weaker corruption checks. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/meta/CkpMetadata.java: ########## @@ -0,0 +1,211 @@ +/* + * 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.utils.meta; + +import org.apache.fluss.annotation.VisibleForTesting; + +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.storage.StoragePath; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** DFS-backed checkpoint metadata used as a lightweight message bus for Hudi instants. */ +public class CkpMetadata implements Serializable, AutoCloseable { + + private static final long serialVersionUID = 1L; + private static final Logger LOG = LoggerFactory.getLogger(CkpMetadata.class); + + private static final int MAX_RETAIN_CKP_NUM = 1; + private static final String CKP_META = "ckp_meta"; + + private final FileSystem fs; + private final Path path; + + private List<CkpMessage> messages; + private List<String> instantCache; + + CkpMetadata(FileSystem fs, String basePath, String uniqueId) { + this.fs = fs; + this.path = new Path(ckpMetaPath(basePath, uniqueId)); + } + + public void bootstrap() throws IOException { + if (!fs.exists(path)) { + fs.mkdirs(path); + } + } + + public void startInstant(String instant) { + Path instantPath = fullPath(CkpMessage.getFileName(instant, CkpMessage.State.INFLIGHT)); + try { + fs.createNewFile(instantPath); + } catch (IOException e) { + throw new HoodieException( + "Exception while adding checkpoint start metadata for instant: " + instant, e); + } + cache(instant); + clean(); + } + + public void commitInstant(String instant) { + Path instantPath = fullPath(CkpMessage.getFileName(instant, CkpMessage.State.COMPLETED)); + try { + fs.createNewFile(instantPath); + } catch (IOException e) { + throw new HoodieException( + "Exception while adding checkpoint commit metadata for instant: " + instant, e); + } + } + + public void abortInstant(String instant) { + Path instantPath = fullPath(CkpMessage.getFileName(instant, CkpMessage.State.ABORTED)); + try { + fs.createNewFile(instantPath); + } catch (IOException e) { + throw new HoodieException( + "Exception while adding checkpoint abort metadata for instant: " + instant, e); + } + } + + @Nullable + public String lastPendingInstant() { + load(); + if (!messages.isEmpty()) { + CkpMessage ckpMsg = messages.get(messages.size() - 1); + if (!ckpMsg.isComplete()) { + return ckpMsg.getInstant(); + } + } + return null; + } Review Comment: `lastPendingInstant()` treats an ABORTED instant as "pending" (because `isComplete()` only checks COMPLETED). If an instant gets aborted, writers waiting on `lastPendingInstant()` may proceed with an aborted instant and write into a failed timeline. Pending should only mean INFLIGHT here. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/RecordWriteBuffer.java: ########## @@ -0,0 +1,368 @@ +/* + * 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.writer; + +import org.apache.fluss.lake.hudi.tiering.HudiWriteTableInfo; +import org.apache.fluss.lake.hudi.utils.meta.CkpMetadata; +import org.apache.fluss.lake.writer.WriterInitContext; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.binary.BinaryRowData; +import org.apache.flink.table.runtime.operators.sort.BinaryInMemorySortBuffer; +import org.apache.flink.util.MutableObjectIterator; +import org.apache.hudi.client.HoodieFlinkWriteClient; +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.client.model.HoodieFlinkInternalRow; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.common.model.WriteOperationType; +import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator; +import org.apache.hudi.common.util.ValidationUtils; +import org.apache.hudi.common.util.collection.MappingIterator; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.sink.buffer.MemorySegmentPoolFactory; +import org.apache.hudi.sink.bulk.RowDataKeyGen; +import org.apache.hudi.sink.exception.MemoryPagesExhaustedException; +import org.apache.hudi.sink.utils.BufferUtils; +import org.apache.hudi.sink.utils.TimeWait; +import org.apache.hudi.table.action.commit.BucketInfo; +import org.apache.hudi.table.action.commit.BucketType; +import org.apache.hudi.util.MutableIteratorWrapperIterator; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Date; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Random; + +/** Buffers records and writes them to Hudi in batches. */ +public class RecordWriteBuffer { + + private static final Logger LOG = LoggerFactory.getLogger(RecordWriteBuffer.class); + + private DataBucket bucket; + + private final TotalSizeTracer tracer; + private final HoodieFlinkWriteClient writeClient; + private final HudiWriteTableInfo hudiTableInfo; + private final Configuration config; + private final CkpMetadata ckpMetadata; + private final Map<String, List<WriteStatus>> writeStatuses; + private final HudiRecordConverter recordConverter; + private final long tieringRoundTimestamp; + + public RecordWriteBuffer( + HudiWriteTableInfo hudiTableInfo, CkpMetadata ckpMetadata, long tieringRoundTimestamp) { + this.hudiTableInfo = hudiTableInfo; + this.config = hudiTableInfo.getFlinkConfig(); + this.tracer = new TotalSizeTracer(config); + this.writeClient = hudiTableInfo.getWriteClient(); + this.ckpMetadata = ckpMetadata; + this.writeStatuses = new HashMap<>(); + this.tieringRoundTimestamp = tieringRoundTimestamp; + this.recordConverter = + HudiRecordConverter.getInstance( + RowDataKeyGen.instance( + hudiTableInfo.getFlinkConfig(), hudiTableInfo.getRowType())); + } + + public void bufferRecord(HoodieFlinkInternalRow internalRow) throws IOException { + boolean success = doBufferRecord(internalRow); + if (!success) { + flushAndDisposeBucket(); + success = doBufferRecord(internalRow); + if (!success) { + throw new IOException("Hudi write buffer is too small to hold a single record."); + } + } + + if (bucket != null) { + boolean totalSizeExceedsThreshold = tracer.trace(bucket.getLastRecordSize()); + if (bucket.isFull() || totalSizeExceedsThreshold) { + flushAndDisposeBucket(); + } + } + } + + public void flushRemaining() { + if (bucket == null) { + return; + } + + String instantTime = getLastPendingInstant(); + if (!bucket.isEmpty()) { + LOG.info( + "Flushing remaining Hudi records for instant {}, size {}.", + instantTime, + bucket.getBufferSize()); + List<WriteStatus> writeStatus = + writeStatuses.getOrDefault(instantTime, new ArrayList<>()); + writeStatus.addAll( + writeRecords(instantTime, config.getString(FlinkOptions.OPERATION), bucket)); + writeStatuses.put(instantTime, writeStatus); + bucket.dispose(); + bucket = null; + } + + tracer.reset(); + writeClient.cleanHandles(); + } + + public Map<String, List<WriteStatus>> getWriteStatuses() { + return writeStatuses; + } + + private boolean doBufferRecord(HoodieFlinkInternalRow internalRow) throws IOException { + try { + if (bucket == null) { + bucket = + new DataBucket( + BufferUtils.createBuffer( + hudiTableInfo.getRowType(), + MemorySegmentPoolFactory.createMemorySegmentPool(config)), + getBucketInfo(internalRow), + config.getDouble(FlinkOptions.WRITE_BATCH_SIZE)); + LOG.debug("Initialized a new Hudi data bucket."); + } + return bucket.writeRow(internalRow.getRowData()); + } catch (MemoryPagesExhaustedException e) { + LOG.warn("Hudi write buffer memory pages are exhausted, flushing current bucket.", e); + return false; + } + } Review Comment: `RecordWriteBuffer` keeps a single `DataBucket`, but `BucketInfo` (partition path / fileId / bucket type) can vary per record. Without flushing/switching buckets on `BucketInfo` changes, records can be converted/written using a stale `BucketInfo`, routing them to the wrong Hudi bucket/file/partition. Also, if `MemoryPagesExhaustedException` happens on the first write into a newly created (still empty) bucket, the subsequent `flushAndDisposeBucket()` will fail because `flushBucket()` rejects empty buckets. -- 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]
