Re: [PR] [SPARK-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]

["HeartSaVioR (via GitHub)" <gi...@apache.org>]

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.
   
   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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org