You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by as...@apache.org on 2026/03/20 20:50:29 UTC
(spark) branch master updated: [SPARK-55999][SS] Enable forceSnapshotUploadOnLag by default
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 8e88f5ab8501 [SPARK-55999][SS] Enable forceSnapshotUploadOnLag by default
8e88f5ab8501 is described below
commit 8e88f5ab850191edb8c02e054fcb15c22ccdd9ca
Author: zifeif2 <zi...@gmail.com>
AuthorDate: Fri Mar 20 13:50:15 2026 -0700
[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 https://github.com/apache/spark/pull/52773
### Was this patch authored or co-authored using generative AI tooling?
yes
Closes #54847 from zifeif2/enable-force-snapshot.
Authored-by: zifeif2 <zi...@gmail.com>
Signed-off-by: Anish Shrigondekar <an...@databricks.com>
---
.../src/main/scala/org/apache/spark/sql/internal/SQLConf.scala | 2 +-
.../execution/streaming/state/StateStoreCoordinatorSuite.scala | 10 ++++++----
.../scala/org/apache/spark/sql/internal/SQLConfSuite.scala | 3 +++
3 files changed, 10 insertions(+), 5 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 2c872d3a7567..22f5b3f6c792 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/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")
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreCoordinatorSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreCoordinatorSuite.scala
index 03257d3da373..8e8cff582c6f 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreCoordinatorSuite.scala
+++ b/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",
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/internal/SQLConfSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/internal/SQLConfSuite.scala
index 5b80ed8fc4e3..2bb3b4247d01 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/internal/SQLConfSuite.scala
+++ b/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",
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org