🤖 nit: this flatMap lambda has grown to ~16 lines doing class loading, type checking, reflection and two distinct exception translations. Could you extract it into a small private helper like instantiateSparkValidator(String validatorClass, HoodieSparkTable table, HoodieEngineContext context, HoodieWriteConfig config) returning Stream<SparkPreCommitValidator>? Would make the stream pipeline read at a glance.

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

🤖 Unlike SparkStreamerValidatorUtils which has .filter(s -> !s.isEmpty()), this path will call Class.forName("") if the config has a trailing comma (e.g. "ValidatorA,"), turning a benign config typo into a HoodieValidationException. Could you add an empty-string filter before the Class.forName call, like the new utility does?

Fixed. Added .filter(s -> !s.isEmpty()) before the Class.forName call, matching the guard already present in SparkStreamerValidatorUtils.

🤖 nit: after the isAssignableFrom check, could you use ReflectionUtils.loadClass(validatorClass, new Class<?>[] {...}, table, context, config) for instantiation instead of getDeclaredConstructor(...).newInstance(...)? That would keep this consistent with the rest of the codebase (and with SparkStreamerValidatorUtils just below), while also getting Hudi's wrapper around instantiation errors for free.

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

Fixed

1. can we trigger the cache only when the validators are not empty;
2. I see we already have a writeStatusValidator for validating the write errors, it should be feasible that we migrate it as one of the pre-commit validators.

Thanks @danny0405!

1. Conditional cache — done in 5f4bca6. The cache/unpersist cycle is now guarded by a validatorsConfigured check derived from hoodie.precommit.validators, so the no-validator path stays a single-pass DAG with zero overhead.

2. Migrating writeStatusValidator to a pre-commit validator — agree with the direction, but I'd like to do this as a follow-up PR rather than expanding scope here. The reason: HoodieStreamerWriteStatusValidator (StreamSync.java:1403) does substantially more than pure validation:

• Counts records/errors and writes to a shared AtomicLong totalSuccessfulRecords consumed after commit (drives the runMetaSync() decision)
• Commits the error table as a side effect, with ROLLBACK_COMMIT/LOG_ERROR strategies
• Rolls back the instant on validation failure and logs the top-100 errors
• Runs inside writeClient.commit() via the WriteStatusValidator callback hook (not before commit), so the framework difference matters

A BasePreCommitValidator only takes TypedProperties and exposes state via ValidationContext — there's no path today to surface the raw JavaRDD<WriteStatus> (needed for error-table commit) or the post-commit counter through that interface. A clean migration would need to either (a) extend ValidationContext to expose the RDD and inject the error-table writer, or (b) split the side-effects out of the callback into a separate post-validation step.

Would you be OK if I file a follow-up issue + PR for that refactor so this PR can stay focused on the streaming-offset framework? Happy to take it on right after this lands.

Filed #18750 to track the writeStatusValidator migration as a follow-up. Will pick it up after this PR lands.

🤖 If SparkStreamerValidatorUtils.runValidators(...) throws (e.g. HoodieValidationException under FAIL policy, or HoodieIOException from loadPreviousCommitMetadata), control leaves the method before the try { ... } finally { unpersist() } block is entered, so the cached RDD is never unpersisted. Could you move the runValidators(...) call inside the same try whose finally does the unpersist() (or wrap the cache+validate+commit together) so the cache is always released on validation failure?

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

Fixed

🤖 If runValidators() or writeClient.commit() throws an exception, the unpersist() call is skipped and the cached RDD stays pinned in executor memory. Could you wrap the validate+commit+unpersist block in a try/finally so unpersist() always runs when weOwnCache is true?

_{- Generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.}