hudi-agent commented on code in PR #19940:
URL: https://github.com/apache/hudi/pull/19940#discussion_r4007681591
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/ErrorTableCommitter.java:
##########
@@ -47,6 +49,41 @@ public final class ErrorTableCommitter {
private ErrorTableCommitter() {
}
+ /**
+ * Materializes the error-table write statuses, then commits the error table
like {@link #commit}
+ * and returns both. The error table's commit releases the RDD it was given,
so a later evaluation
+ * of that RDD re-runs the write itself and lands the same records a second
time under the
+ * committed instant; anything that needs the statuses after the commit must
read them from here.
+ * An RDD the writer did not persist is cached for the duration of the call
so the collect and the
+ * commit share one evaluation.
+ */
+ public static ErrorTableCommitResult
collectAndCommit(BaseErrorTableWriter<?> errorTableWriter,
+
Option<JavaRDD<WriteStatus>> errorTableWriteStatusRDDOpt,
+ boolean
isErrorTableWriteUnificationEnabled,
+ String instantTime,
+ Option<String>
latestCommittedInstant) {
+ Objects.requireNonNull(errorTableWriteStatusRDDOpt,
"errorTableWriteStatusRDDOpt");
+ if (!isErrorTableWriteUnificationEnabled ||
!errorTableWriteStatusRDDOpt.isPresent()) {
+ boolean success = commit(errorTableWriter, errorTableWriteStatusRDDOpt,
isErrorTableWriteUnificationEnabled,
+ instantTime, latestCommittedInstant);
+ return new ErrorTableCommitResult(success, Option.empty());
+ }
+ JavaRDD<WriteStatus> writeStatusRDD = errorTableWriteStatusRDDOpt.get();
+ boolean cacheHere =
writeStatusRDD.getStorageLevel().equals(StorageLevel.NONE());
+ if (cacheHere) {
+ writeStatusRDD.cache();
Review Comment:
🤖 `cache()` is `MEMORY_ONLY`, so a partition that doesn't fit (or gets
evicted) is silently dropped and recomputed by the commit's evaluation — which
is exactly the double write this PR is guarding against. Would
`persist(StorageLevel.MEMORY_AND_DISK())` (or the writer's
`hoodie.write.status.storage.level`, default `MEMORY_AND_DISK_SER`) be a safer
net here? I realize this mirrors the data-table `cache()` in `StreamSync`, so
happy to leave it if you'd rather keep them consistent.
<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:
##########
@@ -45,32 +43,29 @@ 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
+ * @param errorTableWriteStatuses Error-table write statuses
collected before the error table
Review Comment:
🤖 nit: the rewrapped javadoc now reads as one run-on sentence ("...or empty
when none were written when unification is enabled") which is ambiguous about
what depends on unification being enabled. Might be worth splitting back into
two sentences like the original.
<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]
