This is an automated email from the ASF dual-hosted git repository. danny0405 pushed a commit to branch release-1.1.2 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 8734ffbe8eb64d924f07950231fb8bb7eacfaf2b Author: fhan <[email protected]> AuthorDate: Sat Jan 31 12:59:56 2026 +0800 fix: allows eager failure from abnormals for streaming write (#12150) * apply hoodie.write.ignore.failed when write data failed; * the option is default true in Hudi write config for backward compatibility; * the option is default false for Flink streaming ingestion; * fix the commit of coordinator to remove the rollback, now the instant would be committed regardless of the write errors. --------- Co-authored-by: fhan <[email protected]> Co-authored-by: danny0405 <[email protected]> (cherry picked from commit 04326111808a5bc80f8cb2d5da2f75fa3dcf2091) --- .../org/apache/hudi/config/HoodieWriteConfig.java | 19 ++++++ .../java/org/apache/hudi/io/BaseCreateHandle.java | 8 ++- .../hudi/io/FileGroupReaderBasedMergeHandle.java | 4 ++ .../org/apache/hudi/io/HoodieAppendHandle.java | 10 ++- .../org/apache/hudi/io/HoodieWriteMergeHandle.java | 4 ++ .../io/storage/row/HoodieRowDataCreateHandle.java | 5 ++ .../hudi/io/storage/row/HoodieRowCreateHandle.java | 4 ++ .../apache/hudi/configuration/FlinkOptions.java | 1 + .../hudi/sink/StreamWriteOperatorCoordinator.java | 72 +++++++++------------- .../org/apache/hudi/util/FlinkWriteClients.java | 3 +- 10 files changed, 81 insertions(+), 49 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java index 53eb3d383ace..0a74fcfd0f12 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java @@ -919,6 +919,13 @@ public class HoodieWriteConfig extends HoodieConfig { + " or when using a custom Hoodie Concat Handle Implementation controlled by the config " + CONCAT_HANDLE_CLASS_NAME.key() + ", enabling this config results in fallback to the default implementations if instantiation of the custom implementation fails"); + public static final ConfigProperty<Boolean> IGNORE_FAILED = ConfigProperty + .key("hoodie.write.ignore.failed") + .defaultValue(true) + .sinceVersion("") + .withDocumentation("Flag to indicate whether to ignore any non exception error (e.g. write status error)." + + "By default true for backward compatibility."); + /** * Config key with boolean value that indicates whether record being written during MERGE INTO Spark SQL * operation are already prepped. @@ -2961,6 +2968,13 @@ public class HoodieWriteConfig extends HoodieConfig { } } + /** + * Whether to ignore the write failed. + */ + public boolean getIgnoreWriteFailed() { + return getBooleanOrDefault(IGNORE_FAILED); + } + public static class Builder { protected final HoodieWriteConfig writeConfig = new HoodieWriteConfig(); @@ -3542,6 +3556,11 @@ public class HoodieWriteConfig extends HoodieConfig { return this; } + public Builder withWriteIgnoreFailed(boolean ignoreFailedWriteData) { + writeConfig.setValue(IGNORE_FAILED, String.valueOf(ignoreFailedWriteData)); + return this; + } + protected void setDefaults() { writeConfig.setDefaultValue(MARKERS_TYPE, getDefaultMarkersType(engineType)); // Check for mandatory properties diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java index 7f7c0d244b37..b24f040a7ccf 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java @@ -30,6 +30,7 @@ import org.apache.hudi.common.model.IOType; import org.apache.hudi.common.model.MetadataValues; import org.apache.hudi.common.util.Option; import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.exception.HoodieException; import org.apache.hudi.exception.HoodieInsertException; import org.apache.hudi.io.storage.HoodieFileWriter; import org.apache.hudi.storage.StoragePath; @@ -117,10 +118,11 @@ public abstract class BaseCreateHandle<T, I, K, O> extends HoodieWriteHandle<T, // record successful. record.deflate(); } catch (Throwable t) { - // Not throwing exception from here, since we don't want to fail the entire job - // for a single record - writeStatus.markFailure(record, t, recordMetadata); LOG.error("Error writing record " + record, t); + if (!config.getIgnoreWriteFailed()) { + throw new HoodieException(t.getMessage(), t); + } + writeStatus.markFailure(record, t, recordMetadata); } } diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java index 7924a1aa48bf..de6956d4e657 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java @@ -46,6 +46,7 @@ import org.apache.hudi.common.table.read.HoodieReadStats; import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.collection.ClosableIterator; import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.exception.HoodieException; import org.apache.hudi.exception.HoodieUpsertException; import org.apache.hudi.internal.schema.InternalSchema; import org.apache.hudi.internal.schema.utils.AvroSchemaEvolutionUtils; @@ -291,6 +292,9 @@ public class FileGroupReaderBasedMergeHandle<T, I, K, O> extends HoodieWriteMerg recordsWritten++; } catch (Exception e) { LOG.error("Error writing record {}", record, e); + if (!config.getIgnoreWriteFailed()) { + throw new HoodieException(e.getMessage(), e); + } writeStatus.markFailure(record, e, recordMetadata); fileGroupReader.onWriteFailure(record.getRecordKey()); } 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 1c2a07c752c6..aa907b081256 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 @@ -317,6 +317,9 @@ public class HoodieAppendHandle<T, I, K, O> extends HoodieWriteHandle<T, I, K, O hoodieRecord.deflate(); } catch (Exception e) { LOG.error("Error writing record {}", hoodieRecord, e); + if (!config.getIgnoreWriteFailed()) { + throw new HoodieException(e.getMessage(), e); + } writeStatus.markFailure(hoodieRecord, e, recordMetadata); } } @@ -524,10 +527,11 @@ public class HoodieAppendHandle<T, I, K, O> extends HoodieWriteHandle<T, I, K, O flushToDiskIfRequired(record, false); writeToBuffer(record); } catch (Throwable t) { - // Not throwing exception from here, since we don't want to fail the entire job - // for a single record - writeStatus.markFailure(record, t, recordMetadata); LOG.error("Error writing record " + record, t); + if (!config.getIgnoreWriteFailed()) { + throw new HoodieException(t.getMessage(), t); + } + writeStatus.markFailure(record, t, recordMetadata); } } diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java index 8e11d0e941bb..e7a3de46f70e 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java @@ -39,6 +39,7 @@ import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.collection.ExternalSpillableMap; import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.exception.HoodieCorruptedDataException; +import org.apache.hudi.exception.HoodieException; import org.apache.hudi.exception.HoodieIOException; import org.apache.hudi.exception.HoodieUpsertException; import org.apache.hudi.io.storage.HoodieFileReader; @@ -342,6 +343,9 @@ public class HoodieWriteMergeHandle<T, I, K, O> extends HoodieAbstractMergeHandl return true; } catch (Exception e) { LOG.error("Error writing record {}", newRecord, e); + if (!config.getIgnoreWriteFailed()) { + throw new HoodieException(e.getMessage(), e); + } writeStatus.markFailure(newRecord, e, recordMetadata); } return false; diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataCreateHandle.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataCreateHandle.java index 170b6e46c34d..a989cb1c08a7 100644 --- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataCreateHandle.java +++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataCreateHandle.java @@ -31,6 +31,7 @@ import org.apache.hudi.common.model.IOType; import org.apache.hudi.common.util.HoodieTimer; import org.apache.hudi.common.util.Option; import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.exception.HoodieException; import org.apache.hudi.exception.HoodieIOException; import org.apache.hudi.exception.HoodieInsertException; import org.apache.hudi.storage.HoodieStorage; @@ -151,6 +152,10 @@ public class HoodieRowDataCreateHandle implements Serializable { ? HoodieRecordDelegate.create(recordKey, partitionPath, null, newRecordLocation) : null; writeStatus.markSuccess(recordDelegate, Option.empty()); } catch (Throwable t) { + LOG.error("Error writing record " + record, t); + if (!writeConfig.getIgnoreWriteFailed()) { + throw new HoodieException(t.getMessage(), t); + } writeStatus.markFailure(recordKey, partitionPath, t); } } catch (Throwable ge) { diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowCreateHandle.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowCreateHandle.java index 04e3683274e0..fb847d8500a4 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowCreateHandle.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowCreateHandle.java @@ -196,6 +196,10 @@ public class HoodieRowCreateHandle implements Serializable { ? HoodieRecordDelegate.create(recordKey.toString(), partitionPath.toString(), null, newRecordLocation) : null; writeStatus.markSuccess(recordDelegate, Option.empty()); } catch (Exception t) { + LOG.error("Error writing record " + row, t); + if (!writeConfig.getIgnoreWriteFailed()) { + throw new HoodieException(t.getMessage(), t); + } writeStatus.markFailure(recordKey.toString(), partitionPath.toString(), t); } } catch (Exception e) { diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java index 7ba63278497e..ee628fd19211 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java @@ -485,6 +485,7 @@ public class FlinkOptions extends HoodieConfig { .key("write.ignore.failed") .booleanType() .defaultValue(false) + .withFallbackKeys("hoodie.write.ignore.failed") .withDescription("Flag to indicate whether to ignore any non exception error (e.g. writestatus error). within a checkpoint batch. \n" + "By default false. Turning this on, could hide the write status errors while the flink checkpoint moves ahead. \n" + "So, would recommend users to use this with caution."); diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java index 68bb54f8a1ee..a0d9a40f7257 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java @@ -589,52 +589,40 @@ public class StreamWriteOperatorCoordinator */ @SuppressWarnings("unchecked") private void doCommit(long checkpointId, String instant, List<WriteStatus> writeResults) { - // commit or rollback + // commit and error logging + HashMap<String, String> checkpointCommitMetadata = new HashMap<>(); + StreamerUtil.addFlinkCheckpointIdIntoMetaData(conf, checkpointCommitMetadata, checkpointId); + final Map<String, List<String>> partitionToReplacedFileIds = tableState.isOverwrite + ? writeClient.getPartitionToReplacedFileIds(tableState.operationType, writeResults) + : Collections.emptyMap(); + boolean success = writeClient.commit(instant, writeResults, Option.of(checkpointCommitMetadata), + tableState.commitAction, partitionToReplacedFileIds); + if (success) { + this.eventBuffers.reset(checkpointId); + LOG.info("Commit instant [{}] success!", instant); + } else { + throw new HoodieException(String.format("Commit instant [%s] failed!", instant)); + } + long totalErrorRecords = writeResults.stream().map(WriteStatus::getTotalErrorRecords).reduce(Long::sum).orElse(0L); long totalRecords = writeResults.stream().map(WriteStatus::getTotalRecords).reduce(Long::sum).orElse(0L); boolean hasErrors = totalErrorRecords > 0; - if (!hasErrors || this.conf.get(FlinkOptions.IGNORE_FAILED)) { - HashMap<String, String> checkpointCommitMetadata = new HashMap<>(); - StreamerUtil.addFlinkCheckpointIdIntoMetaData(conf, checkpointCommitMetadata, checkpointId); - - if (hasErrors) { - LOG.warn("Some records failed to merge but forcing commit since commitOnErrors set to true. Errors/Total={}/{}", - totalErrorRecords, totalRecords); - } - - final Map<String, List<String>> partitionToReplacedFileIds = tableState.isOverwrite - ? writeClient.getPartitionToReplacedFileIds(tableState.operationType, writeResults) - : Collections.emptyMap(); - boolean success = writeClient.commit(instant, writeResults, Option.of(checkpointCommitMetadata), - tableState.commitAction, partitionToReplacedFileIds); - if (success) { - this.eventBuffers.reset(checkpointId); - LOG.info("Commit instant [{}] success!", instant); - } else { - throw new HoodieException(String.format("Commit instant [%s] failed!", instant)); - } - } else { - if (LOG.isErrorEnabled()) { - LOG.error("Error when writing. Errors/Total={}/{}", totalErrorRecords, totalRecords); - LOG.error("The first 10 files with write errors:"); - writeResults.stream().filter(WriteStatus::hasErrors).limit(10).forEach(ws -> { - if (ws.getGlobalError() != null) { - LOG.error("Global error for partition path {} and fileID {}: {}", - ws.getPartitionPath(), ws.getFileId(), ws.getGlobalError()); - } - if (!ws.getErrors().isEmpty()) { - LOG.error("The first 100 records-level errors for partition path {} and fileID {}:", - ws.getPartitionPath(), ws.getFileId()); - ws.getErrors().entrySet().stream().limit(100).forEach(entry -> - LOG.error("Error for key: {} and Exception: {}", entry.getKey(), entry.getValue().getMessage())); - } - }); - } - - // Rolls back instant - writeClient.rollback(instant); - throw new HoodieException(String.format("Commit instant [%s] failed and rolled back !", instant)); + if (hasErrors && LOG.isErrorEnabled()) { + LOG.error("Error when writing. Errors/Total={}/{}", totalErrorRecords, totalRecords); + LOG.error("The first 10 files with write errors:"); + writeResults.stream().filter(WriteStatus::hasErrors).limit(10).forEach(ws -> { + if (ws.getGlobalError() != null) { + LOG.error("Global error for partition path {} and fileID {}: {}", + ws.getPartitionPath(), ws.getFileId(), ws.getGlobalError()); + } + if (!ws.getErrors().isEmpty()) { + LOG.error("The first 100 records-level errors for partition path {} and fileID {}:", + ws.getPartitionPath(), ws.getFileId()); + ws.getErrors().entrySet().stream().limit(100).forEach(entry -> + LOG.error("Error for key: {} and Exception: {}", entry.getKey(), entry.getValue().getMessage())); + } + }); } } diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/FlinkWriteClients.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/FlinkWriteClients.java index 3fef94c8e60b..10c0b46b12b1 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/FlinkWriteClients.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/FlinkWriteClients.java @@ -241,7 +241,8 @@ public class FlinkWriteClients { .withEmbeddedTimelineServerReuseEnabled(true) // make write client embedded timeline service singleton .withAllowOperationMetadataField(conf.get(FlinkOptions.CHANGELOG_ENABLED)) .withProps(flinkConf2TypedProperties(conf)) - .withSchema(getSourceSchema(conf).toString()); + .withSchema(getSourceSchema(conf).toString()) + .withWriteIgnoreFailed(conf.get(FlinkOptions.IGNORE_FAILED)); // <merge_mode, payload_class, merge_strategy_id> Triple<RecordMergeMode, String, String> mergingBehavior = StreamerUtil.inferMergingBehavior(conf);
