HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4194964639
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,75 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ latestBatchId: Long,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2) {
+ metadata
+ } else {
+ val hasStatefulShufflePartitions = OffsetSeqMetadata.readValueOpt(
+ metadata, SQLConf.STATEFUL_SHUFFLE_PARTITIONS_INTERNAL).isDefined
+ // In old checkpoints, this config may first appear in an offset written
by a failed restart.
Review Comment:
The baseline is simple, we consider the offset log as "source of truth" - we
do not ignore the value of shuffle partitions we see from offset log; it's a
source of truth and we do not try to figure out whether the value is poisoned
or not. If the value is available, it is a valid value. It only matters for the
offset log to be restored; based on the logic, we will only use the offset log
file for the latest batch to know the value. We do not care about the value in
old offset log file.
In any reason we do not see the value of shuffle partitions from offset log
(like the current bug), we try to get the value from state metadata. If we fail
to read the value in "any" reason, we fail the query.
For sure, we do not fall back to read the state metadata for stateless query
- 1) we don't need to override the value of pinned config from offset log to
session config 2) state metadata won't be available anyway.
--
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]
