Corrected bug in user plugins files (#1801)

Stratio · Mar 23, 2017 · 9b7d388 · 9b7d388
1 parent 5914278
commit 9b7d388
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 6 deletions.
diff --git a/driver/src/main/scala/com/stratio/sparta/driver/SpartaClusterJob.scala b/driver/src/main/scala/com/stratio/sparta/driver/SpartaClusterJob.scala
@@ -78,7 +78,7 @@ object SpartaClusterJob extends PluginsFilesUtils {
         statusActor ! Update(PolicyStatusModel(id = policyId, status = Starting, statusInfo = Some(startingInfo)))
         val streamingContextService = StreamingContextService(statusActor)
 
-        val ssc = streamingContextService.clusterStreamingContext(policy)
+        val ssc = streamingContextService.clusterStreamingContext(policy, pluginsFiles)
         statusActor ! Update(PolicyStatusModel(
           id = policyId,
           status = NotDefined,

diff --git a/driver/src/main/scala/com/stratio/sparta/driver/factory/SparkContextFactory.scala b/driver/src/main/scala/com/stratio/sparta/driver/factory/SparkContextFactory.scala
@@ -74,8 +74,8 @@ object SparkContextFactory extends SLF4JLogging {
       sc.getOrElse(instantiateSparkContext(specificConfig, jars))
     }
 
-  def sparkClusterContextInstance(specificConfig: Map[String, String]): SparkContext =
-    sc.getOrElse(instantiateClusterContext(specificConfig))
+  def sparkClusterContextInstance(specificConfig: Map[String, String], files: Seq[String]): SparkContext =
+    sc.getOrElse(instantiateClusterContext(specificConfig, files))
 
   private def instantiateSparkContext(specificConfig: Map[String, String], jars: Seq[File]): SparkContext = {
     sc = Some(SparkContext.getOrCreate(configToSparkConf(specificConfig)))
@@ -86,8 +86,12 @@ object SparkContextFactory extends SLF4JLogging {
     sc.get
   }
 
-  private def instantiateClusterContext(specificConfig: Map[String, String]): SparkContext = {
+  private def instantiateClusterContext(specificConfig: Map[String, String], files: Seq[String]): SparkContext = {
     sc = Some(SparkContext.getOrCreate(configToSparkConf(specificConfig)))
+    files.foreach(f => {
+      log.info(s"Adding jar $f to cluster Spark context")
+      sc.get.addJar(f)
+    })
     sc.get
   }
 

diff --git a/driver/src/main/scala/com/stratio/sparta/driver/service/StreamingContextService.scala b/driver/src/main/scala/com/stratio/sparta/driver/service/StreamingContextService.scala
@@ -60,14 +60,14 @@ case class StreamingContextService(statusActor: ActorRef, generalConfig: Option[
     ssc
   }
 
-  def clusterStreamingContext(policy: PolicyModel)
+  def clusterStreamingContext(policy: PolicyModel, files: Seq[String])
   : StreamingContext = {
     if (autoDeleteCheckpointPath(policy)) deleteCheckpointPath(policy)
 
     val ssc = StreamingContext.getOrCreate(checkpointPath(policy), () => {
       log.info(s"Nothing in checkpoint path: ${checkpointPath(policy)}")
       val outputsSparkConfig = PolicyHelper.getSparkConfigs(policy, OutputsSparkConfiguration, Output.ClassSuffix)
-      sparkClusterContextInstance(outputsSparkConfig)
+      sparkClusterContextInstance(outputsSparkConfig, files)
       SpartaPipeline(policy, statusActor).run()
     })