(spark) branch master updated: [SPARK-52740][SS] Fix NPE in HDFSBackedStateStoreProvider accessing StateStoreConf.sqlConf when checkpoint format version is >=2

[as...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

ashrigondekar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new d91a5da06920 [SPARK-52740][SS] Fix NPE in HDFSBackedStateStoreProvider accessing StateStoreConf.sqlConf when checkpoint format version is >=2
d91a5da06920 is described below

commit d91a5da0692008fe332d02d6b875ee359757b5d6
Author: Livia Zhu <li...@databricks.com>
AuthorDate: Fri Jul 11 17:08:55 2025 -0700

    [SPARK-52740][SS] Fix NPE in HDFSBackedStateStoreProvider accessing StateStoreConf.sqlConf when checkpoint format version is >=2
    
    ### What changes were proposed in this pull request?
    
    This change stores checkpoint format version as a field StateStoreConf directly so we do not need to access the transient sqlConf field.
    
    ### Why are the changes needed?
    
    Since StateStoreConf's sqlConf field is transient, it is set to null when the conf is serialized and deserialized. This causes a NPE in HDFSBackedStateStoreProvider when checkpoint format version >= 2 because it accesses the sqlConf field for the assert error message.
    
    ### Does this PR introduce _any_ user-facing change?
    
    No
    
    ### How was this patch tested?
    
    New unit test
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    Closes #51431 from liviazhu/liviazhu-db/hdfs-npe.
    
    Authored-by: Livia Zhu <li...@databricks.com>
    Signed-off-by: Anish Shrigondekar <an...@databricks.com>
---
 .../state/HDFSBackedStateStoreProvider.scala       |  2 +-
 .../execution/streaming/state/StateStoreConf.scala |  7 +++-
 .../streaming/state/StateStoreSuite.scala          | 39 +++++++++++++++++++++-
 3 files changed, 45 insertions(+), 3 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala
index 25a1ca249c22..0b59b038cbc1 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala
@@ -351,7 +351,7 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
     assert(
       !storeConf.enableStateStoreCheckpointIds,
       "HDFS State Store Provider doesn't support checkpointFormatVersion >= 2 " +
-        s"checkpointFormatVersion ${storeConf.sqlConf.stateStoreCheckpointFormatVersion}")
+        s"checkpointFormatVersion ${storeConf.stateStoreCheckpointFormatVersion}")
 
     this.stateStoreId_ = stateStoreId
     this.keySchema = keySchema
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreConf.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreConf.scala
index 9a994200baeb..cbf8227ac08c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreConf.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreConf.scala
@@ -22,7 +22,9 @@ import org.apache.spark.sql.internal.SQLConf
 
 /** A class that contains configuration parameters for [[StateStore]]s. */
 class StateStoreConf(
-    @transient private[state] val sqlConf: SQLConf,
+    // Should be private because it could be null under serialization (due to
+    // the transient annotation)
+    @transient private val sqlConf: SQLConf,
     val extraOptions: Map[String, String] = Map.empty)
   extends Serializable {
 
@@ -106,6 +108,9 @@ class StateStoreConf(
   /** Whether to unload the store on task completion. */
   val unloadOnCommit = sqlConf.stateStoreUnloadOnCommit
 
+  /** The version of the state store checkpoint format. */
+  val stateStoreCheckpointFormatVersion: Int = sqlConf.stateStoreCheckpointFormatVersion
+
   /**
    * Additional configurations related to state store. This will capture all configs in
    * SQLConf that start with `spark.sql.streaming.stateStore.`
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala
index aa4a50b853a4..dae52e8b5120 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala
@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.execution.streaming.state
 
-import java.io.{File, IOException}
+import java.io.{ByteArrayInputStream, ByteArrayOutputStream, File, IOException, ObjectInputStream, ObjectOutputStream}
 import java.net.URI
 import java.util
 import java.util.UUID
@@ -986,6 +986,43 @@ class StateStoreSuite extends StateStoreSuiteBase[HDFSBackedStateStoreProvider]
     }
   }
 
+  test("SPARK-52740: HDFSBackedStateStoreProvider checkpoint format version 2 with " +
+    "deserialized sqlConf throws error correctly") {
+    val sqlConf = getDefaultSQLConf(SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.defaultValue.get,
+      SQLConf.MAX_BATCHES_TO_RETAIN_IN_MEMORY.defaultValue.get)
+    sqlConf.setConf(SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION, 2)
+    val originalConf = new StateStoreConf(sqlConf)
+
+    // Serialize
+    val baos = new ByteArrayOutputStream()
+    val oos = new ObjectOutputStream(baos)
+    oos.writeObject(originalConf)
+    oos.close()
+
+    // Deserialize
+    val bais = new ByteArrayInputStream(baos.toByteArray)
+    val ois = new ObjectInputStream(bais)
+    val deserializedConf = ois.readObject().asInstanceOf[StateStoreConf]
+    ois.close()
+
+    val provider = new HDFSBackedStateStoreProvider()
+    val hadoopConf = new Configuration()
+    hadoopConf.set(StreamExecution.RUN_ID_KEY, UUID.randomUUID().toString)
+
+    val e = intercept[AssertionError] {
+      provider.init(
+        StateStoreId(newDir(), Random.nextInt(), 0),
+        keySchema,
+        valueSchema,
+        NoPrefixKeyStateEncoderSpec(keySchema),
+        useColumnFamilies = false,
+        deserializedConf,
+        hadoopConf)
+    }
+    assert(e.getMessage.contains(
+      "HDFS State Store Provider doesn't support checkpointFormatVersion >= 2"))
+  }
+
   override def newStoreProvider(): HDFSBackedStateStoreProvider = {
     newStoreProvider(opId = Random.nextInt(), partition = 0)
   }


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