[SPARK-55999][SS] Enable forceSnapshotUploadOnLag by default

### What changes were proposed in this pull request?

1. Config default change (SQLConf.scala): forceSnapshotUploadOnLag default false → true
2. 3 SPARK-51358 tests fixed (StateStoreCoordinatorSuite.scala): Explicitly set forceSnapshotUploadOnLag=false so lag detection tests aren't interfered with by the forced remediation
3. Cleanup (StateStoreCoordinatorSuite.scala): Removed redundant forceSnapshotUploadOnLag -> "true"

### Why are the changes needed?

When state store is lagging in uploading snapshot in maintenance thread, turning on this feature allows state store to upload snapshot in query execution thread, which improves query reliability

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

The feature is tested in #52773

### Was this patch authored or co-authored using generative AI tooling?

yes

Closes #54847 from zifeif2/enable-force-snapshot.

Authored-by: zifeif2 <[email protected]>
Signed-off-by: Anish Shrigondekar <[email protected]>

Author: zifeif2

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -2863,7 +2863,7 @@ object SQLConf {
       )
       .version("4.2.0")
       .booleanConf
-      .createWithDefault(false)
+      .createWithDefault(true)
 
   val STATEFUL_SHUFFLE_PARTITIONS_INTERNAL =
     buildConf("spark.sql.streaming.internal.stateStore.partitions")

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

@@ -440,7 +440,8 @@ class StateStoreCoordinatorSuite extends SparkFunSuite with SharedSparkContext {
         RocksDBConf.ROCKSDB_SQL_CONF_NAME_PREFIX + ".changelogCheckpointing.enabled" -> "true",
         SQLConf.STATE_STORE_COORDINATOR_REPORT_SNAPSHOT_UPLOAD_LAG.key -> "true",
         SQLConf.STATE_STORE_COORDINATOR_MULTIPLIER_FOR_MIN_VERSION_DIFF_TO_LOG.key -> "2",
-        SQLConf.STATE_STORE_COORDINATOR_SNAPSHOT_LAG_REPORT_INTERVAL.key -> "0"
+        SQLConf.STATE_STORE_COORDINATOR_SNAPSHOT_LAG_REPORT_INTERVAL.key -> "0",
+        SQLConf.STATE_STORE_FORCE_SNAPSHOT_UPLOAD_ON_LAG.key -> "false"
       ) {
         case (coordRef, spark) =>
           import spark.implicits._
@@ -477,7 +478,8 @@ class StateStoreCoordinatorSuite extends SparkFunSuite with SharedSparkContext {
         SQLConf.STATE_STORE_COORDINATOR_REPORT_SNAPSHOT_UPLOAD_LAG.key -> "true",
         SQLConf.STATE_STORE_COORDINATOR_MULTIPLIER_FOR_MIN_VERSION_DIFF_TO_LOG.key -> "5",
         SQLConf.STATE_STORE_COORDINATOR_SNAPSHOT_LAG_REPORT_INTERVAL.key -> "0",
-        SQLConf.STATE_STORE_COORDINATOR_MAX_LAGGING_STORES_TO_REPORT.key -> "5"
+        SQLConf.STATE_STORE_COORDINATOR_MAX_LAGGING_STORES_TO_REPORT.key -> "5",
+        SQLConf.STATE_STORE_FORCE_SNAPSHOT_UPLOAD_ON_LAG.key -> "false"
       ) {
         case (coordRef, spark) =>
           import spark.implicits._
@@ -521,7 +523,8 @@ class StateStoreCoordinatorSuite extends SparkFunSuite with SharedSparkContext {
       RocksDBConf.ROCKSDB_SQL_CONF_NAME_PREFIX + ".changelogCheckpointing.enabled" -> "true",
       SQLConf.STATE_STORE_COORDINATOR_REPORT_SNAPSHOT_UPLOAD_LAG.key -> "true",
       SQLConf.STATE_STORE_COORDINATOR_MULTIPLIER_FOR_MIN_VERSION_DIFF_TO_LOG.key -> "2",
-      SQLConf.STATE_STORE_COORDINATOR_SNAPSHOT_LAG_REPORT_INTERVAL.key -> "0"
+      SQLConf.STATE_STORE_COORDINATOR_SNAPSHOT_LAG_REPORT_INTERVAL.key -> "0",
+      SQLConf.STATE_STORE_FORCE_SNAPSHOT_UPLOAD_ON_LAG.key -> "false"
     ) {
       case (coordRef, spark) =>
         import spark.implicits._
@@ -946,7 +949,6 @@ class StateStoreCoordinatorStreamingSuite extends StreamTest {
 object StateStoreCoordinatorSuite {
   // Common configuration for SPARK-54063 tests
   private val spark54063CommonConfigs: Seq[(String, String)] = Seq(
-    SQLConf.STATE_STORE_FORCE_SNAPSHOT_UPLOAD_ON_LAG.key -> "true",
     SQLConf.SHUFFLE_PARTITIONS.key -> "5",
     SQLConf.STREAMING_MAINTENANCE_INTERVAL.key -> "100",
     SQLConf.STATE_STORE_MAINTENANCE_SHUTDOWN_TIMEOUT.key -> "3",

sql/core/src/test/scala/org/apache/spark/sql/internal/SQLConfSuite.scala

@@ -562,6 +562,9 @@ class SQLConfSuite extends QueryTest with SharedSparkSession {
 
   test("[SPARK-54063] STATE_STORE_FORCE_SNAPSHOT_UPLOAD_ON_LAG requires " +
     "STATE_STORE_COORDINATOR_REPORT_SNAPSHOT_UPLOAD_LAG") {
+    // Default values should work fine - both default to true
+    assert(spark.sessionState.conf.stateStoreForceSnapshotUploadOnLag === true)
+
     // This should work fine - both enabled
     withSQLConf(
       SQLConf.STATE_STORE_COORDINATOR_REPORT_SNAPSHOT_UPLOAD_LAG.key -> "true",