voonhous commented on code in PR #18776:
URL: https://github.com/apache/hudi/pull/18776#discussion_r4059224987
##########
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:
**minor:** The native handle is the default log writer from table version 10
on, but the new failure test only drives `HoodieInlineLogAppendHandle` on table
version SIX with `flushAppend` mocked. This `closeLogWriter` change, the
`writeRecord` failure branch (`HoodieAppendHandle:401`) and `write(Map)` are
not reached by any test. Not blocking, but could
`TestHoodieNativeLogAppendHandle` (it already has a `createWriter` harness) get
a case where `writer.close()` throws, asserting `isClosed()`, a null `writer`
and a single close?
##########
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:
**minor:** A Scala `finally { writer.close() }` replaces the original write
exception with any exception from `close()`, which is likely on a writer that
just failed. The Java handles in this PR use `CloseableUtils.closeSuppressing`
for exactly this. Not blocking, but could this be `catch { case t: Throwable =>
CloseableUtils.closeSuppressing(writer, t); throw t }` with `writer.close()` on
the success path?
##########
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:
**minor:** `HoodieParquetFileBinaryCopier.close()` (lines 250-257) calls
`super.close()` before `prefetchExecutor.shutdownNow()` with no `finally`, so
when `end()` throws here the non-daemon executor thread (created at line 202)
and the current input `reader` leak once per failed handle. This path now
reaches that `close()`. Not blocking, but could the executor shutdown and
reader close move into a `finally` around `super.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:
**major:** This failure path calls `copier.close()`, which is the success
finalizer (`finalizeMetadata()` + `writer.end()`). If the failure happens
between row groups (e.g. the prefetch `IOException` in
`HoodieParquetFileBinaryCopier.initNextReader`), `end()` succeeds and leaves a
partial file with a valid footer. This handle creates no write marker, and
`SparkBinaryCopyClusteringExecutionStrategy:136` gives a retry a new fileId, so
nothing reconciles the orphan; before this change it was at least unreadable.
Could the failure path abort instead, releasing the stream and deleting `path`
without calling `end()`?
##########
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:
**minor:** This test still passes if the
`CloseableUtils.closeSuppressing(fileWriter, failure)` line in
`BaseCreateHandle` is deleted: it only checks that the field is null and the
handle is marked closed. Not blocking, but could it capture `HoodieFileWriter
fileWriter = createHandle.fileWriter;` before the `assertThrows` and then
`assertFalse(fileWriter.canWrite())` (`TestFileWriter.canWrite()` returns
`!closed`)?
##########
hudi-common/src/main/java/org/apache/hudi/common/bootstrap/index/hfile/HFileBootstrapIndexWriter.java:
##########
@@ -174,28 +176,60 @@ private void commit() {
* Close Writer Handles.
*/
public void close() {
- try {
- if (!closed) {
- indexByPartitionWriter.close();
- indexByFileIdWriter.close();
- closed = true;
- }
- } catch (IOException ioe) {
- throw new HoodieIOException(ioe.getMessage(), ioe);
+ if (closed) {
+ return;
+ }
+ Exception failure = closeHFileWriter(indexByPartitionWriter, null);
+ failure = closeHFileWriter(indexByFileIdWriter, failure);
+ indexByPartitionWriter = null;
+ indexByFileIdWriter = null;
+ closed = true;
+ if (failure != null) {
+ throw new HoodieException(failure.getMessage(), failure);
Review Comment:
**nit:** `close()` used to throw `HoodieIOException` for an `IOException`
and now throws `HoodieException`, including on the success path through
`finish()`. No in-repo caller depends on it, so feel free to ignore, but could
this keep `HoodieIOException` when `failure` is an `IOException`?
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestHoodieAppendHandle.java:
##########
@@ -87,6 +100,38 @@ private void mockMethodsNeededByConstructor() {
when(mockHoodieTable.getMetaClient()).thenReturn(metaClient);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testFailedFlushClosesWriterAndPreventsAnotherFlush(boolean closeFails)
throws IOException {
+ writeConfig = HoodieWriteConfig.newBuilder()
+ .withProps(writeConfig.getProps())
+ .withWriteTableVersion(HoodieTableVersion.SIX.versionCode())
+ .build();
+ when(mockHoodieTable.getStorage()).thenReturn(metaClient.getStorage());
+ HoodieInlineLogAppendHandle<Object, Object, Object, Object> handle = spy(
+ new HoodieInlineLogAppendHandle<>(writeConfig, TEST_INSTANT_TIME,
mockHoodieTable,
+ TEST_PARTITION_PATH, TEST_FILE_ID, taskContextSupplier));
+ HoodieLogFormat.Writer writer = mock(HoodieLogFormat.Writer.class);
+ handle.writer = writer;
+ handle.recordItr = Collections.emptyIterator();
+ handle.recordList.add(mock(HoodieRecord.class));
Review Comment:
**nit:** This setup line is dead: `flushAppend()` is stubbed to throw and
the later `close()` returns early on `isClosed()`, so the buffered record is
never touched. Feel free to ignore.
##########
hudi-hadoop-common/src/test/java/org/apache/hudi/parquet/io/TestHoodieParquetBinaryCopyBaseSchemaEvolution.java:
##########
@@ -290,6 +299,62 @@ public void
testSchemaEvolutionEnabled_AllowsLegacyConversion() throws Exception
assertEquals(true, legacyConversionAttempted, "Legacy conversion should be
attempted when schema evolution is enabled");
}
+ @Test
+ public void
testCloseParquetFileWriterQuietlyIgnoresWriterWithoutCloseMethod() {
Review Comment:
**nit:** These two reflection tests of the private
`closeParquetFileWriterQuietly` are already covered by the public `close()`
tests below (`testCloseClearsWriterWhenEndFails` uses a non-Closeable mock,
`testCloseReleasesWriterWhenMetadataFails` a Closeable one). On spark4.x a
plain `mock(ParquetFileWriter.class)` is `AutoCloseable`, so this test's name
no longer holds there. Feel free to ignore, but could we drop these two plus
the `getDeclaredMethod`/`Field` helpers in favor of a package-private
`@VisibleForTesting` accessor, maybe with the close tests in their own class?
##########
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);
+ assertTrue(createHandle.isClosed());
+ assertDoesNotThrow(createHandle::close);
+ }
+
+ private static class CreateHandleWithFileWriterWriteFailure extends
HoodieCreateHandle<Object, Object, Object, Object> {
Review Comment:
**nit:** This member class duplicates the same-named local class inside
`testMarkerFileCreatedWhenFileWriterWriteFails` (line 347), which shadows it.
Feel free to ignore, but could we drop the local class so both tests use this
one?
##########
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:
**minor:** On the default build this branch never fires: `ParquetFileWriter`
implements `AutoCloseable` only from parquet 1.14, while the default
`parquet.version` is 1.13.1 and spark3.3/3.4 use 1.12.x (only the spark4.x and
flink2.x profiles are on 1.14+). The PR description already notes this. The
tests reach the branch only via `extraInterfaces(Closeable.class)` mocks. Not
blocking, but could `initFileWriter` wrap the `OutputFile` to keep the
`PositionOutputStream` it creates, so this can close the stream directly on
every parquet version?
##########
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:
**blocker:** With `hoodie.write.executor.type=DISRUPTOR`, marking the handle
closed on failure turns one lost record into the rest of the handle lost, and
the write still succeeds. `DisruptorMessageQueue.setHandlers` swallows the
consumer exception (`markAsFailed`), `BaseHoodieQueueBasedExecutor.execute()`
still calls `finish()`, and `close()` returns a `writeStatus` that never went
through its setup. Same shape in `BaseCreateHandle:132`. Could `close()` on a
handle closed by a failure rethrow that failure instead of returning a status?
<details><summary>Trace (code read, not executed)</summary>
1. `CopyOnWriteInsertHandler.consume` > `BaseCreateHandle.doWrite` fails;
the catch at :132 marks the handle closed, closes and nulls `fileWriter`, and
throws.
2. `DisruptorMessageQueue.setHandlers` catches it, calls `markAsFailed`, and
keeps consuming.
3. Each later record: `CopyOnWriteInsertHandler:105` calls `handle.canWrite`
> `BaseCreateHandle:96` NPEs on the null `fileWriter`, swallowed the same way.
4. `execute()` completes normally > `finish()` > `closeOpenHandles()` >
`close()` takes the `isClosed()` early return with `writeStatus` never passed
through `setupWriteStatus`.
On master the handle kept writing, so only the failing record was lost. The
append path behaves the same for `HoodieEarlyConflictDetectionException`, which
throws even with ignore.failed=true, and its `statuses` skip the log-offset
size fix-up.
</details>
--
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]
