hudi-agent commented on code in PR #20013:
URL: https://github.com/apache/hudi/pull/20013#discussion_r4064507168
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -182,8 +190,29 @@ public List<WriteStatus> logCompact(HoodieWriteConfig
writeConfig,
HoodieAppendHandle<T, ?, ?, ?> appendHandle =
CommonClientUtils.shouldWriteNativeLogs(writeConfig)
? new FileGroupReaderBasedNativeLogAppendHandle<>(writeConfig,
instantTime, table, operation, taskContextSupplier, readerContext)
: new FileGroupReaderBasedInlineLogAppendHandle<>(writeConfig,
instantTime, table, operation, taskContextSupplier, readerContext);
- appendHandle.doAppend();
- return appendHandle.close();
+ try {
+ appendHandle.doAppend();
+ return appendHandle.close();
+ } catch (Exception e) {
+ // doAppend() failed before the handle (and its underlying log
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(appendHandle::close, e);
Review Comment:
🤖 Does this actually reach the stream close in the motivating case?
`HoodieAppendHandle.close()` does `markClosed(); flushAppend(); ...;
closeLogWriter()` — if `doAppend()` failed because the log writer itself is
broken (the GCS resumable-upload case), `flushAppend()` will throw again before
`closeLogWriter()` runs, and since the handle is now marked closed, any later
`close()` is a no-op. Same shape in `HoodieWriteMergeHandle.close()`
(`writeIncomingRecords()` runs before `fileWriter.close()`). It might be worth
an abort-style close that skips the flush and just releases the writer, or
wrapping the flush in try/finally that closes the writer regardless.
<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:
🤖 Agree this is worth covering in the same PR. Concretely,
`MergeUtils.runMerge()` is the upsert entry point from
`BaseSparkCommitActionExecutor:444`, `BaseFlinkCommitActionExecutor:197`,
`BaseJavaCommitActionExecutor:246` and the three CoW table `handleUpdate`
overloads. For the legacy `HoodieWriteMergeHandle`, `doMerge()` delegates to
`HoodieMergeHelper.runMerge()` so the fix here applies, but
`FileGroupReaderBasedMergeHandle.doMerge()` overrides it and never goes through
the helper, so that path still leaks.
##########
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:
🤖 I don't think this one applies — `assertArrayEquals` isn't imported in
`TestHoodieMergeHelper` at all; it's only in `TestHoodieCompactor`, where it's
used twice (the suppressed-exception assertions). I checked both new test files
and every import is referenced, so checkstyle `UnusedImports` should be fine.
##########
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:
🤖 There's a second wrinkle here: `HoodieWriteMergeHandle.close()` calls
`markClosed()` before `writeIncomingRecords()`. So if the consumer's `finish()`
-> `close()` fails partway (which is itself an `execute()` failure), the
best-effort `mergeHandle.close()` added in the `finally` returns early on
`isClosed()` and never reaches `fileWriter.close()`. A dedicated abort path
that just closes the writer regardless of the closed flag would cover both
cases.
##########
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:
🤖 Confirmed from the code: `FileGroupReaderBasedMergeHandle.close()` runs
`super.close()` first (so the writer stream does get released) and then hits
`readStats.getTotalLogReadTimeMs()` at L375, while `readStats` is only assigned
at L330 after the record loop finishes — so any mid-merge failure will surface
a wrapped NPE as the suppressed exception. Worth noting the new
`TestHoodieCompactor` cases mock `HoodieMergeHandle` entirely, so they won't
exercise this; a test against the real handle would catch it.
--
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]
