This is an automated email from the ASF dual-hosted git repository.
MaxGekk 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 e280f1697c33 [SPARK-57763][SS] Add missing error class
UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK
e280f1697c33 is described below
commit e280f1697c3315c03e13c9201120c85a75937ea9
Author: Szehon Ho <[email protected]>
AuthorDate: Thu Jul 2 16:54:25 2026 +0200
[SPARK-57763][SS] Add missing error class
UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK
### What changes were proposed in this pull request?
`ContinuousMemorySink` throws the
`UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK` error class from
`latestBatchData`, `dataSinceBatch`, and `write`, but the error class was never
registered in `error-conditions.json`. This registers the error condition so it
resolves correctly through the error framework, and adds a test.
### Why are the changes needed?
The error class was used but not declared. Hitting this path raised
`INTERNAL_ERROR` ("Cannot find main error class
'UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK'") instead of the intended
error.
### Does this PR introduce _any_ user-facing change?
No. Previously this code path raised an `INTERNAL_ERROR` because the error
class was unregistered; now the intended error message is produced.
### How was this patch tested?
Added a unit test in `ContinuousMemorySuite`. Also ran
`SparkThrowableSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor (Claude Opus 4.8)
Closes #56873 from szehon-ho/SPARK-57763.
Authored-by: Szehon Ho <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 6 +++++
.../streaming/sources/ContinuousMemorySuite.scala | 27 ++++++++++++++++++++++
2 files changed, 33 insertions(+)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index baf168c5c97c..f6b4fc05540c 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -8749,6 +8749,12 @@
},
"sqlState" : "42K0E"
},
+ "UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK" : {
+ "message" : [
+ "The operation <operation> is not supported for ContinuousMemorySink."
+ ],
+ "sqlState" : "0A000"
+ },
"UNSUPPORTED_OVERWRITE" : {
"message" : [
"Can't overwrite the target that is also being read from."
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/sources/ContinuousMemorySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/sources/ContinuousMemorySuite.scala
index c2b479724d62..69e87ee61ee3 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/sources/ContinuousMemorySuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/sources/ContinuousMemorySuite.scala
@@ -18,6 +18,8 @@
package org.apache.spark.sql.execution.streaming.sources
+import org.apache.spark.SparkUnsupportedOperationException
+import org.apache.spark.sql.Row
import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
import org.apache.spark.sql.streaming.{OutputMode, StreamTest}
@@ -41,4 +43,29 @@ class ContinuousMemorySuite extends StreamTest {
StartStream(),
CheckAnswer(2, 3, 4, 5, 6, 7, 8, 11, 12))
}
+
+ test("UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK") {
+ val sink = new ContinuousMemorySink()
+
+ checkError(
+ exception = intercept[SparkUnsupportedOperationException] {
+ sink.latestBatchData
+ },
+ condition = "UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK",
+ parameters = Map("operation" -> "latestBatchData"))
+
+ checkError(
+ exception = intercept[SparkUnsupportedOperationException] {
+ sink.dataSinceBatch(0)
+ },
+ condition = "UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK",
+ parameters = Map("operation" -> "dataSinceBatch"))
+
+ checkError(
+ exception = intercept[SparkUnsupportedOperationException] {
+ sink.write(batchId = 0, needTruncate = false, newRows =
Array.empty[Row])
+ },
+ condition = "UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK",
+ parameters = Map("operation" -> "write"))
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
