[
https://issues.apache.org/jira/browse/SPARK-60533?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Pratham Manja updated SPARK-60533:
----------------------------------
Affects Version/s: 3.5.4
> Enable spark.sql.streaming.checkpoint.renamedFileCheck.enabled by
> default
> --------------------------------------------------------------------------
>
> Key: SPARK-60533
> URL: https://issues.apache.org/jira/browse/SPARK-60533
> Project: Spark
> Issue Type: Improvement
> Components: Structured Streaming
> Affects Versions: 3.5.4, 4.0.0
> Reporter: Pratham Manja
> Priority: Major
> Labels: pull-request-available
>
> h3. Problem
> Structured Streaming writes every checkpoint file (state store
> deltas/snapshots, RocksDB changelogs/zips, offset and commit logs) to a
> temporary file and then renames it into place
> (\{{CheckpointFileManager.createAtomic}} ->
> \{{RenameBasedFSDataOutputStream.close()}}). The rename is treated as
> successful as long as it does not throw. SPARK-40826 added an optional check
> that the renamed file actually exists, but it is disabled by default
> (\{{spark.sql.streaming.checkpoint.renamedFileCheck.enabled=false}}).
> We hit this in production on Azure Data Lake Storage Gen2 (ABFS). A state
> store delta file was never created under its final name, while its temporary
> file was left behind, and no error was raised:
> - The batch was committed to the commit log as if the state file existed.
> - The query kept running, because executors keep the latest state versions in
> memory and do not re-read the file.
> - On the next restart, rebuilding state from disk fails on the missing delta
> file, and every subsequent restart fails the same way. Recovering needs
> manual intervention on the checkpoint.
> h3. Reproduction
> A local file system whose rename reports success without moving the file,
> injected through Spark's default (FileContext-based) checkpoint manager, on a
> stateful query (\{{groupBy().count()}}, one shuffle partition). Reproduced on
> 3.5.4; the code path is unchanged on master.
> With \{{renamedFileCheck.enabled=false}} (current default), the batch commits
> while its state file is missing:
> {noformat}
> rename .3.delta.43cdd488-1b16-4d7b-88c2-b964081f43be.TID4.tmp -> 3.delta
> reports success but nothing is moved
> offsets/ batches 0..4
> commits/ batches 0..4
> state/0/0 1.delta 2.delta 4.delta 5.delta
> leftover temp file:
> .3.delta.43cdd488-1b16-4d7b-88c2-b964081f43be.TID4.tmp
> MISSING: 3.delta, although commits/2 says batch 2 (which wrote it)
> is done
> {noformat}
> After a restart, every attempt fails reading the missing file, and the query
> terminates:
> {noformat}
> ERROR MicroBatchExecution: Query repro [id = 24db9b4e-..., runId =
> f9a053f2-...] terminated with error
> org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in
> stage 11.0 failed 4 times, most recent failure: Lost task 0.3 in stage 11.0
> (TID 13) (executor driver): java.lang.IllegalStateException: Error reading
> delta file file:/tmp/spark-rfc-off/checkpoint/state/0/0/3.delta of
> HDFSStateStoreProvider[id = (op=0,part=0),dir =
> file:/tmp/spark-rfc-off/checkpoint/state/0/0]:
> file:/tmp/spark-rfc-off/checkpoint/state/0/0/3.delta does not exist
> at
> org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider.updateFromDeltaFile(HDFSBackedStateStoreProvider.scala:464)
> at
> org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider.loadMap(HDFSBackedStateStoreProvider.scala:391)
> at
> org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider.getReadStore(HDFSBackedStateStoreProvider.scala:215)
> at
> org.apache.spark.sql.execution.streaming.state.StateStore$.getReadOnly(StateStore.scala:492)
> ...
> {noformat}
> With \{{renamedFileCheck.enabled=true}}, the same fault is caught at commit
> time; the task fails, Spark retries it, the retry writes 3.delta and the
> query continues:
> {noformat}
> WARN TaskSetManager: Lost task 0.0 in stage 5.0 (TID 4) (executor driver):
> java.lang.IllegalStateException: Error committing version 3 into
> HDFSStateStore[id=(op=0,part=0),dir=file:/tmp/spark-rfc-on/checkpoint/state/0/0]
> Caused by: java.lang.IllegalStateException: Renamed temp file
> file:/tmp/spark-rfc-on/checkpoint/state/0/0/.3.delta.4eb8988b-a9da-4f96-8651-05ef69a7da82.TID4.tmp
> to file:/tmp/spark-rfc-on/checkpoint/state/0/0/3.delta. But
> file:/tmp/spark-rfc-on/checkpoint/state/0/0/3.delta does not exist.
> at
> org.apache.spark.sql.execution.streaming.CheckpointFileManager$RenameBasedFSDataOutputStream.close(CheckpointFileManager.scala:165)
> ...
> at
> org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider$HDFSBackedStateStore.commit(HDFSBackedStateStoreProvider.scala:141)
> state/0/0 1.delta 2.delta 3.delta 4.delta 5.delta 6.delta
> {noformat}
> h3. Proposal
> Enable \{{spark.sql.streaming.checkpoint.renamedFileCheck.enabled}} by
> default. A lost rename then fails the write immediately, before the batch is
> committed, so it is retried like any other task failure instead of leaving a
> checkpoint that cannot be restarted. The cost is one existence check per
> checkpoint file written. The flag stays available to opt out.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
