HyukjinKwon commented on code in PR #59247:
URL: https://github.com/apache/spark/pull/59247#discussion_r4277360367
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryStatusAndProgressSuite.scala:
##########
@@ -43,6 +46,111 @@ import org.apache.spark.sql.types.StructType
import org.apache.spark.util.ArrayImplicits._
class StreamingQueryStatusAndProgressSuite extends StreamTest with Eventually
with Matchers {
+ gridTest("SPARK-59986: trigger metadata")(Seq[(Trigger, String)](
+ Trigger.ProcessingTime(0) ->
"""{"type":"ProcessingTime","intervalMs":0}""",
+ Trigger.ProcessingTime(3000000000L) ->
+ """{"type":"ProcessingTime","intervalMs":3000000000}""",
+ Trigger.Continuous(2000) -> """{"type":"Continuous","intervalMs":2000}""",
+ Trigger.RealTime(3000) -> """{"type":"RealTime","batchDurationMs":3000}""",
+ Trigger.AvailableNow() -> """{"type":"AvailableNow"}""",
+ Trigger.Once() -> """{"type":"Once"}""")) {
+ case (trigger, expected) =>
+ assert(Triggers.toJson(trigger) === expected)
+ assert(Triggers.fromJson(expected) === trigger)
+ }
+
+ test("SPARK-59986: unknown trigger metadata identifies the unsupported
trigger") {
+ val trigger = new Trigger {}
+ val exception = intercept[IllegalStateException] {
+ Triggers.toJson(trigger)
+ }
+ assert(exception.getMessage === s"Unknown type of trigger: $trigger")
+ }
+
+ test("SPARK-59986: starting a query with an unknown trigger identifies the
input") {
+ import testImplicits._
+
+ val input = MemoryStream[Int]
+ val trigger = new Trigger {}
+ val exception = intercept[IllegalStateException] {
Review Comment:
**Non-blocking (P2):** This intercept no longer sees an exception. Since
96b71ac, `ProgressReporter` keeps `triggerConfig` instead of eagerly calling
`Triggers.toJson` in its constructor, so nothing on the `start()` path
validates the trigger. The only check left is `MicroBatchExecution.getTrigger`,
reached through the `logicalPlan` lazy val on the query thread after
`startLatch` has let `start()` return. `stop()` only joins the thread and
doesn't rethrow. So `start().stop()` completes normally, and the test fails
with "Expected exception java.lang.IllegalStateException to be thrown, but no
exception was thrown".
The query still terminates asynchronously with `Unknown type of trigger:
...`, the same as on master. But this test and the PR description's "fails
query construction" claim no longer match the code.
To keep the documented behavior, restore a construction-time check. For
example, call `Triggers.toJson(triggerConfig)` when `ProgressReporter` is
constructed, which happens synchronously in `StreamExecution`'s constructor
inside `createQuery`. That makes `start()` throw as described.
**Verification:**
- **Regression:** Starting a streaming query with a custom Trigger subclass
fails synchronously with an IllegalStateException whose message identifies the
trigger.
- **Behavior:** Micro-batch, real-time, and continuous progress for built-in
triggers still reports the configured trigger metadata.
##########
sql/api/src/main/scala/org/apache/spark/sql/execution/streaming/Triggers.scala:
##########
@@ -30,7 +31,33 @@ import org.apache.spark.sql.catalyst.util.SparkIntervalUtils
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.unsafe.types.UTF8String
-private object Triggers {
+private[spark] object Triggers {
+ private implicit val formats: DefaultFormats = DefaultFormats
+
+ def toJson(trigger: Trigger): String = trigger match {
+ case ProcessingTimeTrigger(intervalMs) =>
+ s"""{"type":"ProcessingTime","intervalMs":$intervalMs}"""
+ case ContinuousTrigger(intervalMs) =>
+ s"""{"type":"Continuous","intervalMs":$intervalMs}"""
+ case RealTimeTrigger(batchDurationMs) =>
+ s"""{"type":"RealTime","batchDurationMs":$batchDurationMs}"""
+ case AvailableNowTrigger => """{"type":"AvailableNow"}"""
+ case OneTimeTrigger => """{"type":"Once"}"""
+ case other => throw new IllegalStateException(s"Unknown type of trigger:
$other")
+ }
+
+ def fromJson(json: String): Trigger = {
+ val value = parse(json)
+ (value \ "type").extract[String] match {
+ case "ProcessingTime" => ProcessingTimeTrigger((value \
"intervalMs").extract[Long])
+ case "Continuous" => ContinuousTrigger((value \
"intervalMs").extract[Long])
+ case "RealTime" => RealTimeTrigger((value \
"batchDurationMs").extract[Long])
+ case "AvailableNow" => AvailableNowTrigger
+ case "Once" => OneTimeTrigger
+ case other => throw new IllegalStateException(s"Unknown type of trigger:
$other")
Review Comment:
**Non-blocking (P2):** This decoder is the one part of progress decoding
that isn't forward compatible. `fromJson` throws for any `type` outside the
five built-ins, and the `extract[Long]` calls throw when an interval key is
missing. Both JVM decoders call it with no fallback:
`ObjectToTriggerDeserializer`, used by `StreamingQueryProgress.fromJson` and
`QueryProgressEvent.fromJson`, and
`StreamingQueryProgressSerializer.deserialize` for the status store.
Suppose a Scala Spark Connect client with this change talks to a later
server that adds a trigger kind (`RealTime` shows the set does grow) or changes
a key. `RemoteStreamingQuery.lastProgress` and `recentProgress` then throw. In
`StreamingQueryListenerBus.queryEventHandler`, the same exception reaches the
loop-level catch, which removes every client-side listener. An older history
server replaying a newer event log also stops at that event, because
`ReplayListenerBus` only skips `ClassNotFoundException` and
`UnrecognizedPropertyException`.
Everything around this degrades instead: both mappers set
`FAIL_ON_UNKNOWN_PROPERTIES=false`, offsets go through
`ObjectToStringDeserializer`, and the listener bus only warns on unknown
events. The PySpark `fromJson` in this PR already keeps the dict as-is. Classic
in-process `lastProgress` is unaffected because it never decodes.
Released clients can't be patched, so the tolerance needs to ship with the
field's first release. I'd make this helper total: return `null`, the
documented "metadata unavailable" value, for an unknown `type` or a missing or
malformed interval key. Then use it from both deserializers.
**Verification:**
- **Regression:** Decoding progress JSON with an unknown trigger type, and
with a missing interval key, returns a progress record whose trigger is null
instead of throwing, on both the Jackson and status-store protobuf paths.
- **Compatibility:** Every built-in trigger type still serializes to the
same JSON and decodes back to the same Trigger value.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
