danny0405 commented on code in PR #18776:
URL: https://github.com/apache/hudi/pull/18776#discussion_r4059110069
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java:
##########
@@ -531,6 +532,7 @@ protected void doWrite(HoodieRecord record, HoodieSchema
schema, TypedProperties
} catch (Throwable t) {
log.error("Error writing record " + record, t);
if (!config.getIgnoreWriteFailed()) {
+ closeLogWriterQuietly(t);
Review Comment:
Fixed by marking the append handle closed on failure before releasing the
writer. A later `close()` returns without flushing the buffered records again.
`testFailedFlushClosesWriterAndPreventsAnotherFlush` covers this with buffered
records and with both successful and failing writer cleanup.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieBinaryCopyHandle.java:
##########
@@ -118,14 +118,26 @@ public void write() {
log.info("Schema evolution enabled for binary copy: {}",
schemaEvolutionEnabled);
records = this.writer.binaryCopy(inputFiles,
Collections.singletonList(path), writeScheMessageType, schemaEvolutionEnabled);
} catch (IOException e) {
+ closeWriterAfterFailure(e);
throw new HoodieIOException(e.getMessage(), e);
+ } catch (RuntimeException e) {
+ closeWriterAfterFailure(e);
+ throw e;
} finally {
this.recordsWritten = records;
this.insertRecordsWritten = records;
}
log.info("Finish rewriting " + this.path + ". Using " + timer.endTimer() +
" mills");
}
+ private void closeWriterAfterFailure(Throwable failure) {
+ try {
+ this.writer.close();
Review Comment:
Updated to `CloseableUtils.closeSuppressing(writer::close, failure)`, using
the existing common utility after the rebase. It preserves the original failure
and suppresses close failures, including `RuntimeException`. The method
reference is needed because `HoodieFileBinaryCopier` itself does not implement
`AutoCloseable`.
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieAvroHFileWriter.java:
##########
@@ -105,8 +106,14 @@ public HoodieAvroHFileWriter(String instantTime,
StoragePath file, HoodieHFileCo
.build();
StorageConfiguration<Configuration> storageConf = new
HadoopStorageConfiguration(conf);
StoragePath filePath = new StoragePath(this.file.toUri());
- OutputStream outputStream = HoodieStorageUtils.getStorage(filePath,
storageConf).create(filePath);
- this.writer = new HFileWriterImpl(context, outputStream);
+ OutputStream outputStream = HoodieStorageUtils.getStorage(filePath,
storageConf).create(filePath);
+ try {
+ this.writer = new HFileWriterImpl(context, outputStream);
+ } finally {
+ if (this.writer == null) {
+ closeQuietly(outputStream);
Review Comment:
Fixed: the protected constructor region now includes `appendFileInfo`. On
failure it closes the initialized writer, or the output stream if writer
construction did not complete, and suppresses any cleanup failure onto the
original exception. Added `testConstructorClosesWriterWhenFileInfoFails`,
covering both successful cleanup and a failing close.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieBinaryCopyHandle.java:
##########
@@ -118,14 +118,26 @@ public void write() {
log.info("Schema evolution enabled for binary copy: {}",
schemaEvolutionEnabled);
records = this.writer.binaryCopy(inputFiles,
Collections.singletonList(path), writeScheMessageType, schemaEvolutionEnabled);
} catch (IOException e) {
+ closeWriterAfterFailure(e);
throw new HoodieIOException(e.getMessage(), e);
+ } catch (RuntimeException e) {
+ closeWriterAfterFailure(e);
+ throw e;
} finally {
this.recordsWritten = records;
this.insertRecordsWritten = records;
}
log.info("Finish rewriting " + this.path + ". Using " + timer.endTimer() +
" mills");
}
+ private void closeWriterAfterFailure(Throwable failure) {
+ try {
Review Comment:
Done. Binary-copy cleanup now shares `CloseableUtils.closeSuppressing` with
the other handles. The duplicate `AutoCloseableUtils` class has been removed in
favor of the utility already present on upstream master.
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/parquet/io/HoodieParquetBinaryCopyBase.java:
##########
@@ -147,17 +154,42 @@ protected void initFileWriter(Path outPutFile,
CompressionCodecName newCodecName
@Override
public void close() throws IOException {
+ if (writer == null) {
+ return;
+ }
Map<String, String> extraMetaData = finalizeMetadata();
extraMetaData = extraMetaData == null ? new HashMap<>() : extraMetaData;
extraMetaData.remove("parquet.avro.schema");
extraMetaData.remove("org.apache.spark.sql.parquet.row.metadata");
- writer.end(extraMetaData);
- // Release the buffer
- reusableBlockBuffer = null;
+ try {
+ writer.end(extraMetaData);
+ } finally {
+ writer = null;
+ // Release the buffer
+ reusableBlockBuffer = null;
+ }
}
protected abstract Map<String, String> finalizeMetadata();
+ private void closeParquetFileWriterQuietly(ParquetFileWriter
parquetFileWriter) {
+ if (parquetFileWriter == null) {
+ return;
+ }
+ Method closeMethod;
+ try {
+ // ParquetFileWriter does not implement Closeable in all versions, so
attempt close() reflectively.
+ closeMethod = parquetFileWriter.getClass().getMethod("close");
+ } catch (NoSuchMethodException e) {
+ return;
Review Comment:
Simplified this to an `instanceof AutoCloseable` check and removed
reflection. The comment now identifies Parquet 1.12.x/1.13.x as lacking
`close()`; newer versions implement `AutoCloseable`. The older-version
failure-cleanup limitation is also called out in the PR description: fully
fixing it requires direct output-stream ownership.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java:
##########
@@ -485,10 +483,30 @@ public List<WriteStatus> close() {
return Collections.singletonList(writeStatus);
} catch (IOException e) {
+ closeFileWriterQuietly(e);
throw new HoodieUpsertException("Failed to close UpdateHandle", e);
+ } catch (RuntimeException e) {
+ closeFileWriterQuietly(e);
+ throw e;
+ } finally {
+ keyToNewRecords = null;
+ writtenRecordKeys = null;
Review Comment:
Done: removed the success-path assignments and kept `keyToNewRecords = null`
and `writtenRecordKeys = null` in `finally`.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java:
##########
@@ -168,6 +170,11 @@ protected HoodieRecord<T> updateFileName(HoodieRecord<T>
record, HoodieSchema sc
return record.prependMetaFields(schema, targetSchema, metadataValues,
prop);
}
+ private void closeFileWriterQuietly(Throwable failure) {
Review Comment:
This was an oversight. Failure cleanup now calls `markClosed()` before
closing the file writer. The create-handle regression test also checks
`isClosed()` and verifies that a subsequent `close()` succeeds without
attempting finalization again.
--
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]
