danny0405 commented on code in PR #18776:
URL: https://github.com/apache/hudi/pull/18776#discussion_r4060055211
##########
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:
@voonhous I agree that Disruptor must propagate its stored consumer failure.
The handle already throws from write when `getIgnoreWriteFailed()` is false,
but `setHandlers()` catches it and the executor only checks the stored failure
inside its own catch block. I'm keeping this PR focused on releasing resources
and deferring that executor propagation issue to a separate fix, rather than
adding failure state or rethrow checks to the handles. This concern remains
unaddressed here.
##########
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:
@voonhous Fixed with try-with-resources around `writeIncomingRecords()`, so
the spill map closes even if writing pending records throws. This also
preserves the original exception if map cleanup fails, without a separate
map-cleanup branch. Added tests for the pending-record failure and for
`fileWriter.close()` throwing; they verify cleanup, the original cause, a
single writer close, and a no-op second handle close.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieBinaryCopyHandle.java:
##########
@@ -120,18 +121,31 @@ 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) {
+ closeWriterQuietly(e);
Review Comment:
I'm retaining the existing `close()` finalization path to keep this PR
limited to resource release. It can leave an incomplete copy with a valid
footer, as you describe; aborting and deleting that output would need a
separate lifecycle change. This PR does not address orphan-output cleanup. The
added regression test checks that a copy failure still closes the copier and
preserves the original exception if close also fails.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieBinaryCopyHandle.java:
##########
@@ -120,18 +121,31 @@ 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) {
+ closeWriterQuietly(e);
throw new HoodieIOException(e.getMessage(), e);
+ } catch (RuntimeException e) {
+ closeWriterQuietly(e);
+ throw e;
} finally {
this.recordsWritten = records;
this.insertRecordsWritten = records;
}
log.info("Finish rewriting {}. Using {} mills", this.path,
timer.endTimer());
}
+ private void closeWriterQuietly(Throwable failure) {
+ markClosed();
+ CloseableUtils.closeSuppressing(writer::close, failure);
Review Comment:
Fixed: `close()` now uses try-with-resources for the current reader and a
`finally` block for executor shutdown and buffer release. Both run when
`super.close()` throws. The lifecycle regression test covers simultaneous
finalization and reader-close failures, preserves the original exception, and
checks that the executor terminates.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/SparkHelpers.scala:
##########
@@ -72,14 +72,17 @@ object SparkHelpers {
conf.unwrap().setClassLoader(Thread.currentThread.getContextClassLoader)
val writer = new HoodieAvroParquetWriter(destinationFile, parquetConfig,
instantTime, new SparkTaskContextSupplier(), true)
- for (rec <- sourceRecords) {
- val key: String =
rec.get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString
- if (!keysToSkip.contains(key)) {
+ try {
+ for (rec <- sourceRecords) {
+ val key: String =
rec.get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString
+ if (!keysToSkip.contains(key)) {
- writer.writeAvro(key, rec)
+ writer.writeAvro(key, rec)
+ }
}
+ } finally {
+ writer.close()
Review Comment:
Updated as suggested: the catch calls
`CloseableUtils.closeSuppressing(writer, t)` and rethrows the original write
exception, while the success path calls `writer.close()` directly. The Spark
reactor compiles successfully.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogAppendHandle.java:
##########
@@ -163,8 +163,11 @@ protected void flushAppend() {
@Override
protected void closeLogWriter() {
- if (writer != null) {
- writer.close();
+ try {
+ if (writer != null) {
+ writer.close();
+ }
+ } finally {
writer = null;
Review Comment:
Added native-writer coverage for both `writeRecord` and `write(Map)`
failures, including a cleanup exception, plus a direct writer-close failure.
The tests verify that the handle is closed, the writer reference is null,
subsequent handle close is harmless, and the writer is closed only once.
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/parquet/io/HoodieParquetBinaryCopyBase.java:
##########
@@ -159,24 +160,44 @@ protected void initFileWriter(Path outPutFile,
CompressionCodecName newCodecName
writer.start();
log.info("init writer ");
} catch (Exception e) {
+ closeParquetFileWriterQuietly(e);
log.error("failed to init parquet writer", e);
throw new HoodieException(e);
}
}
@Override
public void close() throws IOException {
- 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;
+ if (writer == null) {
+ return;
+ }
+ try {
+ 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);
+ } catch (IOException | RuntimeException e) {
+ closeParquetFileWriterQuietly(e);
+ throw e;
+ } finally {
+ writer = null;
+ // Release the buffer
+ reusableBlockBuffer = null;
+ }
}
protected abstract Map<String, String> finalizeMetadata();
+ private void closeParquetFileWriterQuietly(Throwable failure) {
+ // Parquet 1.12.x/1.13.x have no close(); newer versions implement
AutoCloseable.
+ ParquetFileWriter parquetFileWriter = writer;
+ writer = null;
+ if (parquetFileWriter instanceof AutoCloseable) {
Review Comment:
Keeping the Parquet 1.12/1.13 limitation explicit rather than introducing
output-stream tracking in this patch. The cleanup comment now states that those
versions expose no close API and a failed start/end can leave the stream open.
Newer versions still use `AutoCloseable`. The revised lifecycle tests use the
actual version's writer interface rather than adding `Closeable` through mock
extra interfaces.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestHoodieCreateHandle.java:
##########
@@ -392,6 +393,40 @@ protected HoodieFileWriter initializeFileWriter() throws
IOException {
assertDoesNotThrow(createHandle::close);
}
+ @Test
+ void testFileWriterClosedWhenDoWriteFails() throws Exception {
+ HoodieWriteConfig failOnWriteConfig = HoodieWriteConfig.newBuilder()
+ .withProps(writeConfig.getProps())
+ .withWriteIgnoreFailed(false)
+ .build();
+ HoodieTable failOnWriteTable = new TestBaseHoodieTable(failOnWriteConfig,
getEngineContext(), metaClient);
+ CreateHandleWithFileWriterWriteFailure createHandle = new
CreateHandleWithFileWriterWriteFailure(
+ failOnWriteConfig, TEST_INSTANT_TIME, failOnWriteTable,
TEST_PARTITION_PATH, TEST_FILE_ID, taskContextSupplier);
+ HoodieRecord testRecord = dataGen.generateInserts(TEST_INSTANT_TIME,
1).get(0);
+
+ HoodieException exception = assertThrows(HoodieException.class, () ->
+ createHandle.doWrite(testRecord, TEST_SCHEMA, new TypedProperties()));
+
+ assertEquals("Simulated file writer write failure",
exception.getMessage());
+ assertNull(createHandle.fileWriter);
Review Comment:
Done: the test captures the file writer before the failing write and asserts
`assertFalse(fileWriter.canWrite())` afterward, so it verifies actual writer
closure as well as the cleared handle reference.
--
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]
