HyukjinKwon commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4171774811
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,42 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 ||
+ OffsetSeqMetadata.readValueOpt(
+ metadata, SQLConf.STATEFUL_SHUFFLE_PARTITIONS_INTERNAL).isDefined ||
+ committedBatchId < 0) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ new Path(checkpointFile("state")).getParent.toString,
+ new
SerializableConfiguration(sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ committedBatchId)
+ stateMetadataReader.stateStoreNumPartitions.map { numPartitions =>
+ logWarning(log"Recovered state-store partition count " +
+ log"${MDC(CONFIG, SQLConf.SHUFFLE_PARTITIONS.key)}=" +
+ log"${MDC(NUM_PARTITIONS, numPartitions)} " +
+ log"from checkpoint state metadata because it was missing from
the offset log")
+ OffsetSeqMetadata.withStatefulShufflePartitions(v2,
numPartitions)
+ }.getOrElse(metadata)
+ } catch {
+ case NonFatal(e) =>
Review Comment:
**Nit (P3):** This try/catch(NonFatal) is a deliberate graceful-degradation
path: if the state-metadata read fails, recovery is skipped and the query still
starts with the unmodified metadata. The new tests cover successful recovery,
restart-preservation, and the empty-metadata case, but none exercises this
fallback. Consider adding a test that points a legacy VERSION_2 checkpoint
(with the offset conf removed, as in the recovery test) at an unreadable or
corrupt state-metadata directory and asserts the query still starts with
unchanged metadata, so a future change that lets the read throw instead of
degrading is caught.
**Verification:**
- **Regression:** Reproduce a failed state-metadata read during recovery on
a legacy VERSION_2 checkpoint missing the conf, and confirm the query still
starts with unmodified metadata; the assertion must fail if the read exception
is allowed to propagate instead of being caught.
--
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]
