hudi-agent commented on code in PR #20013:
URL: https://github.com/apache/hudi/pull/20013#discussion_r4058608886
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -169,8 +169,16 @@ public List<WriteStatus> compact(HoodieWriteConfig
writeConfig,
TaskContextSupplier taskContextSupplier)
throws IOException {
HoodieMergeHandle<T, ?, ?, ?> mergeHandle =
HoodieMergeHandleFactory.create(writeConfig,
instantTime, table, operation, taskContextSupplier,
hoodieReaderContext, maxInstantTime, getEngineRecordType());
- mergeHandle.doMerge();
- return mergeHandle.close();
+ try {
+ mergeHandle.doMerge();
+ return mergeHandle.close();
+ } catch (Exception e) {
+ // doMerge() failed before the handle (and its underlying file
writer/output stream) was
+ // closed. Close it here on a best-effort basis so the writer's stream
isn't leaked, then
+ // surface the original failure.
+ closeQuietlyOnFailure(mergeHandle::close, e);
Review Comment:
🤖 For the compaction handles this close will nearly always throw:
`FileGroupReaderBasedMergeHandle.close()` dereferences `readStats`, which
`doMerge()` only assigns at the very end, so a failure mid-merge leads to an
NPE right after `super.close()` (the stream is released, but every compaction
failure now carries a `HoodieUpsertException(NPE)` as a suppressed exception).
Same for the inline/native log append handles, where `readStats` and even
`writeStatus.getStat()` (set lazily in `init()` on the first record) can be
null. Would it be worth null-guarding those stat updates in the handles'
`close()` so the suppressed exception only shows up when the close itself
genuinely fails?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java:
##########
@@ -162,6 +168,19 @@ public void runMerge(HoodieTable<?, ?, ?, ?> table,
if (executor != null) {
executor.shutdownNow();
executor.awaitTermination();
+ if (!mergeSucceeded) {
+ // executor.execute() failed before the consumer's finish() (which
closes mergeHandle)
+ // could run - e.g. SimpleExecutor.shutdownNow() only closes the
producer iterator and
+ // never touches the consumer/mergeHandle. Close it here on a
best-effort basis so its
+ // underlying file writer/output stream isn't leaked. Any failure
here is logged rather
+ // than thrown, so it doesn't mask the original failure already
propagating out of the
+ // catch block above.
+ try {
+ mergeHandle.close();
Review Comment:
🤖 `HoodieWriteMergeHandle.close()` is a full commit-style flush — it runs
`writeIncomingRecords()` (writing every not-yet-merged incoming record into a
file that's about to be discarded) and `performMergeDataValidationCheck()`
around `fileWriter.close()`. With `hoodie.write.ignore.failed=false`, a broken
writer makes `writeInsertRecord` rethrow inside `writeIncomingRecords()` before
`fileWriter.close()` is ever reached, so the stream still leaks in exactly the
transient-I/O scenario motivating this PR. Have you considered a narrower abort
path on the handle that just closes the file writer (and CDC logger) without
flushing more data?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -169,8 +169,16 @@ public List<WriteStatus> compact(HoodieWriteConfig
writeConfig,
TaskContextSupplier taskContextSupplier)
throws IOException {
HoodieMergeHandle<T, ?, ?, ?> mergeHandle =
HoodieMergeHandleFactory.create(writeConfig,
instantTime, table, operation, taskContextSupplier,
hoodieReaderContext, maxInstantTime, getEngineRecordType());
- mergeHandle.doMerge();
- return mergeHandle.close();
+ try {
Review Comment:
🤖 The regular upsert path has the same shape and isn't covered here:
`MergeUtils.runMerge()` does `mergeHandle.doMerge(); ... mergeHandle.close()`
with no failure-path close, and in 1.x that also drives
`FileGroupReaderBasedMergeHandle` (whose `doMerge()` doesn't go through
`HoodieMergeHelper`), so a failed upsert merge leaks the writer stream just
like a failed compaction. Would it make sense to apply the same try/catch there
(or centralize the pattern) so the fix covers both write paths?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/commit/TestHoodieMergeHelper.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * 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.hudi.table.action.commit;
+
+import org.apache.hudi.common.avro.VariantSchemaUtils;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordMerger;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaCompatibility;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.common.util.queue.HoodieExecutor;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.core.io.storage.HoodieFileReader;
+import org.apache.hudi.core.io.storage.HoodieFileReaderFactory;
+import org.apache.hudi.core.io.storage.HoodieIOFactory;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.execution.ExecutorFactory;
+import org.apache.hudi.io.HoodieWriteMergeHandle;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StorageConfiguration;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.table.HoodieTable;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import java.io.IOException;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
Review Comment:
🤖 `assertArrayEquals` looks unused in this file — checkstyle runs on test
sources with `UnusedImports` enabled, so I'd expect this to fail the build.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
