[SPARK-51922] [SS] Fix UTFDataFormatException thrown from StateStoreChangelogReaderFactory for v1

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

@@ -153,7 +153,7 @@ class RocksDB(
     )
   }
 
-  private val fileManager = createFileManager(dfsRootDir, createTempDir("fileManager"),
+  private[spark] val fileManager = createFileManager(dfsRootDir, createTempDir("fileManager"),
     hadoopConf, conf.compressionCodec, loggingId = loggingId)
   private val byteArrayPair = new ByteArrayPair()
   private val commitLatencyMs = new mutable.HashMap[String, Long]()

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreChangelog.scala

@@ -368,7 +368,10 @@ class StateStoreChangelogReaderFactory(
       // When there is no record being written in the changelog file in V1,
       // the file contains a single int -1 meaning EOF, then the above readUTF()
       // throws with EOFException and we return version 1.
-      case _: java.io.EOFException => 1
+      // Or if the first record in the changelog file in V1 has a large enough
+      // key, readUTF() will throw a UTFDataFormatException so we should return
+      // version 1 (SPARK-51922).
+      case _: java.io.EOFException | _: java.io.UTFDataFormatException => 1
     }
   }
 

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala

@@ -3516,6 +3516,24 @@ class RocksDBSuite extends AlsoTestWithRocksDBFeatures with SharedSparkSession
     }
   }
 
+  testWithChangelogCheckpointingEnabled("SPARK-51922 - Changelog writer v1 with large key" +
+    " does not cause UTFDataFormatException") {
+    val remoteDir = Utils.createTempDir()
+
+    withDB(remoteDir.toString) { db =>
+      db.load(0)
+      val key = new Array[Char](98304).mkString("") // Large key that would trigger UTFException
+      // if handled incorrectly
+      db.put(key, "0")
+      db.commit()
+
+      val changelogReader = db.fileManager.getChangelogReader(1)
+      assert(changelogReader.version === 1)
+      val entries = changelogReader.toSeq
+      assert(entries.size == 1)
+    }
+  }
+
   private def assertAcquiredThreadIsCurrentThread(db: RocksDB): Unit = {
     val threadInfo = db.getAcquiredThreadInfo()
     assert(threadInfo != None,