HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4191363232
##########
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:
Sorry I got lost since it got back and forth multiple times... I just
refreshed my head and reason about the thing by myself.
Here I see we are trying to restore the broken checkpoint probably impacted
by this bug. But this is risky since we are now disrespectful of the config
value in the offset log, not only the case of missing (which is unavoidable to
find a fallback), but also the case the value exists. I understand the scenario
how it happens and how this could help, but self-mitigation is open to produce
another problem and we could be in a loop on fixing issue forever.
Also the self-mitigation isn't even possible when poisoned value of shuffle
partitions is smaller than the actual shuffle partitions; the query won't fail
and succeed to produce incorrect output and commit, destroying the
self-mitigation.
So let's not make this too complicated - I'd rather propose accepting the
limitation. ~If we want to be safer than this, let's just do the check of
shuffle partitions vs state metadata and fail the query if they disagree.~
(This isn't even safe in case of state repartition...)
Once we get rid of this logic, I assume the test logic would be also
simplified (less scenarios to concern).
--
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]
