Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/5554#discussion_r28608698
--- Diff:
streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala
---
@@ -46,21 +48,20 @@ class JobScheduler(val ssc: StreamingContext) extends
Logging {
val listenerBus = new StreamingListenerBus()
// These two are created only when scheduler starts.
- // eventActor not being null means the scheduler has been started and
not stopped
+ // eventLoop not being null means the scheduler has been started and not
stopped
var receiverTracker: ReceiverTracker = null
- private var eventActor: ActorRef = null
-
+ private var eventLoop: EventLoop[JobSchedulerEvent] = null
def start(): Unit = synchronized {
- if (eventActor != null) return // scheduler has already been started
+ if (eventLoop != null) return // scheduler has already been started
logDebug("Starting JobScheduler")
- eventActor = ssc.env.actorSystem.actorOf(Props(new Actor {
- override def receive: PartialFunction[Any, Unit] = {
- case event: JobSchedulerEvent => processEvent(event)
- }
- }), "JobScheduler")
+ eventLoop = new EventLoop[JobSchedulerEvent]("JobScheduler") {
+ override protected def onReceive(event: JobSchedulerEvent): Unit =
processEvent(event)
+ override protected def onError(e: Throwable): Unit =
reportError("Error in job scheduler", e)
+ }
+ eventLoop.start()
listenerBus.start(ssc.sparkContext)
--- End diff --
blank line here
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
