Pratham Manja created SPARK-60533:
-------------------------------------
Summary: 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: 4.0.0
Reporter: Pratham Manja
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]
