This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 3ec54a52d782 fix(common): propagate native log flush failures despite
ignore failed writes (#20133)
3ec54a52d782 is described below
commit 3ec54a52d782d3af4c94000a5b728e4c40fbe6ec
Author: Shuo Cheng <[email protected]>
AuthorDate: Wed Sep 30 10:51:31 2026 +0800
fix(common): propagate native log flush failures despite ignore failed
writes (#20133)
* fix(common): propagate native log flush failures despite ignore failed
writes
---
.../org/apache/hudi/io/HoodieAppendHandle.java | 6 +-
.../hudi/io/HoodieNativeLogAppendHandle.java | 2 +-
.../hudi/io/TestHoodieNativeLogAppendHandle.java | 122 +++++++++++++++++++++
3 files changed, 128 insertions(+), 2 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java
index 8bd2eff5e1ec..56a48a82dd80 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java
@@ -46,6 +46,7 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.ExceptionUtil;
+import org.apache.hudi.exception.HoodieAppendException;
import org.apache.hudi.exception.HoodieEarlyConflictDetectionException;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieUpsertException;
@@ -401,7 +402,10 @@ public abstract class HoodieAppendHandle<T, I, K, O>
extends HoodieWriteHandle<T
return true;
} catch (Exception e) {
log.error("Error writing record {}", hoodieRecord, e);
- if (!config.getIgnoreWriteFailed() || ExceptionUtil.isCausedBy(e,
HoodieEarlyConflictDetectionException.class)) {
+ // A failed log flush affects the entire buffered batch, not just the
current record.
+ if (!config.getIgnoreWriteFailed()
+ || ExceptionUtil.isCausedBy(e, HoodieAppendException.class)
+ || ExceptionUtil.isCausedBy(e,
HoodieEarlyConflictDetectionException.class)) {
closeLogWriterQuietly(e);
throw new HoodieException(e.getMessage(), e);
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogAppendHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogAppendHandle.java
index aae1a6b476dc..b6d76f345dfa 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogAppendHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogAppendHandle.java
@@ -162,7 +162,7 @@ public class HoodieNativeLogAppendHandle<T, I, K, O>
extends HoodieAppendHandle<
writer.flushAppend(getUpdatedHeader(header));
processAppendResults(writer.getLastAppendResults());
}
- } catch (IOException e) {
+ } catch (Exception e) {
throw new HoodieAppendException("Failed while flushing records to native
log for fileId " + fileId, e);
}
}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestHoodieNativeLogAppendHandle.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestHoodieNativeLogAppendHandle.java
index cf68cf4367e8..f700bfe7c8d6 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestHoodieNativeLogAppendHandle.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestHoodieNativeLogAppendHandle.java
@@ -44,6 +44,8 @@ import org.apache.hudi.table.HoodieTable;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.InOrder;
import org.mockito.MockedConstruction;
@@ -53,10 +55,12 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -70,6 +74,7 @@ import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
public class TestHoodieNativeLogAppendHandle {
@@ -222,6 +227,123 @@ public class TestHoodieNativeLogAppendHandle {
}
}
+ private static Stream<Arguments> rolloverFlushFailures() {
+ return Stream.of(new IOException("flush failed"), new
IllegalStateException("metadata failed"))
+ .flatMap(failure -> Stream.of(false, true).map(deleteRecord ->
Arguments.of(failure, deleteRecord)));
+ }
+
+ @ParameterizedTest
+ @MethodSource("rolloverFlushFailures")
+ public void testRolloverFlushFailureIsNotIgnored(Exception failure, boolean
deleteRecord) throws Exception {
+ HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
+ .withProps(config().getProps())
+ .withAllowOperationMetadataField(false)
+ .withWriteIgnoreFailed(true)
+ .build();
+ RuntimeException closeFailure = new IllegalStateException("close failed");
+ try (MockedConstruction<HoodieNativeLogFormatWriter> writers =
mockConstruction(
+ HoodieNativeLogFormatWriter.class, (writer, context) -> {
+ when(writer.hasPendingWrites()).thenReturn(true);
+ when(writer.canWriteDataFile()).thenReturn(false);
+ when(writer.canWriteDeleteFile()).thenReturn(false);
+ doThrow(failure).when(writer).flushAppend(any());
+ doThrow(closeFailure).when(writer).close();
+ })) {
+ TestableNativeLogAppendHandle handle = new
TestableNativeLogAppendHandle(config, table(config));
+ handle.createWriter();
+ handle.doInit = false;
+ HoodieRecord record = mock(HoodieRecord.class);
+ HoodieRecord nextRecord = mock(HoodieRecord.class);
+ when(record.getPartitionPath()).thenReturn("partition");
+ when(record.getMetadata()).thenReturn(Option.empty());
+ when(record.isDelete(any(), any())).thenReturn(deleteRecord);
+ when(record.prependMetaFields(any(), any(), any(),
any())).thenReturn(record);
+ handle.recordItr = Arrays.asList(record, nextRecord).iterator();
+
+ HoodieException exception = assertThrows(HoodieException.class,
handle::doAppend);
+ assertEquals(HoodieAppendException.class,
exception.getCause().getClass());
+ assertSame(failure, exception.getCause().getCause());
+ assertArrayEquals(new Throwable[] {closeFailure},
exception.getCause().getSuppressed());
+ assertTrue(handle.isClosed());
+ assertNull(handle.getWriter());
+ assertNull(handle.recordItr);
+ assertFalse(handle.writeStatus.hasErrors(), "A flush failure must not be
recorded as a single-record failure");
+ assertDoesNotThrow(handle::close);
+
+ HoodieNativeLogFormatWriter writer = writers.constructed().get(0);
+ verify(writer).flushAppend(any());
+ verify(writer).close();
+ verify(writer, never()).appendRecord(any(), any());
+ verify(writer, never()).appendDeleteRecord(any(), any());
+ verifyNoInteractions(nextRecord);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testRolloverFlushErrorPropagatesUnchanged(boolean deleteRecord)
throws Exception {
+ HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
+ .withProps(config().getProps())
+ .withAllowOperationMetadataField(false)
+ .withWriteIgnoreFailed(true)
+ .build();
+ AssertionError failure = new AssertionError("flush failed");
+ try (MockedConstruction<HoodieNativeLogFormatWriter> writers =
mockConstruction(
+ HoodieNativeLogFormatWriter.class, (writer, context) -> {
+ when(writer.hasPendingWrites()).thenReturn(true);
+ doThrow(failure).when(writer).flushAppend(any());
+ })) {
+ TestableNativeLogAppendHandle handle = new
TestableNativeLogAppendHandle(config, table(config));
+ handle.createWriter();
+ handle.doInit = false;
+ HoodieRecord record = mock(HoodieRecord.class);
+ HoodieRecord nextRecord = mock(HoodieRecord.class);
+ when(record.getPartitionPath()).thenReturn("partition");
+ when(record.getMetadata()).thenReturn(Option.empty());
+ when(record.isDelete(any(), any())).thenReturn(deleteRecord);
+ when(record.prependMetaFields(any(), any(), any(),
any())).thenReturn(record);
+ handle.recordItr = Arrays.asList(record, nextRecord).iterator();
+
+ try {
+ // Errors bypass the Exception handlers instead of being wrapped as
append exceptions.
+ assertSame(failure, assertThrows(AssertionError.class,
handle::doAppend));
+ assertFalse(handle.writeStatus.hasErrors());
+ HoodieNativeLogFormatWriter writer = writers.constructed().get(0);
+ verify(writer).flushAppend(any());
+ verify(writer, never()).appendRecord(any(), any());
+ verify(writer, never()).appendDeleteRecord(any(), any());
+ verifyNoInteractions(nextRecord);
+ } finally {
+ handle.closeWriter();
+ }
+ }
+ }
+
+ @Test
+ public void testRecordWriteFailureCanStillBeIgnored() throws Exception {
+ HoodieWriteConfig config = config();
+ config.setValue(HoodieWriteConfig.IGNORE_FAILED, "true");
+ try (MockedConstruction<HoodieNativeLogFormatWriter> writers =
mockConstruction(
+ HoodieNativeLogFormatWriter.class, (writer, context) -> {
+ when(writer.canWriteDataFile()).thenReturn(true);
+ doThrow(new IOException("record write
failed")).when(writer).appendRecord(any(), any());
+ })) {
+ TestableNativeLogAppendHandle handle = new
TestableNativeLogAppendHandle(config, table(config));
+ handle.createWriter();
+ handle.doInit = false;
+ HoodieRecord record = mock(HoodieRecord.class);
+ when(record.getPartitionPath()).thenReturn("partition");
+ when(record.getMetadata()).thenReturn(Option.empty());
+ when(record.prependMetaFields(any(), any(), any(),
any())).thenReturn(record);
+
+ assertFalse(handle.writeRecord(record));
+ assertFalse(handle.isClosed());
+ assertEquals(1, handle.writeStatus.getTotalErrorRecords());
+ verify(writers.constructed().get(0), never()).close();
+ handle.close();
+ }
+ }
+
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testWriteFailureClosesNativeWriter(boolean useRecordMap) throws
Exception {
