HeartSaVioR commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-5969127744
> **AI-assisted review disclaimer:** This review was produced by an
LLM-based coding agent. The
> reproducer and conclusions below were verified by executing the focused
test against the stated
> PR commit.
I think there is still a recovery hole for a checkpoint that has already
gone through one failed
restart on an affected Spark version.
The sequence is:
1. Batch `N` is committed with five-partition state, but its V2 offset
metadata has no
`spark.sql.shuffle.partitions` entry.
2. The query is restarted with `spark.sql.shuffle.partitions=10`. Before
loading state, the old
code writes offset `N + 1` with `10` copied into its metadata.
3. Loading the five-partition state fails, leaving offset `N + 1`
uncommitted but present.
4. On the next restart, this PR sees that the latest offset contains the key
and skips recovery.
It therefore tries ten partitions again and fails in exactly the same way.
5. There should also be another restart after successfully re-executing `N +
1`: re-execution
commits the existing offset rather than rewriting it, so recovery must
not leave the checkpoint
poisoned for the following restart.
I reproduced this against `2bcb0ffedc52ac76ed87d13a7b070f6920adf053`. The
focused test fails with:
```text
CANNOT_LOAD_STATE_STORE.CANNOT_READ_DELTA_FILE_NOT_EXISTS
.../state/0/5/1.delta does not exist
```
Temporarily bypassing the present-key short circuit in
`recoverStatefulShufflePartitions` makes
both restarts in the test pass with five state stores. That was only a
mutation check, not a
suggested implementation: the fix still needs to preserve legitimate
repartition semantics.
The `STREAMING_CHECK_UNFINISHED_REPARTITION_ON_RESTART=false` override below
only isolates this
case because #58929 has not landed in this PR's base. Without it, the
separate SPARK-59662
`None.get` failure happens before this recovery path runs. When testing on
top of #58929, the
override can be removed.
<details>
<summary>Regression test diff (modifies the existing recovery test)</summary>
```diff
@@
- test("VERSION_2 recovers state-store partitions when the offset conf is
missing") {
+ test("VERSION_2 recovers state-store partitions after a failed legacy
restart") {
@@
withSQLConf(
SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
+ SQLConf.STREAMING_CHECK_UNFINISHED_REPARTITION_ON_RESTART.key ->
"false",
SQLConf.SHUFFLE_PARTITIONS.key -> shufflePartitions.toString) {
@@
assert(!offsetLog.get(latestBatchId).get.metadataOpt.get.conf
.contains(SQLConf.SHUFFLE_PARTITIONS.key))
+ // A restart on an affected Spark version copies 10 into a new offset
before failing to load
+ // the five-partition state, leaving that offset uncommitted.
+ val poisonedEndOffset = inputData.addData((2, 0), (3, 0))
+ val poisonedOffset = offsetWithoutShufflePartitions.copy(
+ offsetsMap = offsetWithoutShufflePartitions.offsetsMap.map {
+ case (sourceId, _) => sourceId -> Some(poisonedEndOffset)
+ },
+ metadata = offsetWithoutShufflePartitions.metadata.copy(
+ conf = offsetWithoutShufflePartitions.metadata.conf.updated(
+ SQLConf.SHUFFLE_PARTITIONS.key, "10")))
+ assert(offsetLog.add(latestBatchId + 1, poisonedOffset))
+
val query2 = startQuery(shufflePartitions = 10)
try {
- inputData.addData((2, 0), (3, 0))
query2.processAllAvailable()
assert(query2.streamingQuery.lastExecution.numStateStores === 5)
} finally {
query2.stop()
}
+ // The recovered batch commits the existing poisoned offset rather
than rewriting it.
+ // Verify that the following restart is also recoverable.
+ val query3 = startQuery(shufflePartitions = 10)
+ try {
+ inputData.addData((3, 0), (4, 0))
+ query3.processAllAvailable()
+ assert(query3.streamingQuery.lastExecution.numStateStores === 5)
+ } finally {
+ query3.stop()
+ }
+
assert(offsetLog.getLatest().get._2.metadataOpt.get.conf
.get(SQLConf.SHUFFLE_PARTITIONS.key).contains("5"))
```
</details>
Run with:
```bash
build/sbt 'sql/testOnly
org.apache.spark.sql.execution.streaming.OffsetSeqLogSuite -- -z "failed legacy
restart"'
```
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
