@@ -383,10 +383,13 @@ class SQLAppStatusListenerSuite extends SparkFunSuite with SharedSQLContext with
383383 assertJobs(statusStore.execution(executionId), failed = Seq (0 ))
384384 }
385385
386- sqlStoreTest(" handle one execution with multiple jobs" ) { (store, bus) =>
386+ test(" handle one execution with multiple jobs" ) {
387+ val statusStore = createStatusStore()
388+ val listener = statusStore.listener.get
389+
387390 val executionId = 0
388391 val df = createTestDataFrame
389- bus.postToAll (SparkListenerSQLExecutionStart (
392+ listener.onOtherEvent (SparkListenerSQLExecutionStart (
390393 executionId,
391394 " test" ,
392395 " test" ,
@@ -398,16 +401,16 @@ class SQLAppStatusListenerSuite extends SparkFunSuite with SharedSQLContext with
398401 def twoStageJob (jobId : Int ): Unit = {
399402 val stages = Seq (stageId, stageId + 1 ).map { id => createStageInfo(id, 0 )}
400403 stageId += 2
401- bus.postToAll (SparkListenerJobStart (
404+ listener.onJobStart (SparkListenerJobStart (
402405 jobId = jobId,
403406 time = System .currentTimeMillis(),
404407 stageInfos = stages,
405408 createProperties(executionId)))
406409 stages.foreach { s =>
407- bus.postToAll (SparkListenerStageSubmitted (s))
408- bus.postToAll (SparkListenerStageCompleted (s))
410+ listener.onStageSubmitted (SparkListenerStageSubmitted (s))
411+ listener.onStageCompleted (SparkListenerStageCompleted (s))
409412 }
410- bus.postToAll (SparkListenerJobEnd (
413+ listener.onJobEnd (SparkListenerJobEnd (
411414 jobId = jobId,
412415 time = System .currentTimeMillis(),
413416 JobSucceeded
@@ -416,11 +419,11 @@ class SQLAppStatusListenerSuite extends SparkFunSuite with SharedSQLContext with
416419 // submit two jobs with the same executionId
417420 twoStageJob(0 )
418421 twoStageJob(1 )
419- bus.postToAll (SparkListenerSQLExecutionEnd (
422+ listener.onOtherEvent (SparkListenerSQLExecutionEnd (
420423 executionId, System .currentTimeMillis()))
421424
422- assertJobs(store .execution(0 ), completed = 0 to 1 )
423- assert(store .execution(0 ).get.stages === (0 to 3 ).toSet)
425+ assertJobs(statusStore .execution(0 ), completed = 0 to 1 )
426+ assert(statusStore .execution(0 ).get.stages === (0 to 3 ).toSet)
424427 }
425428
426429 test(" SPARK-11126: no memory leak when running non SQL jobs" ) {
0 commit comments