Skip to content

Commit 8b439d5

Browse files
committed
fix(akka): fix spark 2 compilation issues
1 parent 0abac18 commit 8b439d5

File tree

1 file changed

+9
-5
lines changed

1 file changed

+9
-5
lines changed

job-server/src/main/scala/spark/jobserver/JobManagerActor.scala

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -74,7 +74,7 @@ object JobManagerActor {
7474
* }}}
7575
*/
7676
class JobManagerActor(contextConfig: Config, daoActor: ActorRef)
77-
extends InstrumentedActor with SparkListener {
77+
extends InstrumentedActor {
7878

7979
import CommonMessages._
8080
import JobManagerActor._
@@ -125,9 +125,13 @@ class JobManagerActor(contextConfig: Config, daoActor: ActorRef)
125125
}
126126

127127
// Handle external kill events (e.g. killed via YARN)
128-
override def onApplicationEnd(event: SparkListenerApplicationEnd) {
129-
logger.info("Got Spark Application end event, stopping job manger.")
130-
self ! PoisonPill
128+
private def sparkListener = {
129+
new SparkListener() {
130+
override def onApplicationEnd(event: SparkListenerApplicationEnd) {
131+
logger.info("Got Spark Application end event, stopping job manager.")
132+
self ! PoisonPill
133+
}
134+
}
131135
}
132136

133137
def wrappedReceive: Receive = {
@@ -142,7 +146,7 @@ class JobManagerActor(contextConfig: Config, daoActor: ActorRef)
142146
}
143147
factory = getContextFactory()
144148
jobContext = factory.makeContext(config, contextConfig, contextName)
145-
jobContext.sparkContext.addSparkListener(this)
149+
jobContext.sparkContext.addSparkListener(sparkListener)
146150
sparkEnv = SparkEnv.get
147151
jobCache = new JobCacheImpl(jobCacheSize, daoActor, jobContext.sparkContext, jarLoader)
148152
getSideJars(contextConfig).foreach { jarUri => jobContext.sparkContext.addJar(jarUri) }

0 commit comments

Comments
 (0)