HeartSaVioR commented on code in PR #42940:
URL: https://github.com/apache/spark/pull/42940#discussion_r1329427441
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala:
##########
@@ -52,11 +52,40 @@ class MicroBatchExecution(
@volatile protected var sources: Seq[SparkDataStream] = Seq.empty
- protected val triggerExecutor: TriggerExecutor = trigger match {
- case t: ProcessingTimeTrigger => ProcessingTimeExecutor(t, triggerClock)
- case OneTimeTrigger => SingleBatchExecutor()
- case AvailableNowTrigger => MultiBatchExecutor()
- case _ => throw new IllegalStateException(s"Unknown type of trigger:
$trigger")
+ @volatile protected[sql] var triggerExecutor: TriggerExecutor = _
+
+ protected def getTrigger(): TriggerExecutor = {
+ assert(sources.nonEmpty, "sources should have been retrieved from the
plan!")
+ trigger match {
+ case t: ProcessingTimeTrigger => ProcessingTimeExecutor(t, triggerClock)
+ case OneTimeTrigger => SingleBatchExecutor()
+ case AvailableNowTrigger =>
+ // See SPARK-45178 for more details.
+ if (sparkSession.sqlContext.conf.getConf(
+ SQLConf.STREAMING_TRIGGER_AVAILABLE_NOW_WRAPPER_ENABLED)) {
+ logInfo("Configured to use the wrapper of Trigger.AvailableNow.")
Review Comment:
Yeah let's put queryID and runID at least to differentiate the query. Please
note that this is an INFO log indicating that they turned on the flag (which is
OK), and the WARN log will follow when they are falling back to use wrapper
(which could be problematic).
--
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]
