{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/10779","labels_url":"https://api.github.com/repos/apache/spark/issues/10779/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/10779/comments","events_url":"https://api.github.com/repos/apache/spark/issues/10779/events","html_url":"https://github.com/apache/spark/pull/10779","id":126982410,"number":10779,"title":"[SPARK-12847][Core][Streaming]Remove StreamingListenerBus and post all Streaming events to the same thread as Spark events","user":{"login":"zsxwing","id":1000778,"avatar_url":"https://avatars.githubusercontent.com/u/1000778?v=3","gravatar_id":"","url":"https://api.github.com/users/zsxwing","html_url":"https://github.com/zsxwing","followers_url":"https://api.github.com/users/zsxwing/followers","following_url":"https://api.github.com/users/zsxwing/following{/other_user}","gists_url":"https://api.github.com/users/zsxwing/gists{/gist_id}","starred_url":"https://api.github.com/users/zsxwing/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zsxwing/subscriptions","organizations_url":"https://api.github.com/users/zsxwing/orgs","repos_url":"https://api.github.com/users/zsxwing/repos","events_url":"https://api.github.com/users/zsxwing/events{/privacy}","received_events_url":"https://api.github.com/users/zsxwing/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":7,"created_at":"2016-01-15T23:56:34Z","updated_at":"2016-01-19T03:24:48Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/10779","html_url":"https://github.com/apache/spark/pull/10779","diff_url":"https://github.com/apache/spark/pull/10779.diff","patch_url":"https://github.com/apache/spark/pull/10779.patch"},"body":"Including the following changes:\r\n\r\n1. Add StreamingSparkListenerAdapter to process events in `onOtherEvent` to StreamingListener\r\n2. Remove StreamingListenerBus\r\n3. Merge AsynchronousListenerBus and LiveListenerBus to the same class LiveListenerBus\r\n4. Remove unnecessary generics\r\n5. Add `logEvent` method to SparkListenerEvent so that EventLoggingListener can use it to ignore StreamingListenerEvents"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/172727012","html_url":"https://github.com/apache/spark/pull/10779#issuecomment-172727012","issue_url":"https://api.github.com/repos/apache/spark/issues/10779","id":172727012,"user":{"login":"tdas","id":663212,"avatar_url":"https://avatars.githubusercontent.com/u/663212?v=3","gravatar_id":"","url":"https://api.github.com/users/tdas","html_url":"https://github.com/tdas","followers_url":"https://api.github.com/users/tdas/followers","following_url":"https://api.github.com/users/tdas/following{/other_user}","gists_url":"https://api.github.com/users/tdas/gists{/gist_id}","starred_url":"https://api.github.com/users/tdas/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tdas/subscriptions","organizations_url":"https://api.github.com/users/tdas/orgs","repos_url":"https://api.github.com/users/tdas/repos","events_url":"https://api.github.com/users/tdas/events{/privacy}","received_events_url":"https://api.github.com/users/tdas/received_events","type":"User","site_admin":false},"created_at":"2016-01-19T03:24:48Z","updated_at":"2016-01-19T03:24:48Z","body":"On the streaming side, @andrewor14  and I talked offline and there can be cleaner design, with better abstractions. \r\n\r\nCurrent design basically stores the list of StreamingListener in the SparkListenerBus (using the adaptor), and makes each StreamingListenerEvent extends a SparkListenerEvent. Since there is no StreamingListenerBus anymore, the abstraction is a little hard to understand on what gets posted where and who is calling the callbacks. Also the public API is being changed, which is also awkward - StreamingListener does not extend SparkListener but StreamingListenerEvent extends SparkListenerEvent.\r\n\r\nI think a better design is the following. The goal is simply for the existing StreamingListenerBus to not maintain its own thread and use the SparkListenerBus's thread to post everything, To do that all that needs to be done is for the StreamingListenerBus to forward the events into the SparkListenerBus. This can be done by the following. \r\n\r\n\r\n```\r\nclass StreamingListenerForwardingBus(sparkListenerBus: SparkListenerBus) extends SparkListener {\r\n     \r\n   case class WrappedStreamingListenerEvent(streamingListenerEvent: StreamingListenerEvent) \r\n        extends SparkListenerEvent {\r\n        protected[spark] override def logEvent: Boolean = false\r\n   }\r\n\r\n\r\n   private val listeners = new ArrayBuffer[StreamingListener]()\r\n\r\n   sparkListenerBus.add(this)    // for getting callbacks on spark events\r\n\r\n   def addListener(listener: StreamingListener) { listeners += listener }\r\n\r\n   def post(event: StreamingListenerEvent) {\r\n        sparkListenerBus.post(new WrappedStreamingListenerEvent(event))\r\n   } \r\n\r\n    override def onOtherEvents(event: SparkListenerEvent) {\r\n        event match {\r\n            case WrappedStreamingListenerEvent(sle) => sle match {  .... \r\n                   // call listeners\r\n            }\r\n            case _ => \r\n        }\r\n     }\r\n}\r\n\r\n\r\nclass JobScheduler {\r\n   ...\r\n   val listenerBus = new FakeStreamingListenerBus(sparkContext.listenerBus)\r\n}\r\n```\r\n\r\nThis maintains the clean abstraction that streaming events get posted to streaming bus (internally forwarded to spark bus), AND does not require public API changes (streaming events do not have to extend spark events).\r\n\r\nWhat do you think?\r\n\r\n\r\n"}}