[SPARK-50084][SQL] Assign appropriate error condition for _LEGACY_ERR…

…OR_TEMP_3168: MISSING_TIMEOUT_CONFIGURATION
apache · Oct 23, 2024 · 4151d91 · 4151d91
1 parent 51e915a
commit 4151d91
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 8 deletions.
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
@@ -3471,6 +3471,12 @@
     ],
     "sqlState" : "42803"
   },
+  "MISSING_TIMEOUT_CONFIGURATION" : {
+    "message" : [
+      "The operation has timed out, but no timeout duration is configured. To set a processing time-based timeout, use 'GroupState.setTimeoutDuration()' in your 'mapGroupsWithState' or 'flatMapGroupsWithState' operation. For event-time-based timeout, use 'GroupState.setTimeoutTimestamp()' and define a watermark using 'Dataset.withWatermark()'."
+    ],
+    "sqlState" : "HY000"
+  },
   "MISSING_WINDOW_SPECIFICATION" : {
     "message" : [
       "Window specification is not defined in the WINDOW clause for <windowName>. For more information about WINDOW clauses, please refer to '<docroot>/sql-ref-syntax-qry-select-window.html'."
@@ -8377,11 +8383,6 @@
       "continuous mode is not supported!"
     ]
   },
-  "_LEGACY_ERROR_TEMP_3168" : {
-    "message" : [
-      "hasTimedOut is true however there's no timeout configured"
-    ]
-  },
   "_LEGACY_ERROR_TEMP_3169" : {
     "message" : [
       "AcceptsLatestSeenOffset is not supported with DSv1 streaming source: <unsupportedSources>"

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/GroupStateImpl.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/GroupStateImpl.scala
@@ -218,7 +218,7 @@ private[sql] object GroupStateImpl {
       throw new IllegalArgumentException("eventTimeWatermarkMs must be 0 or positive if present")
     }
     if (hasTimedOut && timeoutConf == NoTimeout) {
-      throw new SparkUnsupportedOperationException("_LEGACY_ERROR_TEMP_3168")
+      throw new SparkUnsupportedOperationException("MISSING_TIMEOUT_CONFIGURATION")
     }
 
     new GroupStateImpl[S](

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/GroupStateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/GroupStateSuite.scala
@@ -302,13 +302,13 @@ class GroupStateSuite extends SparkFunSuite {
         TestGroupState.create[Int](
           Optional.of(5), NoTimeout, 100L, Optional.empty[Long], hasTimedOut = true)
       },
-      condition = "_LEGACY_ERROR_TEMP_3168",
+      condition = "MISSING_TIMEOUT_CONFIGURATION",
       parameters = Map.empty)
     checkError(
       exception = intercept[SparkUnsupportedOperationException] {
         GroupStateImpl.createForStreaming[Int](Some(5), 100L, NO_TIMESTAMP, NoTimeout, true, false)
       },
-      condition = "_LEGACY_ERROR_TEMP_3168",
+      condition = "MISSING_TIMEOUT_CONFIGURATION",
       parameters = Map.empty)
   }