hudi-agent commented on code in PR #18765: URL: https://github.com/apache/hudi/pull/18765#discussion_r3267193026
########## hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/validator/SparkWriteErrorValidator.java: ########## @@ -0,0 +1,124 @@ +/* + * 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 lombok.extern.slf4j.Slf4j; + +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 = totalErroredRecords > 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:</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>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> + */ +@Slf4j +public class SparkWriteErrorValidator extends BasePreCommitValidator { + + 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; Review Comment: 🤖 When `isErrorTableWriteUnificationEnabled=true`, HSWSV added the error-table RDD's `getTotalErrorRecords()` into `totalErroredRecords`, but this validator only sees data-table stats via `ValidationContext`. So the stated `commitOnErrors=false ↔ failure.policy=FAIL` equivalence in the Javadoc is not exact under unification — a user who switches fully to the validator (e.g. sets `commitOnErrors=true` to avoid the redundant Step 4 gate) would silently miss error-table errors that HSWSV would have caught. Could you either weaken the Javadoc to call out this limitation, or have the orchestration in `StreamSync` pass the unified error count through (e.g. via a context attribute) so the validator sees the same totals HSWSV did? <sub><i>- AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SuccessfulRecordCounter.java: ########## @@ -0,0 +1,127 @@ +/* + * 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 totalErroredRecords = 0L; + for (WriteStatus ws : dataTableWriteStatuses) { + totalRecords += ws.getTotalRecords(); + totalErroredRecords += ws.getTotalErrorRecords(); + } + if (isErrorTableWriteUnificationEnabled && errorTableWriteStatusRDDOpt.isPresent()) { + JavaRDD<WriteStatus> errorRdd = errorTableWriteStatusRDDOpt.get(); + totalRecords += sumLong(errorRdd, WriteStatusLongExtractor.TOTAL_RECORDS); + totalErroredRecords += sumLong(errorRdd, WriteStatusLongExtractor.TOTAL_ERROR_RECORDS); + } + return new Counts(totalRecords, totalErroredRecords); + } + + /** + * Lossless long sum over an RDD. Avoids the precision loss of {@code mapToDouble().sum()}, + * which silently rounds counts above 2^53 (about 9 quadrillion). + */ + private static long sumLong(JavaRDD<WriteStatus> rdd, WriteStatusLongExtractor extractor) { + return rdd.map(extractor::extract).fold(0L, Long::sum); + } + + private enum WriteStatusLongExtractor { + TOTAL_RECORDS { Review Comment: 🤖 nit: the `WriteStatusLongExtractor` enum feels heavy for two getters — could you replace it with a `ToLongFunction<WriteStatus>` parameter and call `sumLong(errorRdd, WriteStatus::getTotalRecords)` / `sumLong(errorRdd, WriteStatus::getTotalErrorRecords)`? Removes the enum entirely and the call sites read more directly. <sub><i>- AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java: ########## @@ -872,38 +869,103 @@ private Pair<Option<String>, JavaRDD<WriteStatus>> writeToSinkAndDoMetaSync(Hood 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 { - if (validatorsConfigured) { - List<WriteStatus> writeStatuses = writeStatusRDD.collect(); + 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); + } + } + } - // 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. + // 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. + if (validatorsConfigured) { SparkStreamerValidatorUtils.runValidators(props, instantTime, writeStatuses, checkpointCommitMetadata, metaClient); Review Comment: 🤖 Should this also call `writeClient.rollback(instantTime)` before the validator's `HoodieValidationException` propagates? Step 1 (error-table ROLLBACK_COMMIT) and Step 4 (write-error gate) both rollback the inflight data-table instant with the comment "doesn't leak under LAZY failed-writes cleanup policy." When a Step 2 validator throws, the data-table instant is left inflight with no rollback — and now the error table has already been committed in Step 1, so the divergence is wider than in HSWSV's original flow where the error-table commit came after validators ran. Was preserving "no rollback on validator failure" intentional, or is this an oversight that the same LAZY-cleanup argument should cover? <sub><i>- AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestSuccessfulRecordCounter.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; + +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.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 org.apache.spark.api.java.JavaSparkContext jsc; Review Comment: 🤖 nit: could you add proper imports for `JavaSparkContext` / `SparkConf` instead of inline fully-qualified names here and in `setUp`? Reads better and matches the style elsewhere in the suite. <sub><i>- AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
