HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4173224334
##########
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:
I assume this only happens with broken checkpoint (mostly in state
metadata), so it doesn't happen often.
But when it really happens, we must accept the fact that we have no way to
know the true value of state shuffle partitions for this query, and fail the
query. We shouldn't do swallowing the exception and falling back to use the
session config. If the value of session config is smaller than the actual state
shuffle partitions, we will end up with silent correctness issue, which is much
worse than breaking query.
It's just that we need a good error message - the error message should say
that the checkpoint is broken and they have to drop the checkpoint and
reprocess, so nuance is very important; users have to take the action and cost
because of Spark bug + some other issue (maybe they led it or another system
error).
--
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]
