This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d1b9e3f7dc4f653b653d88edd93cde4cd5b80a5d Author: Xinli Shang <[email protected]> AuthorDate: Mon Jun 1 10:16:49 2026 -0500 feat(utilities): migrate HoodieStreamerWriteStatusValidator into pre-commit validator framework (#18765) * feat(utilities): migrate HoodieStreamerWriteStatusValidator into pre-commit validator framework (#18750) Completes the migration tracked in issue #18750 by deleting HoodieStreamerWriteStatusValidator (HSWSV) and replacing it with explicit pre-commit orchestration in StreamSync. Implements all 4 phases of the plan in one change. What changed - StreamSync.writeToSinkAndDoMetaSync() now orchestrates explicitly: 1. run user-configured pre-commit validators 2. count records (SuccessfulRecordCounter) 3. commit error table with strategy handling (ErrorTableCommitter) 4. apply the write-error gate (preserves commitOnErrors semantics) 5. call writeClient.commit() WITHOUT a WriteStatusValidator callback - HoodieStreamerWriteStatusValidator inner class removed (~100 LOC). - HSWSV's three concerns extracted into named helpers: - SuccessfulRecordCounter — pure counting, supports error-table unification - ErrorTableCommitter — error-table commit, returns success/failure - WriteErrorReporter — top-N errored-status logging - SparkWriteErrorValidator added as an opt-in BasePreCommitValidator that applies the same write-error check using the framework's failure.policy. - HoodiePreCommitValidatorConfig.VALIDATOR_CLASS_NAMES doc references the new validator. What did NOT change - WriteStatusValidator interface and the writeClient.commit() hook are preserved — DataSourceUtils.SparkDataSourceWriteStatusValidator is another active caller in the Spark datasource path. - Default behavior for users who do not configure hoodie.precommit.validators is unchanged: the inline error gate in StreamSync preserves HSWSV semantics (commitOnErrors=false fails on any write error; commitOnErrors=true logs a warning and proceeds). - Error-table failure strategies (ROLLBACK_COMMIT, LOG_ERROR) preserved. Note: because the orchestration now runs before writeClient.commit(), ROLLBACK_COMMIT no longer needs to roll back — the commit simply doesn't happen. Verification - mvn -pl hudi-utilities -am test-compile: BUILD SUCCESS - mvn -pl hudi-utilities test -Dtest=TestSparkKafkaOffsetValidator,TestSparkValidationContext, TestSparkStreamerValidatorUtils,TestSparkWriteErrorValidator, TestSuccessfulRecordCounter -> 49/49 pass - mvn -pl hudi-utilities test -Dtest=TestStreamSync,TestHoodieStreamerUtils -> 51/51 pass - mvn -pl hudi-utilities,hudi-client/hudi-client-common checkstyle:check -> 0 violations - mvn -pl hudi-utilities,hudi-client/hudi-client-common apache-rat:check -> 0 unapproved Diff: 700 insertions, 135 deletions across 8 files (6 new, 2 modified). * fix(utilities): address self-review and Gemini-review findings on #18750 Follow-up to the previous commit on this branch. Addresses two batches of issues found in self-review and one external Gemini review. StreamSync orchestration ------------------------ - Reorder: commit error table BEFORE running validators. Previously, if SparkWriteErrorValidator or another validator failed first, the error-table records for the batch were lost. Now error records survive validator-driven aborts. - Split path: only collect writeStatuses to the driver when validators are configured. The default no-validator path now uses a distributed Spark aggregation (SuccessfulRecordCounter.computeFromRdd), restoring pre-#18750 no-overhead behavior. - Reuse already-collected list for WriteErrorReporter when available, to avoid a redundant Spark action. - Document the latent error-table / data-table inconsistency that predates #18750 (preserved, not regressed). - Document why SparkWriteErrorValidator is intentionally redundant with the inline Step 4 gate. SparkWriteErrorValidator ------------------------ - Update stale Javadoc: HSWSV is deleted in this branch, not "running alongside". - Fix user-facing error message: previously referenced a non-existent config key hoodie.streamer.commit.on.errors=true. Now references the real CLI flag --commit-on-errors. - Reject invalid failure.policy values at construction with a clear error listing allowed values, instead of a raw IllegalArgumentException with a confusing message. SuccessfulRecordCounter ----------------------- - Add computeFromRdd entry point so callers without a pre-collected list can compute counts via distributed Spark aggregation. - Switch from mapToDouble(...).sum().longValue() to a long fold to avoid silent precision loss above 2^53 record counts. - Add null guards on public entry points. ErrorTableCommitter ------------------- - Reword "side-effect-only" Javadoc which read as the opposite of intent. - Add null guards on public entry point. WriteErrorReporter ------------------ - Add List<WriteStatus> overload so callers that already have the collected list avoid an extra Spark action. - Demote the "Printing out the top N errors" header from ERROR to INFO (the header is not itself an error). - Standardize on Lombok @Slf4j for logger setup (was manual LoggerFactory), matching the convention used by other classes in the package. Tests ----- - TestErrorTableCommitter: 9 new tests covering unification on/off paths, success/failure propagation, RDD-no-op contract, and null safety. - TestWriteErrorReporter: 6 new tests covering null and empty inputs, max-cap enforcement, and the List overload. - TestSuccessfulRecordCounter: 5 new tests covering the unification path and computeFromRdd via a real local Spark context, plus null safety. - TestSparkWriteErrorValidator: 3 new tests for invalid policy parsing and regression test that the error message references --commit-on-errors. Verification ------------ - mvn -pl hudi-utilities test (helper + validator tests): 73/73 pass - mvn -pl hudi-utilities test -Dtest=TestStreamSync,TestHoodieStreamerUtils: 51/51 pass - mvn -pl hudi-utilities,hudi-client/hudi-client-common checkstyle:check: 0 violations - mvn -pl hudi-utilities,hudi-client/hudi-client-common apache-rat:check: 0 unapproved * ci: re-trigger CI run The previous run hit a Microsoft Container Registry block on the Azurite image pull, failing the integration-tests job for reasons unrelated to this change. Empty commit to trigger a fresh CI run. The underlying flake is fixed in #18772 (gracefully skips when MCR is blocked); once that merges, this PR can rebase and the failure mode will no longer block CI. * fix(utilities): address review feedback on #18750 - Always cache+collect writeStatuses; drop the dual-path (validators vs. no-validators) collect branching and the null sentinel (@danny0405). - Use the 6-arg writeClient.commit() overload — the trailing Option.empty() WriteStatusValidator slot was redundant (@danny0405). - Re-add writeClient.rollback(instantTime) before throwing on error-table ROLLBACK_COMMIT and the write-error gate so the inflight data-table instant doesn't leak under LAZY failed-writes cleanup policy (preserves HSWSV behavior). - Widen the latent-quirk comment on Step 1 to call out that a Step 2 validator failure (including the offset validator) has the same error-table-vs-data-table divergence as the Step 4 gate. - SparkWriteErrorValidator: import java.util.Arrays and drop the inline FQN. - Delete dead helpers exposed only by the removed no-validators path: SuccessfulRecordCounter.computeFromRdd and WriteErrorReporter.logTopErrors(JavaRDD), plus their tests. Verified: TestSparkKafkaOffsetValidator, TestSparkValidationContext, TestSparkStreamerValidatorUtils, TestSparkWriteErrorValidator, TestSuccessfulRecordCounter, TestWriteErrorReporter 61/61. TestStreamSync, TestHoodieStreamerUtils 51/51. * fix(utilities): address hudi-agent review feedback on #18750 - StreamSync Step 2: roll back the inflight data-table instant when a validator throws HoodieValidationException, so it doesn't leak under LAZY failed-writes cleanup. Same argument as Step 1 ROLLBACK_COMMIT and the Step 4 gate. Error-table records committed in Step 1 are preserved by design. - SparkWriteErrorValidator Javadoc: call out that under hoodie.errortable.write.unification.enabled=true the validator is strictly weaker than HSWSV because ValidationContext exposes only data-table stats. Users who need unified error counts should keep the inline commitOnErrors gate enabled. - SuccessfulRecordCounter: replace WriteStatusLongExtractor enum with a Function<WriteStatus, Long> parameter; call sites use method references WriteStatus::getTotalRecords / ::getTotalErrorRecords. - TestSuccessfulRecordCounter: replace inline org.apache.spark.* FQNs with imports for SparkConf and JavaSparkContext. Verified: TestSuccessfulRecordCounter, TestSparkWriteErrorValidator, TestErrorTableCommitter, TestWriteErrorReporter, TestSparkKafkaOffsetValidator, TestSparkValidationContext, TestSparkStreamerValidatorUtils 70/70. TestStreamSync, TestHoodieStreamerUtils 51/51. Checkstyle / apache-rat: 0 violations on hudi-utilities and hudi-client-common. * fix(utilities): wrap and log validator failure on #18750 Address Danny's review comment on StreamSync.java:936. The Step 2 catch block now logs a contextual error and wraps the HoodieValidationException in a HoodieStreamerWriteException with the instant time, matching the rollback+wrap pattern used by Step 1 (error-table ROLLBACK_COMMIT) and Step 4 (write-error gate). The original validation exception is preserved as the cause. * fix(utilities): address hudi-agent review nits on #18750 - SuccessfulRecordCounter.Counts: rename getTotalErroredRecords() / totalErroredRecords to getTotalErrorRecords() / totalErrorRecords so the terminology matches WriteStatus.getTotalErrorRecords() (no past participle). Propagate the rename through StreamSync, SparkWriteErrorValidator Javadoc, and TestSuccessfulRecordCounter. - WriteErrorReporter and SparkWriteErrorValidator: replace Lombok @Slf4j with explicit `private static final Logger LOG = LoggerFactory.getLogger(...)` to match the streamer package style (StreamSync etc.). No behavior change. Tests: TestSuccessfulRecordCounter, TestSparkWriteErrorValidator, TestSparkValidationContext, TestSparkKafkaOffsetValidator, TestSparkStreamerValidatorUtils all pass (56/56). * fix(utilities): one-pass RDD aggregate in SuccessfulRecordCounter; clarify WARN_LOG+commitOnErrors error message * fix(utilities): restore StreamSync.sumRecordAndErrorCounts referenced by upstream TestStreamSyncWriteStatusValidation The merge resolution dropped the in-StreamSync helper added in upstream master because the migrated SuccessfulRecordCounter already has the equivalent one-pass aggregate. However the upstream merge also brought in TestStreamSyncWriteStatusValidation which calls StreamSync.sumRecordAndErrorCounts directly, breaking test compilation. Restore the helper as a @VisibleForTesting static method so the upstream test compiles. Production code paths use SuccessfulRecordCounter and are unaffected. --------- Co-authored-by: Xinli Shang <[email protected]> (cherry picked from commit 8490c968d9dc71b44d4ca907aa1a50ba3111407c) --- .../config/HoodiePreCommitValidatorConfig.java | 4 +- .../utilities/streamer/ErrorTableCommitter.java | 86 ++++++++ .../apache/hudi/utilities/streamer/StreamSync.java | 244 +++++++++------------ .../streamer/SuccessfulRecordCounter.java | 106 +++++++++ .../utilities/streamer/WriteErrorReporter.java | 69 ++++++ .../validator/SparkWriteErrorValidator.java | 136 ++++++++++++ .../streamer/TestErrorTableCommitter.java | 135 ++++++++++++ .../streamer/TestSuccessfulRecordCounter.java | 200 +++++++++++++++++ .../utilities/streamer/TestWriteErrorReporter.java | 99 +++++++++ .../validator/TestSparkWriteErrorValidator.java | 198 +++++++++++++++++ 10 files changed, 1132 insertions(+), 145 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodiePreCommitValidatorConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodiePreCommitValidatorConfig.java index 169494b7244a..f4999bc39e16 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodiePreCommitValidatorConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodiePreCommitValidatorConfig.java @@ -46,7 +46,9 @@ public class HoodiePreCommitValidatorConfig extends HoodieConfig { .withDocumentation("Comma separated list of class names that can be invoked to validate commit. " + "Available streaming offset validators: " + "org.apache.hudi.sink.validator.FlinkKafkaOffsetValidator (Flink Kafka), " - + "org.apache.hudi.utilities.streamer.validator.SparkKafkaOffsetValidator (Spark/HoodieStreamer Kafka)"); + + "org.apache.hudi.utilities.streamer.validator.SparkKafkaOffsetValidator (Spark/HoodieStreamer Kafka). " + + "Available write-error validators: " + + "org.apache.hudi.utilities.streamer.validator.SparkWriteErrorValidator (Spark/HoodieStreamer write errors)."); public static final String VALIDATOR_TABLE_VARIABLE = "<TABLE_NAME>"; public static final ConfigProperty<String> EQUALITY_SQL_QUERIES = ConfigProperty diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/ErrorTableCommitter.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/ErrorTableCommitter.java new file mode 100644 index 000000000000..1986e911a06b --- /dev/null +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/ErrorTableCommitter.java @@ -0,0 +1,86 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hudi.utilities.streamer; + +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.common.util.Option; + +import org.apache.spark.api.java.JavaRDD; + +import java.util.Objects; + +/** + * Commits the error-table side of a HoodieStreamer commit. + * + * <p>Two paths exist, mirroring the original {@code HoodieStreamerWriteStatusValidator} behavior:</p> + * <ul> + * <li><b>Unified write path</b> ({@code isErrorTableWriteUnificationEnabled=true}): commit the + * error-table write statuses produced alongside the data-table write.</li> + * <li><b>Legacy path</b>: invoke {@link BaseErrorTableWriter#upsertAndCommit(String, Option)} + * which performs both the upsert and the commit internally.</li> + * </ul> + * + * <p>This helper performs the commit and reports success/failure. It deliberately does <i>not</i> + * handle the {@code ROLLBACK_COMMIT} / {@code LOG_ERROR} failure strategies — that policy decision + * lives in {@code StreamSync.writeToSinkAndDoMetaSync()}, which understands the surrounding + * orchestration. Extracted from {@code HoodieStreamerWriteStatusValidator} as part of #18750.</p> + */ +public final class ErrorTableCommitter { + + private ErrorTableCommitter() { + } + + /** + * Commit the error-table writes for the given instant. + * + * @param errorTableWriter The configured error-table writer. Must not be null. + * @param errorTableWriteStatusRDDOpt Optional error-table write status RDD, populated when + * unification is enabled. Must not be null + * ({@link Option#empty()} if no RDD). + * @param isErrorTableWriteUnificationEnabled Whether unified-write mode is enabled. + * @param instantTime Instant being committed. + * @param latestCommittedInstant Optional latest completed instant, passed to legacy + * {@code upsertAndCommit}. Must not be null + * ({@link Option#empty()} if none). + * @return {@code true} if the error-table commit succeeded (or was a no-op); + * {@code false} if it failed and the caller must apply a failure-policy action. + */ + public static boolean commit(BaseErrorTableWriter<?> errorTableWriter, + Option<JavaRDD<WriteStatus>> errorTableWriteStatusRDDOpt, + boolean isErrorTableWriteUnificationEnabled, + String instantTime, + Option<String> latestCommittedInstant) { + Objects.requireNonNull(errorTableWriter, "errorTableWriter"); + Objects.requireNonNull(errorTableWriteStatusRDDOpt, "errorTableWriteStatusRDDOpt"); + Objects.requireNonNull(instantTime, "instantTime"); + Objects.requireNonNull(latestCommittedInstant, "latestCommittedInstant"); + + if (isErrorTableWriteUnificationEnabled) { + // In unification mode the error-table writes were produced upstream by the unified write + // path. Commit them here; nothing to do when the optional RDD is absent (true no-op). + if (errorTableWriteStatusRDDOpt.isPresent()) { + return errorTableWriter.commit(errorTableWriteStatusRDDOpt.get()); + } + return true; + } + // Legacy path: writer performs both upsert and commit internally. + return errorTableWriter.upsertAndCommit(instantTime, latestCommittedInstant); + } +} diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java index edeafb2755e3..9811fe9b0178 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java @@ -26,7 +26,6 @@ import org.apache.hudi.HoodieSchemaConversionUtils; import org.apache.hudi.HoodieSchemaUtils; import org.apache.hudi.HoodieSparkSqlWriter; import org.apache.hudi.HoodieSparkUtils; -import org.apache.hudi.callback.common.WriteStatusValidator; import org.apache.hudi.client.HoodieWriteResult; import org.apache.hudi.client.SparkRDDWriteClient; import org.apache.hudi.client.WriteStatus; @@ -40,7 +39,6 @@ import org.apache.hudi.common.config.HoodieStorageConfig; import org.apache.hudi.common.config.HoodieTimeGeneratorConfig; import org.apache.hudi.common.config.RecordMergeMode; import org.apache.hudi.common.config.TypedProperties; -import org.apache.hudi.common.data.HoodieData; import org.apache.hudi.common.model.DefaultHoodieRecordPayload; import org.apache.hudi.common.model.HoodieKey; import org.apache.hudi.common.model.HoodieRecord; @@ -77,7 +75,6 @@ import org.apache.hudi.config.HoodiePayloadConfig; import org.apache.hudi.config.HoodiePreCommitValidatorConfig; import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.config.metrics.HoodieMetricsConfig; -import org.apache.hudi.data.HoodieJavaRDD; import org.apache.hudi.exception.HoodieException; import org.apache.hudi.exception.HoodieIOException; import org.apache.hudi.exception.HoodieMetaSyncException; @@ -872,38 +869,113 @@ public class StreamSync implements Serializable, Closeable { Map<String, String> checkpointCommitMetadata = extractCheckpointMetadata(inputBatch, props, writeClient.getConfig().getWriteVersion().versionCode(), cfg); AtomicLong totalSuccessfulRecords = new AtomicLong(0); Option<String> latestCommittedInstant = getLatestCommittedInstant(); - WriteStatusValidator writeStatusValidator = new HoodieStreamerWriteStatusValidator(cfg.commitOnErrors, instantTime, - cfg, errorTableWriter, errorTableWriteStatusRDDOpt, errorWriteFailureStrategy, isErrorTableWriteUnificationEnabled, writeClient, latestCommittedInstant, - totalSuccessfulRecords); String commitActionType = CommitUtils.getCommitActionType(cfg.operation, HoodieTableType.valueOf(cfg.tableType)); - // Cache the RDD only when pre-commit validators are configured. Validators collect the RDD - // before commit, so without caching the same DAG would re-evaluate inside writeClient.commit(). - // When no validators are configured, commit consumes the RDD once and caching adds no value. - // shouldUnpersist is true only when we created the cache here (validators present and storage - // level was NONE), so the finally block knows to release it. - boolean validatorsConfigured = !StringUtils.isNullOrEmpty(props.getString( - HoodiePreCommitValidatorConfig.VALIDATOR_CLASS_NAMES.key(), - HoodiePreCommitValidatorConfig.VALIDATOR_CLASS_NAMES.defaultValue())); - boolean shouldUnpersist = validatorsConfigured && writeStatusRDD.getStorageLevel().equals(StorageLevel.NONE()); + // Pre-commit orchestration (issue #18750): the legacy HoodieStreamerWriteStatusValidator + // ran inside writeClient.commit() via the WriteStatusValidator callback and combined three + // concerns — count records, commit the error table, and gate on write errors. Each is now + // an explicit step here before writeClient.commit(), so the writer no longer receives a + // callback. Step order is deliberate (see comments below). + // + // The RDD is cached once and the write statuses are collected once on the driver. Both the + // count/error-logging steps and writeClient.commit() consume the materialized partitions + // rather than re-evaluating the upstream DAG. shouldUnpersist tracks whether we engaged the + // cache here so the finally block knows to release it. + boolean shouldUnpersist = writeStatusRDD.getStorageLevel().equals(StorageLevel.NONE()); if (shouldUnpersist) { writeStatusRDD.cache(); } boolean success; try { + List<WriteStatus> writeStatuses = writeStatusRDD.collect(); + boolean validatorsConfigured = !StringUtils.isNullOrEmpty(props.getString( + HoodiePreCommitValidatorConfig.VALIDATOR_CLASS_NAMES.key(), + HoodiePreCommitValidatorConfig.VALIDATOR_CLASS_NAMES.defaultValue())); + + // Step 1: Commit the error table BEFORE running validators or the write-error gate. + // Error records captured here are a genuine artifact of the write attempt and should + // survive even when a validator later blocks the data-table commit (otherwise the + // operator loses the captured errors and the next run has nothing to triage against). + // Latent design quirk (preserved from HSWSV): if error-table commit succeeds and any + // subsequent step fails (Step 2 validator including the offset validator, Step 4 gate, + // or writeClient.commit), the error table will have a committed instant for a data-table + // instant that never lands. Downstream consumers of the error table should tolerate this + // divergence. + if (errorTableWriter.isPresent()) { + boolean errorTableSuccess = ErrorTableCommitter.commit(errorTableWriter.get(), + errorTableWriteStatusRDDOpt, isErrorTableWriteUnificationEnabled, instantTime, + latestCommittedInstant); + if (!errorTableSuccess) { + switch (errorWriteFailureStrategy) { + case ROLLBACK_COMMIT: + // Roll back the inflight data-table instant so it doesn't leak under LAZY + // failed-writes cleanup policy (preserves HSWSV behavior). + writeClient.rollback(instantTime); + throw new HoodieStreamerWriteException("Error table commit failed for instant " + instantTime); + case LOG_ERROR: + LOG.error("Error table write failed for instant {}", instantTime); + break; + default: + throw new HoodieStreamerWriteException("Write failure strategy not implemented for " + errorWriteFailureStrategy); + } + } + } + + // Step 2: Run user-configured pre-commit validators (offset, custom, and the opt-in + // SparkWriteErrorValidator). Validators are intentionally stronger than commitOnErrors + // — a failure here aborts the data-table commit regardless of the gate in Step 4. + // Roll back the inflight data-table instant on validation failure so it doesn't leak + // under LAZY failed-writes cleanup policy (consistent with Step 1 ROLLBACK_COMMIT and + // the Step 4 gate below). Error-table records already committed in Step 1 are preserved + // by design — see Step 1's latent-quirk note. if (validatorsConfigured) { - List<WriteStatus> writeStatuses = writeStatusRDD.collect(); - - // Run pre-commit streaming offset validators (if configured). - // Placement before writeClient.commit() is intentional: offset validation is a stronger - // guard than commitOnErrors — if offset deviation indicates potential data loss, the commit - // must be prevented regardless of the commitOnErrors policy. - SparkStreamerValidatorUtils.runValidators(props, instantTime, writeStatuses, - checkpointCommitMetadata, metaClient); + try { + SparkStreamerValidatorUtils.runValidators(props, instantTime, writeStatuses, + checkpointCommitMetadata, metaClient); + } catch (HoodieValidationException e) { + LOG.error("Pre-commit validators failed for instant {}", instantTime, e); + writeClient.rollback(instantTime); + throw new HoodieStreamerWriteException("Pre-commit validators failed for instant " + instantTime, e); + } + } + + // Step 3: Count records. Drives the runMetaSync() decision below the try/finally. + SuccessfulRecordCounter.Counts counts = SuccessfulRecordCounter.compute( + writeStatuses, errorTableWriteStatusRDDOpt, isErrorTableWriteUnificationEnabled); + totalSuccessfulRecords.set(counts.getTotalSuccessfulRecords()); + LOG.info("instantTime={}, totalRecords={}, totalErrorRecords={}, totalSuccessfulRecords={}", + instantTime, counts.getTotalRecords(), counts.getTotalErrorRecords(), + counts.getTotalSuccessfulRecords()); + if (counts.getTotalRecords() == 0) { + LOG.info("No new data, perform empty commit."); + } + + // Step 4: Apply the legacy HSWSV write-error gate. + // commitOnErrors=false (default): any error -> log top N + fail. + // commitOnErrors=true: log a warning, proceed to commit. + // This gate is redundant with SparkWriteErrorValidator when that validator is configured + // with failure.policy=FAIL — both will reject the same commits. The redundancy is + // intentional: the gate preserves HSWSV's default behavior for users who do not configure + // any validators, while the validator gives users running multiple validators a unified + // failure-policy story. + if (counts.hasErrors()) { + if (cfg.commitOnErrors) { + LOG.warn("Some records failed to be merged but forcing commit since commitOnErrors set. Errors/Total={}/{}", + counts.getTotalErrorRecords(), counts.getTotalRecords()); + } else { + LOG.error("Delta Sync found errors when writing. Errors/Total={}/{}", + counts.getTotalErrorRecords(), counts.getTotalRecords()); + WriteErrorReporter.logTopErrors(writeStatuses); + // Roll back the inflight data-table instant so it doesn't leak under LAZY + // failed-writes cleanup policy (preserves HSWSV behavior). + writeClient.rollback(instantTime); + throw new HoodieStreamerWriteException("Commit " + instantTime + " has write errors and commitOnErrors=false"); + } } - success = writeClient.commit(instantTime, writeStatusRDD, Option.of(checkpointCommitMetadata), commitActionType, partitionToReplacedFileIds, Option.empty(), - Option.of(writeStatusValidator)); + // Step 5: Commit. No WriteStatusValidator callback — all checks are above. + success = writeClient.commit(instantTime, writeStatusRDD, Option.of(checkpointCommitMetadata), + commitActionType, partitionToReplacedFileIds, Option.empty()); } finally { if (shouldUnpersist) { writeStatusRDD.unpersist(); @@ -1407,17 +1479,9 @@ public class StreamSync implements Serializable, Closeable { * Sums {@link WriteStatus#getTotalRecords()} and {@link WriteStatus#getTotalErrorRecords()} over the * given RDD in a single Spark action, returned as a {@code (totalRecords, totalErroredRecords)} tuple. * - * <p>Folding both counters into one {@code aggregate} pass avoids re-deserializing every cached - * {@link WriteStatus} block a second time. Issuing two separate {@code mapToDouble(...).sum()} - * actions on the persisted error-table {@code WriteStatus} RDD doubles Kryo deserialization of - * cached partitions during commit validation and adds gratuitous heap pressure on memory-strained - * executors. - * - * <p>{@code aggregate} is used (instead of {@code mapPartitions(...).reduce(...)}) so that a - * 0-partition RDD (e.g. {@code sc.emptyRDD()}, which {@link BaseErrorTableWriter#upsert} can - * return for an empty commit) returns {@code (0L, 0L)} rather than raising - * {@code UnsupportedOperationException} as {@code reduce} would. The mutable {@code long[2]} - * accumulator keeps per-record allocations at zero. + * <p>{@code aggregate} (not {@code reduce}) is used so a 0-partition RDD returns {@code (0L, 0L)} + * instead of raising {@code UnsupportedOperationException}; the mutable {@code long[2]} accumulator + * avoids per-record allocations. */ @VisibleForTesting static Tuple2<Long, Long> sumRecordAndErrorCounts(JavaRDD<WriteStatus> writeStatuses) { @@ -1435,112 +1499,4 @@ public class StreamSync implements Serializable, Closeable { }); return new Tuple2<>(counts[0], counts[1]); } - - /** - * WriteStatus Validator for commits to hoodie streamer data table. - * The writes to error table is taken care as well. - */ - static class HoodieStreamerWriteStatusValidator implements WriteStatusValidator { - - private final boolean commitOnErrors; - private final String instantTime; - private final HoodieStreamer.Config cfg; - private final Option<BaseErrorTableWriter> errorTableWriter; - private final Option<JavaRDD<WriteStatus>> errorTableWriteStatusRDDOpt; - private final HoodieErrorTableConfig.ErrorWriteFailureStrategy errorWriteFailureStrategy; - private final boolean isErrorTableWriteUnificationEnabled; - private final SparkRDDWriteClient writeClient; - private final Option<String> latestCommittedInstant; - private final AtomicLong totalSuccessfulRecords; - - HoodieStreamerWriteStatusValidator(boolean commitOnErrors, - String instantTime, - HoodieStreamer.Config cfg, - Option<BaseErrorTableWriter> errorTableWriter, - Option<JavaRDD<WriteStatus>> errorTableWriteStatusRDDOpt, - HoodieErrorTableConfig.ErrorWriteFailureStrategy errorWriteFailureStrategy, - boolean isErrorTableWriteUnificationEnabled, - SparkRDDWriteClient writeClient, - Option<String> latestCommittedInstant, - AtomicLong totalSuccessfulRecords) { - this.commitOnErrors = commitOnErrors; - this.instantTime = instantTime; - this.cfg = cfg; - this.errorTableWriter = errorTableWriter; - this.errorTableWriteStatusRDDOpt = errorTableWriteStatusRDDOpt; - this.errorWriteFailureStrategy = errorWriteFailureStrategy; - this.isErrorTableWriteUnificationEnabled = isErrorTableWriteUnificationEnabled; - this.writeClient = writeClient; - this.latestCommittedInstant = latestCommittedInstant; - this.totalSuccessfulRecords = totalSuccessfulRecords; - } - - @Override - public boolean validate(long tableTotalRecords, long tableTotalErroredRecords, Option<HoodieData<WriteStatus>> writeStatusesOpt) { - - long totalRecords = tableTotalRecords; - long totalErroredRecords = tableTotalErroredRecords; - if (isErrorTableWriteUnificationEnabled && errorTableWriteStatusRDDOpt.isPresent()) { - Tuple2<Long, Long> errorTableCounts = sumRecordAndErrorCounts(errorTableWriteStatusRDDOpt.get()); - totalRecords += errorTableCounts._1; - totalErroredRecords += errorTableCounts._2; - } - long totalSuccessfulRecords = totalRecords - totalErroredRecords; - this.totalSuccessfulRecords.set(totalSuccessfulRecords); - LOG.info("instantTime={}, totalRecords={}, totalErrorRecords={}, totalSuccessfulRecords={}", - instantTime, totalRecords, totalErroredRecords, totalSuccessfulRecords); - if (totalRecords == 0) { - LOG.info("No new data, perform empty commit."); - } - boolean hasErrorRecords = totalErroredRecords > 0; - if (!hasErrorRecords || commitOnErrors) { - if (hasErrorRecords) { - LOG.warn("Some records failed to be merged but forcing commit since commitOnErrors set. Errors/Total={}/{}", - totalErroredRecords, totalRecords); - } - } - - if (errorTableWriter.isPresent()) { - boolean errorTableSuccess = true; - // Commit the error events triggered so far to the error table - if (isErrorTableWriteUnificationEnabled && errorTableWriteStatusRDDOpt.isPresent()) { - errorTableSuccess = errorTableWriter.get().commit(errorTableWriteStatusRDDOpt.get()); - } else if (!isErrorTableWriteUnificationEnabled) { - errorTableSuccess = errorTableWriter.get().upsertAndCommit(instantTime, latestCommittedInstant); - } - if (!errorTableSuccess) { - switch (errorWriteFailureStrategy) { - case ROLLBACK_COMMIT: - LOG.info("Commit " + instantTime + " failed!"); - writeClient.rollback(instantTime); - throw new HoodieStreamerWriteException("Error table commit failed"); - case LOG_ERROR: - LOG.error("Error Table write failed for instant " + instantTime); - break; - default: - throw new HoodieStreamerWriteException("Write failure strategy not implemented for " + errorWriteFailureStrategy); - } - } - } - boolean canProceed = !hasErrorRecords || commitOnErrors; - if (canProceed) { - return canProceed; - } else { - LOG.error("Delta Sync found errors when writing. Errors/Total=" + totalErroredRecords + "/" + totalRecords); - LOG.error("Printing out the top 100 errors"); - ValidationUtils.checkArgument(writeStatusesOpt.isPresent(), "RDD <WriteStatus> is expected to be present when there are errors "); - HoodieJavaRDD.getJavaRDD(writeStatusesOpt.get()).filter(WriteStatus::hasErrors).take(100).forEach(writeStatus -> { - LOG.error("Global error " + writeStatus.getGlobalError()); - if (!writeStatus.getErrors().isEmpty()) { - writeStatus.getErrors().forEach((k,v) -> { - LOG.trace("Error for key %s : %s ", k, v); - }); - } - }); - // Rolling back instant - writeClient.rollback(instantTime); - throw new HoodieStreamerWriteException("Commit " + instantTime + " failed and rolled-back !"); - } - } - } } diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SuccessfulRecordCounter.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SuccessfulRecordCounter.java new file mode 100644 index 000000000000..01439249d760 --- /dev/null +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SuccessfulRecordCounter.java @@ -0,0 +1,106 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hudi.utilities.streamer; + +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.common.util.Option; + +import org.apache.spark.api.java.JavaRDD; + +import java.util.List; +import java.util.Objects; + +/** + * Computes record counts for a HoodieStreamer commit, summing across the data-table + * write statuses and (optionally) the error-table write statuses when error-table + * write unification is enabled. + * + * <p>Extracted from {@code HoodieStreamerWriteStatusValidator} (issue #18750) so the + * counting logic can be invoked from the explicit pre-commit orchestration in + * {@code StreamSync} without going through the {@code WriteStatusValidator} callback.</p> + */ +public final class SuccessfulRecordCounter { + + private SuccessfulRecordCounter() { + } + + /** + * Compute total / errored / successful record counts from a pre-collected list of write statuses. + * + * @param dataTableWriteStatuses Pre-collected data-table write statuses. Must not be null. + * @param errorTableWriteStatusRDDOpt Optional error-table write status RDD; only consulted + * when unification is enabled. Must not be null + * ({@link Option#empty()} when no error table). + * @param isErrorTableWriteUnificationEnabled Whether error-table records contribute to the totals. + * @return immutable {@link Counts} snapshot. + */ + public static Counts compute(List<WriteStatus> dataTableWriteStatuses, + Option<JavaRDD<WriteStatus>> errorTableWriteStatusRDDOpt, + boolean isErrorTableWriteUnificationEnabled) { + Objects.requireNonNull(dataTableWriteStatuses, "dataTableWriteStatuses"); + Objects.requireNonNull(errorTableWriteStatusRDDOpt, "errorTableWriteStatusRDDOpt"); + + long totalRecords = 0L; + long totalErrorRecords = 0L; + for (WriteStatus ws : dataTableWriteStatuses) { + totalRecords += ws.getTotalRecords(); + totalErrorRecords += ws.getTotalErrorRecords(); + } + if (isErrorTableWriteUnificationEnabled && errorTableWriteStatusRDDOpt.isPresent()) { + JavaRDD<WriteStatus> errorRdd = errorTableWriteStatusRDDOpt.get(); + long[] sums = errorRdd.aggregate( + new long[]{0L, 0L}, + (acc, ws) -> new long[]{acc[0] + ws.getTotalRecords(), acc[1] + ws.getTotalErrorRecords()}, + (a, b) -> new long[]{a[0] + b[0], a[1] + b[1]}); + totalRecords += sums[0]; + totalErrorRecords += sums[1]; + } + return new Counts(totalRecords, totalErrorRecords); + } + + /** Immutable count snapshot. */ + public static final class Counts { + public static final Counts ZERO = new Counts(0L, 0L); + + private final long totalRecords; + private final long totalErrorRecords; + + public Counts(long totalRecords, long totalErrorRecords) { + this.totalRecords = totalRecords; + this.totalErrorRecords = totalErrorRecords; + } + + public long getTotalRecords() { + return totalRecords; + } + + public long getTotalErrorRecords() { + return totalErrorRecords; + } + + public long getTotalSuccessfulRecords() { + return totalRecords - totalErrorRecords; + } + + public boolean hasErrors() { + return totalErrorRecords > 0; + } + } +} diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/WriteErrorReporter.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/WriteErrorReporter.java new file mode 100644 index 000000000000..2cb7c2fe6050 --- /dev/null +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/WriteErrorReporter.java @@ -0,0 +1,69 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hudi.utilities.streamer; + +import org.apache.hudi.client.WriteStatus; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; + +/** + * Logs the first N errored write statuses so the operator can triage a failed commit. + * + * <p>Extracted from {@code HoodieStreamerWriteStatusValidator} (#18750).</p> + */ +public final class WriteErrorReporter { + + private static final Logger LOG = LoggerFactory.getLogger(WriteErrorReporter.class); + + private static final int DEFAULT_MAX_ERRORS = 100; + + private WriteErrorReporter() { + } + + public static void logTopErrors(List<WriteStatus> writeStatuses) { + logTopErrors(writeStatuses, DEFAULT_MAX_ERRORS); + } + + /** + * Log up to {@code maxErrors} errored write statuses from a pre-collected list. Each errored + * status's global error is logged at ERROR; per-key errors are logged at TRACE. The header + * line is INFO. No-op when the list is null or {@code maxErrors <= 0}. + */ + public static void logTopErrors(List<WriteStatus> writeStatuses, int maxErrors) { + if (writeStatuses == null || maxErrors <= 0) { + return; + } + LOG.info("Printing out the top {} errored write statuses", maxErrors); + writeStatuses.stream() + .filter(WriteStatus::hasErrors) + .limit(maxErrors) + .forEach(WriteErrorReporter::logOne); + } + + private static void logOne(WriteStatus writeStatus) { + LOG.error("Global error: {}", writeStatus.getGlobalError()); + if (!writeStatus.getErrors().isEmpty()) { + writeStatus.getErrors().forEach((k, v) -> LOG.trace("Error for key {} : {}", k, v)); + } + } +} diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/validator/SparkWriteErrorValidator.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/validator/SparkWriteErrorValidator.java new file mode 100644 index 000000000000..07591b5b7973 --- /dev/null +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/validator/SparkWriteErrorValidator.java @@ -0,0 +1,136 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hudi.utilities.streamer.validator; + +import org.apache.hudi.client.validator.BasePreCommitValidator; +import org.apache.hudi.client.validator.ValidationContext; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.config.HoodiePreCommitValidatorConfig; +import org.apache.hudi.config.HoodiePreCommitValidatorConfig.ValidationFailurePolicy; +import org.apache.hudi.exception.HoodieValidationException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Arrays; + +/** + * Pre-commit validator that fails the commit when records failed to write. + * + * <p>Equivalent of the legacy {@code HoodieStreamerWriteStatusValidator}'s boolean error check + * ({@code hasErrorRecords = totalErrorRecords > 0}), wired through the pre-commit validator + * framework (issue #18750). Pure validation: no side effects (no error-table commit, no + * top-100 error logging, no instant rollback). Those side effects are handled separately by + * {@code StreamSync}'s pre-commit orchestration.</p> + * + * <p><b>Relationship with the inline write-error gate in {@code StreamSync}:</b> the default + * commit path in {@code StreamSync} already applies an equivalent error check via the + * {@code commitOnErrors} flag. This validator exists so that users running multiple validators + * (e.g. write-error + offset checks) can express a unified pass/fail story through a single + * {@code failure.policy} knob. Enabling this validator while leaving {@code commitOnErrors=false} + * means both checks run and either can block the commit — they are intentionally not mutually + * exclusive.</p> + * + * <p>Behavior mapping from the legacy HSWSV (data-table only — see caveat below):</p> + * <ul> + * <li>{@code commitOnErrors = false} (HSWSV default) ↔ {@code failure.policy = FAIL}</li> + * <li>{@code commitOnErrors = true} ↔ {@code failure.policy = WARN_LOG}</li> + * </ul> + * + * <p><b>Unification caveat:</b> when {@code hoodie.errortable.write.unification.enabled=true}, + * HSWSV's error check summed errors across <em>both</em> the data-table and the error-table write + * statuses. This validator only sees data-table stats via {@link ValidationContext} (specifically + * {@link ValidationContext#getTotalWriteErrors()} / {@link ValidationContext#getTotalRecordsWritten()}, + * which are derived from {@code HoodieWriteStat} on the data table). Under unification it is + * therefore strictly weaker than HSWSV: error-table-only errors will not trip this validator. Users + * who rely on unified error counts should keep the inline {@code commitOnErrors} gate in + * {@code StreamSync} enabled (i.e. leave {@code --commit-on-errors} off), which still consults + * the unified count via {@code SuccessfulRecordCounter}.</p> + * + * <p>Configuration:</p> + * <ul> + * <li>{@code hoodie.precommit.validators}: Include + * {@code org.apache.hudi.utilities.streamer.validator.SparkWriteErrorValidator}</li> + * <li>{@code hoodie.precommit.validators.failure.policy}: FAIL (default) or WARN_LOG</li> + * </ul> + * + * <p>Like {@link SparkKafkaOffsetValidator}, this class extends {@link BasePreCommitValidator} + * and must be invoked via {@link SparkStreamerValidatorUtils} — not {@code SparkValidatorUtils}, + * which expects a different constructor signature.</p> + */ +public class SparkWriteErrorValidator extends BasePreCommitValidator { + + private static final Logger LOG = LoggerFactory.getLogger(SparkWriteErrorValidator.class); + + private final ValidationFailurePolicy failurePolicy; + + public SparkWriteErrorValidator(TypedProperties config) { + super(config); + String policyStr = config.getString( + HoodiePreCommitValidatorConfig.VALIDATION_FAILURE_POLICY.key(), + HoodiePreCommitValidatorConfig.VALIDATION_FAILURE_POLICY.defaultValue()); + try { + this.failurePolicy = ValidationFailurePolicy.valueOf(policyStr); + } catch (IllegalArgumentException e) { + throw new HoodieValidationException(String.format( + "Invalid value '%s' for %s. Allowed values: %s.", + policyStr, + HoodiePreCommitValidatorConfig.VALIDATION_FAILURE_POLICY.key(), + Arrays.toString(ValidationFailurePolicy.values())), e); + } + } + + @Override + public void validateWithMetadata(ValidationContext context) throws HoodieValidationException { + long totalErrors = context.getTotalWriteErrors(); + long totalRecordsWritten = context.getTotalRecordsWritten(); + // Total considered for the commit = successfully-written + failed. HSWSV computed this from + // the raw WriteStatus RDD; we derive the equivalent from HoodieWriteStat fields exposed by + // ValidationContext. + long totalRecords = totalRecordsWritten + totalErrors; + + if (totalRecords == 0) { + // Empty commit (mirrors HSWSV "No new data, perform empty commit."). + LOG.info("Empty commit (no records written, no errors). Skipping write-error validation " + + "for instant {}.", context.getInstantTime()); + return; + } + + if (totalErrors == 0) { + LOG.info("Write-error validation passed for instant {}: 0 errors out of {} records.", + context.getInstantTime(), totalRecords); + return; + } + + String errorMsg = String.format( + "Write-error validation failed for instant %s. " + + "Errors: %d, Total: %d. " + + "To allow commits despite write errors, both %s=WARN_LOG (bypasses this validator) " + + "and --commit-on-errors (bypasses the StreamSync inline gate) are required.", + context.getInstantTime(), totalErrors, totalRecords, + HoodiePreCommitValidatorConfig.VALIDATION_FAILURE_POLICY.key()); + + if (failurePolicy == ValidationFailurePolicy.WARN_LOG) { + LOG.warn("{} (failure policy is WARN_LOG, commit will proceed)", errorMsg); + } else { + throw new HoodieValidationException(errorMsg); + } + } +} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestErrorTableCommitter.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestErrorTableCommitter.java new file mode 100644 index 000000000000..429807974342 --- /dev/null +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestErrorTableCommitter.java @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hudi.utilities.streamer; + +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.common.util.Option; + +import org.apache.spark.api.java.JavaRDD; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests for {@link ErrorTableCommitter} covering the two write paths (unification on/off), + * the success/failure pass-through contract, and the no-op when no RDD is present. + */ +public class TestErrorTableCommitter { + + private static final String INSTANT = "20260520120000000"; + + @SuppressWarnings("unchecked") + private static JavaRDD<WriteStatus> rdd() { + return (JavaRDD<WriteStatus>) Mockito.mock(JavaRDD.class); + } + + @Test + public void testUnificationCommitSuccess() { + BaseErrorTableWriter<?> writer = Mockito.mock(BaseErrorTableWriter.class); + JavaRDD<WriteStatus> rdd = rdd(); + Mockito.when(writer.commit(rdd)).thenReturn(true); + + boolean result = ErrorTableCommitter.commit(writer, Option.of(rdd), true, INSTANT, Option.empty()); + + assertTrue(result); + Mockito.verify(writer).commit(rdd); + Mockito.verify(writer, Mockito.never()).upsertAndCommit(Mockito.any(), Mockito.any()); + } + + @Test + public void testUnificationCommitFailurePropagates() { + BaseErrorTableWriter<?> writer = Mockito.mock(BaseErrorTableWriter.class); + JavaRDD<WriteStatus> rdd = rdd(); + Mockito.when(writer.commit(rdd)).thenReturn(false); + + boolean result = ErrorTableCommitter.commit(writer, Option.of(rdd), true, INSTANT, Option.empty()); + + assertFalse(result); + } + + @Test + public void testUnificationWithoutRddIsNoOpAndReturnsTrue() { + BaseErrorTableWriter<?> writer = Mockito.mock(BaseErrorTableWriter.class); + + boolean result = ErrorTableCommitter.commit(writer, Option.empty(), true, INSTANT, Option.empty()); + + assertTrue(result); + Mockito.verifyNoInteractions(writer); + } + + @Test + public void testLegacyPathUsesUpsertAndCommit() { + BaseErrorTableWriter<?> writer = Mockito.mock(BaseErrorTableWriter.class); + Option<String> latest = Option.of("20260520115959000"); + Mockito.when(writer.upsertAndCommit(INSTANT, latest)).thenReturn(true); + + boolean result = ErrorTableCommitter.commit(writer, Option.empty(), false, INSTANT, latest); + + assertTrue(result); + Mockito.verify(writer).upsertAndCommit(INSTANT, latest); + Mockito.verify(writer, Mockito.never()).commit(Mockito.any()); + } + + @Test + public void testLegacyPathFailurePropagates() { + BaseErrorTableWriter<?> writer = Mockito.mock(BaseErrorTableWriter.class); + Mockito.when(writer.upsertAndCommit(Mockito.anyString(), Mockito.any())).thenReturn(false); + + boolean result = ErrorTableCommitter.commit(writer, Option.empty(), false, INSTANT, Option.empty()); + + assertFalse(result); + } + + @Test + public void testLegacyPathIgnoresRddEvenWhenProvided() { + // When unification is OFF, the RDD must not be touched even if accidentally passed in. + BaseErrorTableWriter<?> writer = Mockito.mock(BaseErrorTableWriter.class); + JavaRDD<WriteStatus> rdd = rdd(); + Mockito.when(writer.upsertAndCommit(Mockito.anyString(), Mockito.any())).thenReturn(true); + + ErrorTableCommitter.commit(writer, Option.of(rdd), false, INSTANT, Option.empty()); + + Mockito.verify(writer, Mockito.never()).commit(Mockito.any()); + Mockito.verify(writer).upsertAndCommit(INSTANT, Option.empty()); + } + + @Test + public void testNullWriterRejected() { + assertThrows(NullPointerException.class, () -> + ErrorTableCommitter.commit(null, Option.empty(), false, INSTANT, Option.empty())); + } + + @Test + public void testNullRddOptionRejected() { + BaseErrorTableWriter<?> writer = Mockito.mock(BaseErrorTableWriter.class); + assertThrows(NullPointerException.class, () -> + ErrorTableCommitter.commit(writer, null, false, INSTANT, Option.empty())); + } + + @Test + public void testNullInstantRejected() { + BaseErrorTableWriter<?> writer = Mockito.mock(BaseErrorTableWriter.class); + assertThrows(NullPointerException.class, () -> + ErrorTableCommitter.commit(writer, Option.empty(), false, null, Option.empty())); + } +} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestSuccessfulRecordCounter.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestSuccessfulRecordCounter.java new file mode 100644 index 000000000000..1484a1d3e58e --- /dev/null +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestSuccessfulRecordCounter.java @@ -0,0 +1,200 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hudi.utilities.streamer; + +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.common.util.Option; + +import org.apache.spark.SparkConf; +import org.apache.spark.api.java.JavaRDD; +import org.apache.spark.api.java.JavaSparkContext; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.util.Arrays; +import java.util.Collections; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests for {@link SuccessfulRecordCounter}. Covers the driver-side (collected list) + * counting and error-table unification paths, plus null safety on public entry points. + */ +public class TestSuccessfulRecordCounter { + + private static JavaSparkContext jsc; + + @BeforeAll + public static void setUp() { + SparkConf conf = new SparkConf() + .setAppName("TestSuccessfulRecordCounter") + .setMaster("local[2]"); + jsc = new JavaSparkContext(conf); + } + + @AfterAll + public static void tearDown() { + if (jsc != null) { + jsc.close(); + } + } + + @Test + public void testEmptyInputReturnsZero() { + SuccessfulRecordCounter.Counts counts = SuccessfulRecordCounter.compute( + Collections.emptyList(), Option.empty(), false); + + assertEquals(0L, counts.getTotalRecords()); + assertEquals(0L, counts.getTotalErrorRecords()); + assertEquals(0L, counts.getTotalSuccessfulRecords()); + assertFalse(counts.hasErrors()); + } + + @Test + public void testSingleWriteStatusNoErrors() { + WriteStatus ws = Mockito.mock(WriteStatus.class); + Mockito.when(ws.getTotalRecords()).thenReturn(1000L); + Mockito.when(ws.getTotalErrorRecords()).thenReturn(0L); + + SuccessfulRecordCounter.Counts counts = SuccessfulRecordCounter.compute( + Collections.singletonList(ws), Option.empty(), false); + + assertEquals(1000L, counts.getTotalRecords()); + assertEquals(0L, counts.getTotalErrorRecords()); + assertEquals(1000L, counts.getTotalSuccessfulRecords()); + assertFalse(counts.hasErrors()); + } + + @Test + public void testMultipleWriteStatusesAreSummed() { + WriteStatus a = Mockito.mock(WriteStatus.class); + Mockito.when(a.getTotalRecords()).thenReturn(100L); + Mockito.when(a.getTotalErrorRecords()).thenReturn(5L); + + WriteStatus b = Mockito.mock(WriteStatus.class); + Mockito.when(b.getTotalRecords()).thenReturn(200L); + Mockito.when(b.getTotalErrorRecords()).thenReturn(10L); + + SuccessfulRecordCounter.Counts counts = SuccessfulRecordCounter.compute( + Arrays.asList(a, b), Option.empty(), false); + + assertEquals(300L, counts.getTotalRecords()); + assertEquals(15L, counts.getTotalErrorRecords()); + assertEquals(285L, counts.getTotalSuccessfulRecords()); + assertTrue(counts.hasErrors()); + } + + @Test + public void testUnificationDisabledIgnoresErrorTableRdd() { + WriteStatus ws = Mockito.mock(WriteStatus.class); + Mockito.when(ws.getTotalRecords()).thenReturn(50L); + Mockito.when(ws.getTotalErrorRecords()).thenReturn(2L); + + // Even if an error-table RDD is provided, unification=false means it must be ignored. + // Pass an "always throws" mock to prove the helper never touches it. + @SuppressWarnings("unchecked") + JavaRDD<WriteStatus> rdd = (JavaRDD<WriteStatus>) Mockito.mock(JavaRDD.class); + Mockito.when(rdd.mapToDouble(Mockito.any())).thenThrow(new AssertionError("RDD must not be consulted when unification is disabled")); + Mockito.when(rdd.map(Mockito.any())).thenThrow(new AssertionError("RDD must not be consulted when unification is disabled")); + + SuccessfulRecordCounter.Counts counts = SuccessfulRecordCounter.compute( + Collections.singletonList(ws), Option.of(rdd), false); + + assertEquals(50L, counts.getTotalRecords()); + assertEquals(2L, counts.getTotalErrorRecords()); + assertEquals(48L, counts.getTotalSuccessfulRecords()); + } + + @Test + public void testHasErrorsBoundary() { + WriteStatus ws = Mockito.mock(WriteStatus.class); + Mockito.when(ws.getTotalRecords()).thenReturn(10L); + Mockito.when(ws.getTotalErrorRecords()).thenReturn(1L); + + SuccessfulRecordCounter.Counts counts = SuccessfulRecordCounter.compute( + Collections.singletonList(ws), Option.empty(), false); + + assertTrue(counts.hasErrors()); + } + + // ========== Unification path (real Spark) ========== + + @Test + public void testUnificationEnabledSumsErrorTable() { + WriteStatus dataA = stat(100L, 5L); + WriteStatus dataB = stat(200L, 10L); + WriteStatus errA = stat(50L, 50L); + WriteStatus errB = stat(25L, 25L); + JavaRDD<WriteStatus> errorRdd = jsc.parallelize(Arrays.asList(errA, errB)); + + SuccessfulRecordCounter.Counts counts = SuccessfulRecordCounter.compute( + Arrays.asList(dataA, dataB), Option.of(errorRdd), true); + + assertEquals(375L, counts.getTotalRecords()); // 100 + 200 + 50 + 25 + assertEquals(90L, counts.getTotalErrorRecords()); // 5 + 10 + 50 + 25 + assertEquals(285L, counts.getTotalSuccessfulRecords()); + assertTrue(counts.hasErrors()); + } + + // ========== RDD-based path (real Spark) ========== + + @Test + public void testUnificationWithRealSparkErrorRdd() { + WriteStatus dataA = stat(100L, 5L); + JavaRDD<WriteStatus> errorRdd = jsc.parallelize(Collections.singletonList(stat(50L, 50L))); + + SuccessfulRecordCounter.Counts counts = SuccessfulRecordCounter.compute( + Collections.singletonList(dataA), Option.of(errorRdd), true); + + assertEquals(150L, counts.getTotalRecords()); + assertEquals(55L, counts.getTotalErrorRecords()); + assertEquals(95L, counts.getTotalSuccessfulRecords()); + } + + // ========== Null safety ========== + + @Test + public void testNullDataTableListRejected() { + assertThrows(NullPointerException.class, () -> + SuccessfulRecordCounter.compute(null, Option.empty(), false)); + } + + @Test + public void testNullErrorTableOptionRejected() { + assertThrows(NullPointerException.class, () -> + SuccessfulRecordCounter.compute(Collections.emptyList(), null, false)); + } + + // ========== Helper ========== + + private static WriteStatus stat(long totalRecords, long totalErrorRecords) { + // Use a real WriteStatus so it serializes for Spark closures (Mockito mocks are not Serializable). + // @Data on WriteStatus exposes setters for totalRecords/totalErrorRecords. + WriteStatus ws = new WriteStatus(false, 0.0); + ws.setTotalRecords(totalRecords); + ws.setTotalErrorRecords(totalErrorRecords); + return ws; + } +} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestWriteErrorReporter.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestWriteErrorReporter.java new file mode 100644 index 000000000000..d13378edb74e --- /dev/null +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestWriteErrorReporter.java @@ -0,0 +1,99 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hudi.utilities.streamer; + +import org.apache.hudi.client.WriteStatus; + +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; + +/** + * Tests for {@link WriteErrorReporter}. Verifies the no-op contract for null/empty inputs and + * that the List overload short-circuits without touching a Spark RDD. Logging output itself is + * intentionally not asserted — the value of the logger is human triage, not test assertions. + */ +public class TestWriteErrorReporter { + + @Test + public void testNullListIsNoOp() { + assertDoesNotThrow(() -> WriteErrorReporter.logTopErrors((List<WriteStatus>) null)); + assertDoesNotThrow(() -> WriteErrorReporter.logTopErrors((List<WriteStatus>) null, 10)); + } + + @Test + public void testEmptyListIsNoOp() { + assertDoesNotThrow(() -> WriteErrorReporter.logTopErrors(Collections.emptyList())); + } + + @Test + public void testZeroMaxErrorsIsNoOp() { + WriteStatus ws = errored("global err"); + assertDoesNotThrow(() -> WriteErrorReporter.logTopErrors(Collections.singletonList(ws), 0)); + assertDoesNotThrow(() -> WriteErrorReporter.logTopErrors(Collections.singletonList(ws), -5)); + } + + @Test + public void testListWithErrorsLogsWithoutThrowing() { + List<WriteStatus> statuses = Arrays.asList( + errored("err1"), + clean(), + errored("err2")); + assertDoesNotThrow(() -> WriteErrorReporter.logTopErrors(statuses, 10)); + } + + @Test + public void testMaxCapLimitsIteration() { + // Build a list with 5 errored statuses; cap at 2. Should iterate only the first 2 + // (no throw, no interaction beyond the first 2 verified by reaching the end of the call). + WriteStatus a = errored("a"); + WriteStatus b = errored("b"); + WriteStatus c = errored("c"); + WriteStatus d = errored("d"); + WriteStatus e = errored("e"); + assertDoesNotThrow(() -> WriteErrorReporter.logTopErrors(Arrays.asList(a, b, c, d, e), 2)); + // The other statuses must not have been touched. Their getErrors() should not have been called. + Mockito.verify(c, Mockito.never()).getErrors(); + Mockito.verify(d, Mockito.never()).getErrors(); + Mockito.verify(e, Mockito.never()).getErrors(); + } + + // ========== Helpers ========== + + private static WriteStatus errored(String globalError) { + WriteStatus ws = Mockito.mock(WriteStatus.class); + Mockito.when(ws.hasErrors()).thenReturn(true); + Mockito.when(ws.getGlobalError()).thenReturn(new RuntimeException(globalError)); + Mockito.when(ws.getErrors()).thenReturn(new HashMap<>()); + return ws; + } + + private static WriteStatus clean() { + WriteStatus ws = Mockito.mock(WriteStatus.class); + Mockito.when(ws.hasErrors()).thenReturn(false); + return ws; + } +} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/validator/TestSparkWriteErrorValidator.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/validator/TestSparkWriteErrorValidator.java new file mode 100644 index 000000000000..92460a94e943 --- /dev/null +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/validator/TestSparkWriteErrorValidator.java @@ -0,0 +1,198 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hudi.utilities.streamer.validator; + +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.model.HoodieWriteStat; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.config.HoodiePreCommitValidatorConfig; +import org.apache.hudi.exception.HoodieValidationException; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests for {@link SparkWriteErrorValidator}. + */ +public class TestSparkWriteErrorValidator { + + private static final String INSTANT = "20260520120000000"; + + // ========== Helpers ========== + + private static TypedProperties failConfig() { + TypedProperties props = new TypedProperties(); + props.setProperty(HoodiePreCommitValidatorConfig.VALIDATION_FAILURE_POLICY.key(), "FAIL"); + return props; + } + + private static TypedProperties warnConfig() { + TypedProperties props = new TypedProperties(); + props.setProperty(HoodiePreCommitValidatorConfig.VALIDATION_FAILURE_POLICY.key(), "WARN_LOG"); + return props; + } + + private static HoodieWriteStat stat(String partition, long numInserts, long numUpdates, long writeErrors) { + HoodieWriteStat s = new HoodieWriteStat(); + s.setPartitionPath(partition); + s.setNumInserts(numInserts); + s.setNumUpdateWrites(numUpdates); + s.setTotalWriteErrors(writeErrors); + return s; + } + + private static SparkValidationContext context(List<HoodieWriteStat> writeStats) { + return new SparkValidationContext( + INSTANT, + Option.of(new HoodieCommitMetadata()), + Option.of(writeStats), + Option.empty()); + } + + // ========== Tests ========== + + @Test + public void testNoErrorsPasses() { + // 1000 records written, no errors -> passes + SparkValidationContext ctx = context(Collections.singletonList(stat("p1", 1000, 0, 0))); + + SparkWriteErrorValidator validator = new SparkWriteErrorValidator(failConfig()); + assertDoesNotThrow(() -> validator.validateWithMetadata(ctx)); + } + + @Test + public void testErrorsWithFailPolicyThrows() { + // 500 written + 50 errors -> fails under FAIL policy (mirrors commitOnErrors=false) + SparkValidationContext ctx = context(Collections.singletonList(stat("p1", 500, 0, 50))); + + SparkWriteErrorValidator validator = new SparkWriteErrorValidator(failConfig()); + HoodieValidationException ex = assertThrows(HoodieValidationException.class, + () -> validator.validateWithMetadata(ctx)); + assertTrue(ex.getMessage().contains("Errors: 50"), "message should report error count"); + assertTrue(ex.getMessage().contains("Total: 550"), "message should report total record count"); + } + + @Test + public void testErrorsWithWarnPolicyDoesNotThrow() { + // 500 written + 50 errors -> passes under WARN_LOG (mirrors commitOnErrors=true) + SparkValidationContext ctx = context(Collections.singletonList(stat("p1", 500, 0, 50))); + + SparkWriteErrorValidator validator = new SparkWriteErrorValidator(warnConfig()); + assertDoesNotThrow(() -> validator.validateWithMetadata(ctx)); + } + + @Test + public void testEmptyCommitPasses() { + // 0 records, 0 errors -> empty commit, validation is skipped + SparkValidationContext ctx = context(Collections.singletonList(stat("p1", 0, 0, 0))); + + SparkWriteErrorValidator validator = new SparkWriteErrorValidator(failConfig()); + assertDoesNotThrow(() -> validator.validateWithMetadata(ctx)); + } + + @Test + public void testNoWriteStatsTreatedAsEmpty() { + SparkValidationContext ctx = new SparkValidationContext( + INSTANT, + Option.of(new HoodieCommitMetadata()), + Option.empty(), + Option.empty()); + + SparkWriteErrorValidator validator = new SparkWriteErrorValidator(failConfig()); + assertDoesNotThrow(() -> validator.validateWithMetadata(ctx)); + } + + @Test + public void testErrorsAcrossMultiplePartitionsAreSummed() { + // p1: 100 written + 5 errors. p2: 200 written + 10 errors. Total errors > 0 -> fail. + SparkValidationContext ctx = context(Arrays.asList( + stat("p1", 100, 0, 5), + stat("p2", 200, 0, 10))); + + SparkWriteErrorValidator validator = new SparkWriteErrorValidator(failConfig()); + HoodieValidationException ex = assertThrows(HoodieValidationException.class, + () -> validator.validateWithMetadata(ctx)); + assertTrue(ex.getMessage().contains("Errors: 15"), + "errors should be summed across partitions, got: " + ex.getMessage()); + assertTrue(ex.getMessage().contains("Total: 315"), + "total should be inserts + updates + errors across partitions, got: " + ex.getMessage()); + } + + @Test + public void testUpdatesCountedTowardTotal() { + // 0 inserts, 100 updates, 1 error -> 1/101 -> fail under FAIL + SparkValidationContext ctx = context(Collections.singletonList(stat("p1", 0, 100, 1))); + + SparkWriteErrorValidator validator = new SparkWriteErrorValidator(failConfig()); + HoodieValidationException ex = assertThrows(HoodieValidationException.class, + () -> validator.validateWithMetadata(ctx)); + assertTrue(ex.getMessage().contains("Total: 101"), + "updates should count toward total, got: " + ex.getMessage()); + } + + @Test + public void testDefaultPolicyIsFail() { + // No failure.policy set -> default is FAIL + TypedProperties props = new TypedProperties(); + SparkValidationContext ctx = context(Collections.singletonList(stat("p1", 10, 0, 1))); + + SparkWriteErrorValidator validator = new SparkWriteErrorValidator(props); + assertThrows(HoodieValidationException.class, () -> validator.validateWithMetadata(ctx)); + } + + @Test + public void testInvalidFailurePolicyRejected() { + TypedProperties props = new TypedProperties(); + props.setProperty(HoodiePreCommitValidatorConfig.VALIDATION_FAILURE_POLICY.key(), "garbage"); + HoodieValidationException ex = assertThrows(HoodieValidationException.class, + () -> new SparkWriteErrorValidator(props)); + assertTrue(ex.getMessage().contains("Invalid value 'garbage'"), + "message should name the bad value, got: " + ex.getMessage()); + assertTrue(ex.getMessage().contains("FAIL") && ex.getMessage().contains("WARN_LOG"), + "message should list allowed values, got: " + ex.getMessage()); + } + + @Test + public void testLowercasePolicyRejected() { + // Java enum valueOf is case-sensitive; lowercase should fail loudly with a clear message. + TypedProperties props = new TypedProperties(); + props.setProperty(HoodiePreCommitValidatorConfig.VALIDATION_FAILURE_POLICY.key(), "fail"); + assertThrows(HoodieValidationException.class, () -> new SparkWriteErrorValidator(props)); + } + + @Test + public void testErrorMessageReferencesCommitOnErrorsFlag() { + // Regression: the prior message referenced a non-existent config key + // "hoodie.streamer.commit.on.errors". The user-facing fix should mention --commit-on-errors. + SparkValidationContext ctx = context(Collections.singletonList(stat("p1", 10, 0, 1))); + HoodieValidationException ex = assertThrows(HoodieValidationException.class, + () -> new SparkWriteErrorValidator(failConfig()).validateWithMetadata(ctx)); + assertTrue(ex.getMessage().contains("--commit-on-errors"), + "should reference the real CLI flag, got: " + ex.getMessage()); + } +}
