[HUDI-8400] apply 'write.ignore.failed' when write data failed v2

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java

@@ -811,6 +811,12 @@ public class HoodieWriteConfig extends HoodieConfig {
    */
   public static final String WRITES_FILEID_ENCODING = "_hoodie.writes.fileid.encoding";
 
+  public static final ConfigProperty<Boolean> IGNORE_FAILED = ConfigProperty
+      .key("hoodie.write.ignore.failed")
+      .defaultValue(false)
+      .sinceVersion("")
+      .withDocumentation("Flag to indicate whether to ignore any non exception error (e.g. writestatus error).");
+
   private ConsistencyGuardConfig consistencyGuardConfig;
   private FileSystemRetryConfig fileSystemRetryConfig;
 
@@ -2788,6 +2794,13 @@ public int getSecondaryIndexParallelism() {
     return metadataConfig.getSecondaryIndexParallelism();
   }
 
+  /**
+   * Whether to ignore the write failed.
+   */
+  public boolean getIgnoreWriteFailed() {
+    return getBooleanOrDefault(IGNORE_FAILED);
+  }
+
   public static class Builder {
 
     protected final HoodieWriteConfig writeConfig = new HoodieWriteConfig();
@@ -3330,6 +3343,11 @@ public Builder withWriteRecordPositionsEnabled(boolean shouldWriteRecordPosition
       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

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java

@@ -499,6 +499,7 @@ protected void doWrite(HoodieRecord record, Schema schema, TypedProperties props
       // for a single record
       writeStatus.markFailure(record, t, recordMetadata);
       LOG.error("Error writing record " + record, t);
+      ignoreWriteFailed(t);
     }
   }
 

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCreateHandle.java

@@ -170,6 +170,7 @@ protected void doWrite(HoodieRecord record, Schema schema, TypedProperties props
       // for a single record
       writeStatus.markFailure(record, t, recordMetadata);
       LOG.error("Error writing record " + record, t);
+      ignoreWriteFailed(t);
     }
   }
 

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java

@@ -304,6 +304,7 @@ private boolean writeRecord(HoodieRecord<T> newRecord, Option<HoodieRecord> comb
       HoodieUpsertException failureEx = new HoodieUpsertException("mismatched partition path, record partition: "
           + newRecord.getPartitionPath() + " but trying to insert into partition: " + partitionPath);
       writeStatus.markFailure(newRecord, failureEx, recordMetadata);
+      ignoreWriteFailed(failureEx);
       return false;
     }
     try {
@@ -333,6 +334,7 @@ private boolean writeRecord(HoodieRecord<T> newRecord, Option<HoodieRecord> comb
     } catch (Exception e) {
       LOG.error("Error writing record  " + newRecord, e);
       writeStatus.markFailure(newRecord, e, recordMetadata);
+      ignoreWriteFailed(e);
     }
     return false;
   }

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteHandle.java

@@ -174,6 +174,16 @@ protected void doWrite(HoodieRecord record, Schema schema, TypedProperties props
     // NO_OP
   }
 
+  /**
+   * ignore failed write based on 'getIgnoreWriteFailed' config
+   * @param throwable
+   */
+  protected void ignoreWriteFailed(Throwable throwable) {
+    if (config.getIgnoreWriteFailed()) {
+      throw new HoodieException(throwable.getMessage(), throwable);
+    }
+  }
+
   /**
    * Perform the actual writing of the given record into the backing file.
    */

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.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;
@@ -142,7 +143,11 @@ public void write(String recordKey, String partitionPath, RowData record) throws
             ? HoodieRecordDelegate.create(recordKey, partitionPath, null, newRecordLocation) : null;
         writeStatus.markSuccess(recordDelegate, Option.empty());
       } catch (Throwable t) {
+        LOG.error("Failed to write : key is " + recordKey + ", data is " + rowData, t);
         writeStatus.markFailure(recordKey, partitionPath, t);
+        if (!writeConfig.getIgnoreWriteFailed()) {
+          throw new HoodieException(t.getMessage(), t);
+        }
       }
     } catch (Throwable ge) {
       writeStatus.setGlobalError(ge);

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java

@@ -439,6 +439,7 @@ private FlinkOptions() {
       .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.");

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/FlinkWriteClients.java

@@ -225,7 +225,8 @@ public static HoodieWriteConfig getHoodieClientConfig(
             .withAutoCommit(false)
             .withAllowOperationMetadataField(conf.getBoolean(FlinkOptions.CHANGELOG_ENABLED))
             .withProps(flinkConf2TypedProperties(conf))
-            .withSchema(getSourceSchema(conf).toString());
+            .withSchema(getSourceSchema(conf).toString())
+            .withWriteIgnoreFailed(conf.get(FlinkOptions.IGNORE_FAILED));
 
     Option<HoodieLockConfig> lockConfig = getLockConfig(conf);
     if (lockConfig.isPresent()) {