anishshri-db commented on code in PR #42940:
URL: https://github.com/apache/spark/pull/42940#discussion_r1329378160
##########
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.")
+ MultiBatchExecutor()
+ } else {
+ val supportsTriggerAvailableNow = sources.distinct.forall { src =>
+ val supports = src.isInstanceOf[SupportsTriggerAvailableNow]
+ if (!supports) {
+ logWarning(s"source [$src] does not support
Trigger.AvailableNow. Failing back to " +
Review Comment:
Nit: Typo ? `Falling back to` ?
--
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]
