The event queue is java.util.concurrent.LinkedBlockingQueue with capacity of 10000
LiveListenerBus takes a SparkConf when created.
start(sc: SparkContext): Unit
start starts processing events.
Internally, it saves the input
SparkContext for later use and starts listenerThread. It makes sure that it only happens when
LiveListenerBus has not been started before (i.e.
started is disabled).
LiveListenerBus has already been started, a
IllegalStateException is thrown:
[name] already started!
post(event: SparkListenerEvent): Unit
post puts the input
event onto the internal
eventQueue queue and releases the internal
eventLock semaphore. If the event placement was not successful (and it could happen since it is tapped at 10000 events) onDropEvent method is called.
The event publishing is only possible when
stopped flag has been enabled.
FIXME Who’s enabling the
LiveListenerBus has been stopped, the following ERROR appears in the logs:
ERROR [name] has already stopped! Dropping event [event]
onDropEvent(event: SparkListenerEvent): Unit
onDropEvent is called when no further events can be added to the internal
eventQueue queue (while posting a SparkListenerEvent event).
It simply prints out the following ERROR message to the logs and ensures that it happens only once.
ERROR Dropping SparkListenerEvent because no remaining room in event queue. This likely means one of the SparkListeners is too slow and cannot keep up with the rate at which tasks are being started by the scheduler.
It uses the internal
stop releases the internal
eventLock semaphore and waits until listenerThread dies. It can only happen after all events were posted (and polling
eventQueue gives nothing).
It checks that
started flag is enabled (i.e.
true) and throws a
Attempted to stop [name] that has not yet started!
stopped flag is enabled.
|FIXME There is some logic around no events in the queue.|
|Spark Property||Default Value||Description|
addToStatusQueue(listener: SparkListenerInterface): Unit