[
https://issues.apache.org/jira/browse/SPARK-57879?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Max Gekk updated SPARK-57879:
-----------------------------
Description:
h2. Problem
{{ContinuousMemorySink}} overrides three methods from {{MemorySink}} and
unconditionally throws {{SparkUnsupportedOperationException}} with {{errorClass
= "UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK"}}:
* {{latestBatchData}} (ContinuousMemory.scala:84)
* {{dataSinceBatch(sinceBatchId: Long)}} (ContinuousMemory.scala:91)
* {{write(batchId: Long, needTruncate: Boolean, newRows: Array[Row])}}
(ContinuousMemory.scala:102)
However, the error class {{UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK}}
was *never registered* in {{error-conditions.json}}. Hitting any of these code
paths produces a misleading {{INTERNAL_ERROR}}: _"Cannot find main error class
'UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK'"_ instead of the intended
error.
h2. Impact
{{latestBatchData}} and {{dataSinceBatch}} are reachable via public API:
{{StreamTest.testStream}} holds the sink as {{MemorySink}}, and since
{{ContinuousMemorySink}} overrides those methods, {{CheckLastBatch}} or
{{CheckNewAnswer}} actions on a {{ContinuousMemorySink}} dispatch to the
throwing overrides -- yielding the wrong error class. Currently avoided only
because all {{StreamRealTimeModeSuite}} tests use {{CheckAnswer}} (which calls
{{allData}}, not overridden). {{write(batchId, ...)}} is effectively dead code
under normal streaming execution because {{ContinuousMemorySink}} uses an
RPC-based write path.
h2. Fix
Register the error class in {{error-conditions.json}} with {{sqlState =
"0A000"}} and a {{<operation>}} parameter placeholder. Add tests in
{{MemorySinkSuite}} that directly instantiate {{ContinuousMemorySink}} and call
all three methods, verifying the correct
{{UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK}} error is produced.
> Add missing error class UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK
> ------------------------------------------------------------------------
>
> Key: SPARK-57879
> URL: https://issues.apache.org/jira/browse/SPARK-57879
> Project: Spark
> Issue Type: Bug
> Components: Structured Streaming
> Affects Versions: 5.0.0
> Reporter: Max Gekk
> Priority: Major
>
> h2. Problem
> {{ContinuousMemorySink}} overrides three methods from {{MemorySink}} and
> unconditionally throws {{SparkUnsupportedOperationException}} with
> {{errorClass = "UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK"}}:
> * {{latestBatchData}} (ContinuousMemory.scala:84)
> * {{dataSinceBatch(sinceBatchId: Long)}} (ContinuousMemory.scala:91)
> * {{write(batchId: Long, needTruncate: Boolean, newRows: Array[Row])}}
> (ContinuousMemory.scala:102)
> However, the error class {{UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK}}
> was *never registered* in {{error-conditions.json}}. Hitting any of these
> code paths produces a misleading {{INTERNAL_ERROR}}: _"Cannot find main error
> class 'UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK'"_ instead of the
> intended error.
> h2. Impact
> {{latestBatchData}} and {{dataSinceBatch}} are reachable via public API:
> {{StreamTest.testStream}} holds the sink as {{MemorySink}}, and since
> {{ContinuousMemorySink}} overrides those methods, {{CheckLastBatch}} or
> {{CheckNewAnswer}} actions on a {{ContinuousMemorySink}} dispatch to the
> throwing overrides -- yielding the wrong error class. Currently avoided only
> because all {{StreamRealTimeModeSuite}} tests use {{CheckAnswer}} (which
> calls {{allData}}, not overridden). {{write(batchId, ...)}} is effectively
> dead code under normal streaming execution because {{ContinuousMemorySink}}
> uses an RPC-based write path.
> h2. Fix
> Register the error class in {{error-conditions.json}} with {{sqlState =
> "0A000"}} and a {{<operation>}} parameter placeholder. Add tests in
> {{MemorySinkSuite}} that directly instantiate {{ContinuousMemorySink}} and
> call all three methods, verifying the correct
> {{UNSUPPORTED_OPERATION_FOR_CONTINUOUS_MEMORY_SINK}} error is produced.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
