Hi all, We've hit silent data loss in Spark jobs writing to HDFS when Observer reads are enabled and FileOutputCommitter v1 is used. The root cause matches SPARK-36810, where an executor write followed by a driver read can be stale. We would like to share a concrete data-loss instance in the committer.
Setup: Hadoop-3.4.1, Spark-3.5.6 Observer: dfs.ha.tail-edits.in-progress=true, dfs.ha.tail-edits.period=0ms Client: ObserverReadProxyProvider, with a non-zero dfs.client.failover.observer.auto-msync-period What happened: A downstream data mismatch led us to a Spark write which succeeded even though its committed output was deleted during the job commit. The audit logs show: 1. The executor commits its task with a rename on the Active: _temporary/0/_temporary/attempt_... -> _temporary/0/task_... 2. About 5ms later, the driver's commitJob() lists _temporary/0 (getAllCommittedTaskPaths()), and an Observer serves that listing. 3. The listing doesn't include task_..., so it isn't merged. cleanupJob() then deletes _temporary, task output included, and the job writes _SUCCESS. No error is raised. Why it happens - The driver is notified the task has finished through spark RPC - The driver's stateID reflects only its own earlier calls. The mkdirs call in setupJob() reached the Active 419ms before listing. and any call that reaches the Active resets the auto-msync timer(ObserverReadProxyProvider). So no msync happened before listing, and the same would have been true with the documented 500ms value. - The Observer therefore answered immediately from a namespace that didn't include the executor's rename. Only 5ms separated the rename on the active from listing on the Observer. Even with tail-edits.period=0ms, Observer lag regularly exceeds that. Auto-msync is timed from the client's own last call to the Active, but this race depends on another client's commit a few ms earlier, Meanwhile commitJob() merges what's listed and delete the rest based on one listing that isn't guaranteed to see other clients task commits. Reproduction To make the millisecond window reproducible, we widened it on a test cluster. - Observer dfs.ha.tail-edits.period=30 - Client dfs.client.failover.observer.auto-msync-period=-1, this keeps the driver from msyncing during the widened window, In production the window was only 5ms and no msync happened before listing. - A spark job(Yarn mode) writing 1000 rows in 4 tasks, last task committing 75s after the job setup. - The job succeeded but the 250 rows written by the last task are missing. Proposal Call FileSystem.msync() once in FileOutputCommitter.commitJob() before listing committed tasks. It costs one RPC to the Active per job commit. It should address SPARK-36810 by calling FileSystem.msync() before making any read calls where the client thinks the read could have possibly happened elsewhere. If there is an agreement, we can submit a patch. Thanks D M Murali Krishna
