diff --git a/.gitignore b/.gitignore
index eaefdef..fce2c13 100755
--- a/.gitignore
+++ b/.gitignore
@@ -52,4 +52,4 @@
 repo/**
 
 #python
-.zip
\ No newline at end of file
+*.zip
\ No newline at end of file
diff --git a/common/src/main/scala/org/apache/amaterasu/common/dataobjects/Export.scala b/common/src/main/scala/org/apache/amaterasu/common/dataobjects/Export.scala
deleted file mode 100644
index 8db81bc..0000000
--- a/common/src/main/scala/org/apache/amaterasu/common/dataobjects/Export.scala
+++ /dev/null
@@ -1,19 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.common.dataobjects
-
-case class Export(dataset: String, format: String)
diff --git a/common/src/main/scala/org/apache/amaterasu/common/dataobjects/JobData.scala b/common/src/main/scala/org/apache/amaterasu/common/dataobjects/JobData.scala
deleted file mode 100755
index acf5c4a..0000000
--- a/common/src/main/scala/org/apache/amaterasu/common/dataobjects/JobData.scala
+++ /dev/null
@@ -1,21 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.common.dataobjects
-
-import org.joda.time.DateTime
-
-case class JobData(src: String, branch: String = "master", id: String, timeCreated: DateTime, startTime: DateTime, endTime: DateTime)
\ No newline at end of file
diff --git a/common/src/main/scala/org/apache/amaterasu/common/utils/FileUtils.scala b/common/src/main/scala/org/apache/amaterasu/common/utils/FileUtils.scala
deleted file mode 100644
index 9813296..0000000
--- a/common/src/main/scala/org/apache/amaterasu/common/utils/FileUtils.scala
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.common.utils
-
-import java.io.File
-
-object FileUtils {
-
-  def getAllFiles(dir: File): Array[File] = {
-    val these = dir.listFiles
-    these ++ these.filter(_.isDirectory).flatMap(getAllFiles)
-  }
-
-}
diff --git a/executor/build.gradle b/executor/build.gradle
deleted file mode 100644
index 443fc38..0000000
--- a/executor/build.gradle
+++ /dev/null
@@ -1,105 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-plugins {
-    id 'com.github.johnrengelman.shadow' version '1.2.4'
-    id 'com.github.maiflai.scalatest' version '0.22'
-    id 'scala'
-    id 'java'
-}
-
-shadowJar {
-    zip64 true
-}
-
-//sourceCompatibility = 1.8
-//targetCompatibility = 1.8
-
-repositories {
-    maven {
-        url "https://plugins.gradle.org/m2/"
-    }
-    mavenCentral()
-}
-
-test {
-    maxParallelForks = 1
-    forkEvery = 1
-}
-
-configurations {
-    provided
-}
-
-sourceSets {
-    main.compileClasspath += configurations.provided
-    test.compileClasspath += configurations.provided
-    test.runtimeClasspath += configurations.provided
-}
-
-dependencies {
-
-    compile group: 'org.scala-lang', name: 'scala-library', version: '2.11.8'
-    compile group: 'org.scala-lang', name: 'scala-reflect', version: '2.11.8'
-    compile group: 'io.netty', name: 'netty-all', version: '4.0.42.Final'
-    compile group: 'org.apache.commons', name: 'commons-lang3', version: '3.5'
-    compile group: 'org.apache.maven', name: 'maven-core', version: '3.0.5'
-    compile group: 'org.reflections', name: 'reflections', version: '0.9.10'
-    compile group: 'com.fasterxml.jackson.core', name: 'jackson-core', version: '2.9.8'
-    compile group: 'com.fasterxml.jackson.core', name: 'jackson-annotations', version: '2.9.8'
-    compile group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: '2.9.8'
-    compile group: 'com.fasterxml.jackson.module', name: 'jackson-module-scala_2.11', version: '2.9.8'
-
-    compile('com.jcabi:jcabi-aether:0.10.1') {
-        exclude group: 'org.jboss.netty'
-    }
-
-
-    compile project(':common')
-    compile project(':amaterasu-sdk')
-
-
-
-}
-
-sourceSets {
-    test {
-        resources.srcDirs += [file('src/test/resources')]
-    }
-
-    // this is done so Scala will compile before Java
-    main {
-        scala {
-            srcDirs = ['src/main/scala', 'src/main/java']
-        }
-        java {
-            srcDirs = []
-        }
-    }
-}
-
-test {
-
-    maxParallelForks = 1
-}
-
-task copyToHome(type: Copy) {
-    dependsOn shadowJar
-    from 'build/libs'
-    into '../build/amaterasu/dist'
-    from 'build/resources/main'
-    into '../build/amaterasu/dist'
-}
diff --git a/executor/src/main/scala/org/apache/amaterasu/executor/common/executors/ProvidersFactory.scala b/executor/src/main/scala/org/apache/amaterasu/executor/common/executors/ProvidersFactory.scala
deleted file mode 100644
index 5f78bc2..0000000
--- a/executor/src/main/scala/org/apache/amaterasu/executor/common/executors/ProvidersFactory.scala
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.executor.common.executors
-
-import java.io.ByteArrayOutputStream
-
-import org.apache.amaterasu.common.configuration.ClusterConfig
-import org.apache.amaterasu.common.dataobjects.ExecData
-import org.apache.amaterasu.common.execution.actions.Notifier
-import org.apache.amaterasu.sdk.{AmaterasuRunner, RunnersProvider}
-import org.reflections.Reflections
-
-import scala.collection.JavaConversions._
-
-//TODO: Check if we can use this in the YARN impl
-class ProvidersFactory {
-
-  var providers: Map[String, RunnersProvider] = _
-
-  def getRunner(groupId: String, id: String): Option[AmaterasuRunner] = {
-    val provider = providers.get(groupId)
-    provider match {
-      case Some(provider) => Some(provider.getRunner(id))
-      case None => None
-    }
-  }
-}
-
-object ProvidersFactory {
-
-  def apply(data: ExecData,
-            jobId: String,
-            outStream: ByteArrayOutputStream,
-            notifier: Notifier,
-            executorId: String,
-            hostName: String,
-            propFile: String = null): ProvidersFactory = {
-
-    val result = new ProvidersFactory()
-    val reflections = new Reflections(getClass.getClassLoader)
-    val runnerTypes = reflections.getSubTypesOf(classOf[RunnersProvider]).toSet
-    val config = if (propFile != null) {
-      import java.io.FileInputStream
-      ClusterConfig.apply(new FileInputStream(propFile))
-    } else {
-      new ClusterConfig()
-    }
-
-    result.providers = runnerTypes.map(r => {
-
-      val provider = Manifest.classType(r).runtimeClass.newInstance.asInstanceOf[RunnersProvider]
-
-      provider.init(data, jobId, outStream, notifier, executorId, config, hostName)
-      notifier.info(s"a provider for group ${provider.getGroupIdentifier} was created")
-      (provider.getGroupIdentifier, provider)
-    }).toMap
-
-    result
-  }
-
-}
\ No newline at end of file
diff --git a/executor/src/main/scala/org/apache/amaterasu/executor/mesos/executors/MesosActionsExecutor.scala b/executor/src/main/scala/org/apache/amaterasu/executor/mesos/executors/MesosActionsExecutor.scala
deleted file mode 100755
index f204db8..0000000
--- a/executor/src/main/scala/org/apache/amaterasu/executor/mesos/executors/MesosActionsExecutor.scala
+++ /dev/null
@@ -1,153 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.executor.mesos.executors
-
-import java.io.{ByteArrayInputStream, ByteArrayOutputStream}
-
-import com.fasterxml.jackson.databind.ObjectMapper
-import com.fasterxml.jackson.module.scala.DefaultScalaModule
-import org.apache.amaterasu.common.dataobjects.{ExecData, TaskData}
-import org.apache.amaterasu.common.logging.Logging
-import org.apache.amaterasu.executor.common.executors.ProvidersFactory
-import org.apache.mesos.Protos._
-import org.apache.mesos.protobuf.ByteString
-import org.apache.mesos.{Executor, ExecutorDriver, MesosExecutorDriver}
-
-import scala.collection.JavaConverters._
-import scala.concurrent.ExecutionContext.Implicits.global
-import scala.concurrent.Future
-import scala.util.{Failure, Success}
-
-class MesosActionsExecutor extends Logging with Executor {
-
-  var master: String = _
-  var executorDriver: ExecutorDriver = _
-  var jobId: String = _
-  var actionName: String = _
-  //  var sparkScalaRunner: SparkScalaRunner = _
-  //  var pySparkRunner: PySparkRunner = _
-  var notifier: MesosNotifier = _
-  var providersFactory: ProvidersFactory = _
-
-  val mapper = new ObjectMapper()
-  mapper.registerModule(DefaultScalaModule)
-
-
-  override def shutdown(driver: ExecutorDriver) = {
-
-  }
-
-  override def killTask(driver: ExecutorDriver, taskId: TaskID) = ???
-
-  override def disconnected(driver: ExecutorDriver) = ???
-
-  override def reregistered(driver: ExecutorDriver, slaveInfo: SlaveInfo) = {
-    this.executorDriver = driver
-  }
-
-  override def error(driver: ExecutorDriver, message: String) = {
-
-    val status = TaskStatus.newBuilder
-      .setData(ByteString.copyFromUtf8(message))
-      .setState(TaskState.TASK_ERROR).build()
-
-    driver.sendStatusUpdate(status)
-
-  }
-
-  override def frameworkMessage(driver: ExecutorDriver, data: Array[Byte]) = ???
-
-  override def registered(driver: ExecutorDriver, executorInfo: ExecutorInfo, frameworkInfo: FrameworkInfo, slaveInfo: SlaveInfo): Unit = {
-
-    this.executorDriver = driver
-    val data = mapper.readValue(new ByteArrayInputStream(executorInfo.getData.toByteArray), classOf[ExecData])
-
-    // this is used to resolve the spark drier address
-    val hostName = slaveInfo.getHostname
-    notifier = new MesosNotifier(driver)
-    notifier.info(s"Executor ${executorInfo.getExecutorId.getValue} registered")
-    val outStream = new ByteArrayOutputStream()
-    providersFactory = ProvidersFactory(data, jobId, outStream, notifier, executorInfo.getExecutorId.getValue, hostName, "./amaterasu.properties")
-
-  }
-
-  override def launchTask(driver: ExecutorDriver, taskInfo: TaskInfo): Unit = {
-
-
-    notifier.info(s"launching task: ${taskInfo.getTaskId.getValue}")
-    log.debug(s"launching task: $taskInfo")
-    val status = TaskStatus.newBuilder
-      .setTaskId(taskInfo.getTaskId)
-      .setState(TaskState.TASK_STARTING).build()
-
-    notifier.info(s"container info: ${taskInfo.getContainer.getDocker.getImage}, ${taskInfo.getContainer.getType}")
-
-    driver.sendStatusUpdate(status)
-
-    val task = Future {
-
-      val taskData = mapper.readValue(new ByteArrayInputStream(taskInfo.getData.toByteArray), classOf[TaskData])
-
-      val status = TaskStatus.newBuilder
-        .setTaskId(taskInfo.getTaskId)
-        .setState(TaskState.TASK_RUNNING).build()
-      driver.sendStatusUpdate(status)
-      val runner = providersFactory.getRunner(taskData.getGroupId, taskData.getTypeId)
-      runner match {
-        case Some(r) => r.executeSource(taskData.getSrc, actionName, taskData.getExports)
-        case None =>
-          notifier.error("", s"Runner not found for group: ${taskData.getGroupId}, type ${taskData.getTypeId}. Please verify the tasks")
-          None
-      }
-
-    }
-
-    task onComplete {
-
-      case Failure(t) =>
-        println(s"launching task Failed: ${t.getMessage}")
-        System.exit(1)
-
-      case Success(ts) =>
-
-        driver.sendStatusUpdate(TaskStatus.newBuilder()
-          .setTaskId(taskInfo.getTaskId)
-          .setState(TaskState.TASK_FINISHED).build())
-        notifier.info(s"Complete task: ${taskInfo.getTaskId.getValue}")
-
-    }
-
-  }
-
-}
-
-object MesosActionsExecutor extends Logging {
-
-  def main(args: Array[String]) {
-    System.loadLibrary("mesos")
-    log.debug("Starting a new ActionExecutor")
-
-    val executor = new MesosActionsExecutor
-    executor.jobId = args(0)
-    executor.master = args(1)
-    executor.actionName = args(2)
-
-    val driver = new MesosExecutorDriver(executor)
-    driver.run()
-  }
-
-}
diff --git a/executor/src/main/scala/org/apache/amaterasu/executor/mesos/executors/MesosNotifier.scala b/executor/src/main/scala/org/apache/amaterasu/executor/mesos/executors/MesosNotifier.scala
deleted file mode 100755
index b256386..0000000
--- a/executor/src/main/scala/org/apache/amaterasu/executor/mesos/executors/MesosNotifier.scala
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.executor.mesos.executors
-
-import com.fasterxml.jackson.databind.ObjectMapper
-import com.fasterxml.jackson.module.scala.DefaultScalaModule
-import org.apache.amaterasu.common.execution.actions.enums.{NotificationLevel, NotificationType}
-import org.apache.amaterasu.common.execution.actions.{Notification, Notifier}
-import org.apache.mesos.ExecutorDriver
-
-
-class MesosNotifier(driver: ExecutorDriver) extends Notifier {
-
-  private val mapper = new ObjectMapper()
-  mapper.registerModule(DefaultScalaModule)
-
-  override def success(line: String): Unit = {
-
-
-    getLog.info(s"successfully executed line: $line")
-
-    val notification = new Notification(line, "", NotificationType.Success, NotificationLevel.Code)
-    val msg = mapper.writeValueAsBytes(notification)
-
-    driver.sendFrameworkMessage(msg)
-
-  }
-
-  override def error(line: String, message: String): Unit = {
-
-    getLog.error(s"Error executing line: $line message: $message")
-
-    val notification = new Notification(line, message, NotificationType.Error, NotificationLevel.Code)
-    val msg = mapper.writeValueAsBytes(notification)
-
-    driver.sendFrameworkMessage(msg)
-
-  }
-
-  override def info(message: String): Unit = {
-
-    getLog.info(message)
-
-    val notification = new Notification("", message, NotificationType.Info, NotificationLevel.Execution)
-    val msg = mapper.writeValueAsBytes(notification)
-
-    driver.sendFrameworkMessage(msg)
-  }
-}
\ No newline at end of file
diff --git a/executor/src/main/scala/org/apache/amaterasu/executor/yarn/executors/ActionsExecutor.scala b/executor/src/main/scala/org/apache/amaterasu/executor/yarn/executors/ActionsExecutor.scala
deleted file mode 100644
index cda6351..0000000
--- a/executor/src/main/scala/org/apache/amaterasu/executor/yarn/executors/ActionsExecutor.scala
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.executor.yarn.executors
-
-import java.io.ByteArrayOutputStream
-import java.net.{InetAddress, URLDecoder}
-
-import com.fasterxml.jackson.databind.ObjectMapper
-import com.fasterxml.jackson.module.scala.DefaultScalaModule
-import org.apache.amaterasu.common.dataobjects.{ExecData, TaskData}
-import org.apache.amaterasu.common.logging.Logging
-import org.apache.amaterasu.common.utils.ActiveNotifier
-import org.apache.amaterasu.executor.common.executors.ProvidersFactory
-
-import scala.collection.JavaConverters._
-
-
-class ActionsExecutor extends Logging {
-
-  var master: String = _
-  var jobId: String = _
-  var actionName: String = _
-  var taskData: TaskData = _
-  var execData: ExecData = _
-  var providersFactory: ProvidersFactory = _
-
-  def execute(): Unit = {
-    val runner = providersFactory.getRunner(taskData.getGroupId, taskData.getTypeId)
-    runner match {
-      case Some(r) => {
-        try {
-          r.executeSource(taskData.getSrc, actionName, taskData.getExports)
-          log.info("Completed action")
-          System.exit(0)
-        } catch {
-          case e: Exception => {
-            log.error("Exception in execute source", e)
-            System.exit(100)
-          }
-        }
-      }
-      case None =>
-        log.error("", s"Runner not found for group: ${taskData.getGroupId}, type ${taskData.getTypeId}. Please verify the tasks")
-        System.exit(101)
-    }
-  }
-}
-
-object ActionsExecutorLauncher extends Logging with App {
-
-  val hostName = InetAddress.getLocalHost.getHostName
-
-  println(s"Hostname resolved to: $hostName")
-  val mapper = new ObjectMapper()
-  mapper.registerModule(DefaultScalaModule)
-
-  log.info("Starting actions executor")
-
-  val jobId = this.args(0)
-  val master = this.args(1)
-  val actionName = this.args(2)
-  val notificationsAddress = this.args(6)
-
-  log.info("parsing task data")
-  val taskData = mapper.readValue(URLDecoder.decode(this.args(3), "UTF-8"), classOf[TaskData])
-  log.info("parsing executor data")
-  val execData = mapper.readValue(URLDecoder.decode(this.args(4), "UTF-8"), classOf[ExecData])
-  val taskIdAndContainerId = this.args(5)
-
-  val actionsExecutor: ActionsExecutor = new ActionsExecutor
-  actionsExecutor.master = master
-  actionsExecutor.actionName = actionName
-  actionsExecutor.taskData = taskData
-  actionsExecutor.execData = execData
-
-  log.info("Setup executor")
-  val baos = new ByteArrayOutputStream()
-  val notifier = new ActiveNotifier(notificationsAddress)
-
-  actionsExecutor.providersFactory = ProvidersFactory(execData, jobId, baos, notifier, taskIdAndContainerId, hostName, propFile = "./amaterasu.properties")
-  actionsExecutor.execute()
-}
diff --git a/executor/src/test/resources/codegen.pyc b/executor/src/test/resources/codegen.pyc
deleted file mode 100644
index ab32046..0000000
--- a/executor/src/test/resources/codegen.pyc
+++ /dev/null
Binary files differ
diff --git a/frameworks/python/dispatcher/build.gradle b/frameworks/python/dispatcher/build.gradle
index b1cbd39..35ab5aa 100644
--- a/frameworks/python/dispatcher/build.gradle
+++ b/frameworks/python/dispatcher/build.gradle
@@ -71,7 +71,8 @@
 dependencies {
     compile project(':common')
     compile project(':amaterasu-sdk')
-    compile project(':leader')
+    compile project(':leader-common')
+    
     compile "org.jetbrains.kotlin:kotlin-stdlib-jdk8:$kotlin_version"
     compile "org.jetbrains.kotlin:kotlin-reflect"
     compile 'com.uchuhimo:konf:0.11'
diff --git a/frameworks/python/dispatcher/src/main/kotlin/org/apache/amaterasu/frameworks/python/dispatcher/PythonSetupProvider.kt b/frameworks/python/dispatcher/src/main/kotlin/org/apache/amaterasu/frameworks/python/dispatcher/PythonSetupProvider.kt
index 9d83c65..d5a691e 100644
--- a/frameworks/python/dispatcher/src/main/kotlin/org/apache/amaterasu/frameworks/python/dispatcher/PythonSetupProvider.kt
+++ b/frameworks/python/dispatcher/src/main/kotlin/org/apache/amaterasu/frameworks/python/dispatcher/PythonSetupProvider.kt
@@ -30,18 +30,14 @@
     private var conf: ClusterConfig? = null
     private var runnerProviders: Map<String, RunnerSetupProvider> = mapOf()
 
-    override val groupIdentifier: String
-        get() = "python"
-    override val groupResources: Array<File>
-        get() = arrayOf()
+    override val groupIdentifier: String = "python"
+    override val groupResources: List<File> = listOf()
 
     override fun getDriverConfiguration(configManager: ConfigManager): DriverConfiguration {
         return DriverConfiguration(conf!!.taskMem(), 1) //TODO: this should be configured on env level
     }
-    override val environmentVariables: Map<String, String>
-        get() = mapOf()
-    override val configurationItems: Array<String>
-        get() = arrayOf()
+    override val environmentVariables: Map<String, String> = mapOf()
+    override val configurationItems: List<String> = listOf()
 
     override fun init(env: String, conf: ClusterConfig) {
         this.env = env
diff --git a/frameworks/spark/dispatcher/src/main/kotlin/org/apache/amaterasu/frameworks/spark/dispatcher/SparkSetupProvider.kt b/frameworks/spark/dispatcher/src/main/kotlin/org/apache/amaterasu/frameworks/spark/dispatcher/SparkSetupProvider.kt
index e3e503f..7a5a42d 100644
--- a/frameworks/spark/dispatcher/src/main/kotlin/org/apache/amaterasu/frameworks/spark/dispatcher/SparkSetupProvider.kt
+++ b/frameworks/spark/dispatcher/src/main/kotlin/org/apache/amaterasu/frameworks/spark/dispatcher/SparkSetupProvider.kt
@@ -54,11 +54,11 @@
         }
     }
 
-    override val groupResources: Array<File> by lazy {
+    override val groupResources: List<File> by lazy {
         when (conf.mode()) {
-            "mesos" -> arrayOf(File("spark-${conf.webserver().sparkVersion()}.tgz"), File("spark-runtime-${conf.version()}.jar"))
-            "yarn" -> arrayOf(File("spark-runtime-${conf.version()}.jar"), File("executor-${conf.version()}-all.jar"), File(conf.spark().home()))
-            else -> arrayOf()
+            "mesos" -> listOf(File("spark-${conf.webserver().sparkVersion()}.tgz"), File("spark-runtime-${conf.version()}-all.jar"))
+            "yarn" -> listOf(File("spark-runtime-${conf.version()}-all.jar"),  File(conf.spark().home()))
+            else -> listOf()
         }
     }
 
@@ -67,7 +67,7 @@
     }
 
     override val groupIdentifier: String = "spark"
-    override val configurationItems = arrayOf("sparkProperties", "sparkOptions")
+    override val configurationItems = listOf("sparkProperties", "sparkOptions")
 
     override fun getDriverConfiguration(configManager: ConfigManager): DriverConfiguration {
         val sparkOptions: Map<String, Any> = configManager.config["sparkOptions"]
diff --git a/frameworks/spark/dispatcher/src/main/kotlin/org/apache/amaterasu/frameworks/spark/dispatcher/runners/providers/SparkSubmitScalaRunnerProvider.kt b/frameworks/spark/dispatcher/src/main/kotlin/org/apache/amaterasu/frameworks/spark/dispatcher/runners/providers/SparkSubmitScalaRunnerProvider.kt
index f413fd1..a1b7a24 100644
--- a/frameworks/spark/dispatcher/src/main/kotlin/org/apache/amaterasu/frameworks/spark/dispatcher/runners/providers/SparkSubmitScalaRunnerProvider.kt
+++ b/frameworks/spark/dispatcher/src/main/kotlin/org/apache/amaterasu/frameworks/spark/dispatcher/runners/providers/SparkSubmitScalaRunnerProvider.kt
@@ -37,7 +37,7 @@
                 SparkCommandLineHelper.getOptions(sparkOptions) + " " +
                 SparkCommandLineHelper.getProperties(sparkProperties) + " " +
                 master +
-                " --jars spark-runtime-${conf.version()}.jar >&1"
+                " --jars spark-runtime-${conf.version()}-all.jar >&1"
     }
 
     override fun getActionUserResources(jobId: String, actionData: ActionData): Array<String> = arrayOf()
diff --git a/frameworks/spark/runtime/build.gradle b/frameworks/spark/runtime/build.gradle
index 3bfd7d4..1f2659d 100644
--- a/frameworks/spark/runtime/build.gradle
+++ b/frameworks/spark/runtime/build.gradle
@@ -54,8 +54,7 @@
 }
 
 dependencies {
-
-    compile project(':executor')
+    compile project(':common')
     provided('org.apache.spark:spark-repl_2.11:2.2.1')
     provided('org.apache.spark:spark-core_2.11:2.2.1')
 
@@ -81,7 +80,8 @@
 }
 
 task copyToHome(type: Copy) {
-    dependsOn jar
+    dependsOn shadowJar
+
     from 'build/libs'
     into '../../../build/amaterasu/dist'
     from 'build/resources/main'
diff --git a/leader-common/build.gradle b/leader-common/build.gradle
index 928a444..b2734cf 100644
--- a/leader-common/build.gradle
+++ b/leader-common/build.gradle
@@ -34,7 +34,6 @@
 
 plugins {
     id "com.github.johnrengelman.shadow" version "2.0.4"
-    id 'scala'
 }
 
 apply plugin: 'kotlin'
@@ -66,12 +65,10 @@
 }
 
 dependencies {
-    compile 'org.scala-lang:scala-library:2.11.8'
 
     compile project(':common')
     compile project(':amaterasu-sdk')
 
-    compile group: 'com.fasterxml.jackson.module', name: 'jackson-module-scala_2.11', version: '2.6.3'
     compile group: 'com.fasterxml.jackson.core', name: 'jackson-core', version: '2.9.8'
     compile group: 'com.fasterxml.jackson.core', name: 'jackson-annotations', version: '2.9.8'
     compile group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: '2.9.8'
@@ -87,6 +84,16 @@
     compile group: 'com.beust', name: 'klaxon', version: '5.0.1'
     compile group: 'com.github.ajalt', name: 'clikt', version: '1.6.0'
 
+    compile group: 'org.eclipse.jetty', name: 'jetty-plus', version: '9.4.14.v20181114'
+    compile group: 'org.eclipse.jetty', name: 'jetty-server', version: '9.4.14.v20181114'
+    compile group: 'org.eclipse.jetty', name: 'jetty-http', version: '9.4.14.v20181114'
+    compile group: 'org.eclipse.jetty', name: 'jetty-io', version: '9.4.14.v20181114'
+    compile group: 'org.eclipse.jetty', name: 'jetty-servlet', version: '9.4.14.v20181114'
+    compile group: 'org.eclipse.jetty.toolchain', name: 'jetty-test-helper', version: '4.0'
+
+    compile group: 'org.jsoup', name: 'jsoup', version: '1.10.2'
+    //compile group: 'io.ktor', name: 'ktor-client-core', version: '0.9.5'
+
     compile "org.jetbrains.kotlin:kotlin-stdlib-jdk8:$kotlin_version"
     compile "org.jetbrains.kotlin:kotlin-reflect"
 
@@ -116,3 +123,8 @@
 compileTestKotlin {
     kotlinOptions.jvmTarget = "1.8"
 }
+
+task copyToHome(type: Copy) {
+    from 'src/main/scripts'
+    into '../build/amaterasu/'
+}
\ No newline at end of file
diff --git a/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/execution/JobLoader.kt b/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/execution/JobLoader.kt
index 0a2863b..0f63e1d 100644
--- a/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/execution/JobLoader.kt
+++ b/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/execution/JobLoader.kt
@@ -90,7 +90,7 @@
 
             val status = ActionStatus.valueOf(String(client.data.forPath("/$jobId/$task")))
             if (status == ActionStatus.Queued || status == ActionStatus.Started) {
-                jobManager.reQueueAction(task.substring(task.indexOf("task-") + 5))
+                jobManager.requeueAction(task.substring(task.indexOf("task-") + 5))
             }
 
         }
diff --git a/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/execution/JobManager.kt b/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/execution/JobManager.kt
index 1853e8c..3c77ba2 100644
--- a/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/execution/JobManager.kt
+++ b/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/execution/JobManager.kt
@@ -72,7 +72,7 @@
             return nextAction
         }
 
-    fun reQueueAction(actionId: String) {
+    fun requeueAction(actionId: String) {
 
         log.info("requeing action $actionId")
         registeredActions.forEach { log.info("key ${it.key}") }
@@ -124,14 +124,14 @@
 
         val action = registeredActions[actionId]
         val id = action!!.handleFailure(message)
-        if (!id.isEmpty())
+        if (id.isNotEmpty())
             registeredActions[id]?.execute()
 
         //delete all future actions
         cancelFutureActions(action)
     }
 
-    fun cancelFutureActions(action: Action) {
+    private fun cancelFutureActions(action: Action) {
 
         if (action.data.status != ActionStatus.Failed)
             action.announceCanceled()
diff --git a/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/execution/frameworls/FrameworkProvidersFactory.kt b/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/execution/frameworks/FrameworkProvidersFactory.kt
similarity index 75%
rename from leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/execution/frameworls/FrameworkProvidersFactory.kt
rename to leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/execution/frameworks/FrameworkProvidersFactory.kt
index 48f619a..18a2fc8 100644
--- a/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/execution/frameworls/FrameworkProvidersFactory.kt
+++ b/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/execution/frameworks/FrameworkProvidersFactory.kt
@@ -14,24 +14,26 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.amaterasu.leader.common.execution.frameworls
+package org.apache.amaterasu.leader.common.execution.frameworks
 
 import org.apache.amaterasu.common.configuration.ClusterConfig
-import org.apache.amaterasu.common.configuration.ConfigManager
 import org.apache.amaterasu.common.logging.KLogging
 import org.apache.amaterasu.sdk.frameworks.FrameworkSetupProvider
 import org.reflections.Reflections
 
 class FrameworkProvidersFactory(val env: String, val config: ClusterConfig) : KLogging() {
 
-    var  providers: Map<String, FrameworkSetupProvider>
+    val providers: MutableMap<String, FrameworkSetupProvider> = mutableMapOf()
+
+    val groups: List<String> by lazy { providers.keys.toList() }
+
+    fun getFramework(groupId: String): FrameworkSetupProvider = providers.getValue(groupId)
 
     init {
-        val reflections =  Reflections(ClassLoader::class.java)
+        val reflections =  Reflections(this::class.java.classLoader)
         val runnerTypes = reflections.getSubTypesOf(FrameworkSetupProvider::class.java)
 
-
-        providers = runnerTypes.map  {
+        val loadedProviders = runnerTypes.map {
 
             val provider = it.newInstance()
 
@@ -41,5 +43,7 @@
             provider.groupIdentifier to provider
 
         }.toMap()
+
+        providers.putAll(loadedProviders)
     }
 }
\ No newline at end of file
diff --git a/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/utilities/ActiveReportListener.kt b/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/utilities/ActiveReportListener.kt
index 6bda13c..f73d717 100644
--- a/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/utilities/ActiveReportListener.kt
+++ b/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/utilities/ActiveReportListener.kt
@@ -41,7 +41,7 @@
         else -> println("===> Unknown message")
     }
 
-    private fun printNotification(notification: Notification) = when (notification.notType) {
+    fun printNotification(notification: Notification) = when (notification.notType) {
 
         NotificationType.Info ->
             println("===> ${notification.msg} ".brightWhite().bold())
diff --git a/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/utilities/HttpServer.kt b/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/utilities/HttpServer.kt
new file mode 100644
index 0000000..5b6563f
--- /dev/null
+++ b/leader-common/src/main/kotlin/org/apache/amaterasu/leader/common/utilities/HttpServer.kt
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.amaterasu.leader.common.utilities
+
+import org.apache.log4j.BasicConfigurator
+import org.apache.log4j.Level
+import org.eclipse.jetty.server.Server
+import org.eclipse.jetty.server.ServerConnector
+import org.eclipse.jetty.server.handler.DefaultHandler
+import org.eclipse.jetty.server.handler.HandlerList
+import org.eclipse.jetty.server.handler.ResourceHandler
+import org.jsoup.Jsoup
+import org.jsoup.select.Elements
+import org.apache.log4j.Logger
+import org.eclipse.jetty.util.log.StdErrLog
+
+class HttpServer {
+    lateinit var server: Server
+
+    fun start(port: String, serverRoot: String) {
+        BasicConfigurator.configure()
+        initLogging()
+
+        server = Server()
+        val connector = ServerConnector(server)
+        connector.port = port.toInt()
+        server.addConnector(connector)
+
+        val handler = ResourceHandler()
+        handler.isDirectoriesListed = true
+        handler.welcomeFiles = arrayOf("index.html")
+        handler.resourceBase = serverRoot
+        val handlers = HandlerList()
+        handlers.handlers = arrayOf(handler, DefaultHandler())
+
+        server.handler = handlers
+        server.start()
+    }
+
+    private fun initLogging() {
+        System.setProperty("org.eclipse.jetty.util.log.class", StdErrLog::javaClass.name)
+        Logger.getLogger("org.eclipse.jetty").level = Level.ALL
+        Logger.getLogger("org.eclipse.jetty.websocket").level = Level.OFF
+    }
+
+    /**
+     * Method: getFilesInDirectory
+     * Description: provides a list of files in the given directory URL.
+     * @Params: amaNode: Hostname of the URL, port: Port # of the host, directory: destination directory to fetch files
+     * Note: Should the files in URL root be fetched, provide an empty value to directory.
+     */
+    fun getFilesInDirectory(amaNode: String, port: String, directory: String = ""): Array<String> {
+
+        val htmlDoc = Jsoup.connect("http://$amaNode:$port/$directory").get()
+        val files: Elements = htmlDoc.body().select("a")
+        return files.map { it.attr("href") }
+                .filter { !it.contains("..") }
+                .map { it.replace("/", "") }.toTypedArray()
+    }
+
+    fun stop() {
+        if (::server.isInitialized) {
+            server.stop()
+        }
+    }
+
+}
\ No newline at end of file
diff --git a/leader-common/src/main/scala/org/apache/amaterasu/leader/common/execution/frameworks/FrameworkProvidersFactory.scala b/leader-common/src/main/scala/org/apache/amaterasu/leader/common/execution/frameworks/FrameworkProvidersFactory.scala
deleted file mode 100644
index a748690..0000000
--- a/leader-common/src/main/scala/org/apache/amaterasu/leader/common/execution/frameworks/FrameworkProvidersFactory.scala
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.leader.common.execution.frameworks
-
-import org.apache.amaterasu.common.configuration.ClusterConfig
-import org.apache.amaterasu.common.logging.Logging
-import org.apache.amaterasu.sdk.frameworks.FrameworkSetupProvider
-import org.reflections.Reflections
-
-import scala.collection.JavaConversions._
-
-class FrameworkProvidersFactory {
-  var providers: Map[String, FrameworkSetupProvider] = _
-
-  def groups: Array[String] = {
-    providers.keys.toArray
-  }
-
-  def getFramework(groupId: String): FrameworkSetupProvider = {
-    providers(groupId)
-  }
-
-}
-
-object FrameworkProvidersFactory extends Logging {
-
-  def apply(env: String, config: ClusterConfig): FrameworkProvidersFactory = {
-
-    val result = new FrameworkProvidersFactory()
-
-    val reflections = new Reflections(getClass.getClassLoader)
-    val runnerTypes = reflections.getSubTypesOf(classOf[FrameworkSetupProvider]).toSet
-
-    result.providers = runnerTypes.map(r => {
-
-      val provider = Manifest.classType(r).runtimeClass.newInstance.asInstanceOf[FrameworkSetupProvider]
-
-      provider.init(env, config)
-      log.info(s"a provider for group ${provider.getGroupIdentifier} was created")
-      log.info(s"config = $config")
-      (provider.getGroupIdentifier, provider)
-
-    }).toMap
-
-    result
-  }
-}
\ No newline at end of file
diff --git a/leader/src/main/scripts/amaterasu.properties b/leader-common/src/main/scripts/amaterasu.properties
similarity index 100%
rename from leader/src/main/scripts/amaterasu.properties
rename to leader-common/src/main/scripts/amaterasu.properties
diff --git a/leader/src/main/scripts/log4j.properties b/leader-common/src/main/scripts/log4j.properties
similarity index 100%
rename from leader/src/main/scripts/log4j.properties
rename to leader-common/src/main/scripts/log4j.properties
diff --git a/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/configuration/ConfigManagerTests.kt b/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/configuration/ConfigManagerTests.kt
index a7b2e47..35c4b0e 100644
--- a/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/configuration/ConfigManagerTests.kt
+++ b/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/configuration/ConfigManagerTests.kt
@@ -16,7 +16,6 @@
  */
 package org.apache.amaterasu.leader.common.configuration
 
-import com.uchuhimo.konf.source.yaml.toYaml
 import org.apache.amaterasu.common.configuration.ConfigManager
 import org.apache.amaterasu.common.configuration.Job
 import org.jetbrains.spek.api.Spek
diff --git a/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/dsl/ActionStatusTests.kt b/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/dsl/ActionStatusTests.kt
new file mode 100644
index 0000000..c7c3d99
--- /dev/null
+++ b/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/dsl/ActionStatusTests.kt
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.amaterasu.leader.common.dsl
+
+import org.apache.amaterasu.common.configuration.enums.ActionStatus
+import org.apache.amaterasu.common.dataobjects.ActionData
+import org.apache.amaterasu.leader.common.execution.actions.SequentialAction
+import org.apache.curator.framework.CuratorFrameworkFactory
+import org.apache.curator.retry.ExponentialBackoffRetry
+import org.apache.curator.test.TestingServer
+import org.apache.zookeeper.CreateMode
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.given
+import org.jetbrains.spek.api.dsl.it
+import java.util.concurrent.LinkedBlockingQueue
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+
+class ActionStatusTests : Spek({
+
+    // setting up a testing zookeeper server (curator TestServer)
+    val retryPolicy = ExponentialBackoffRetry(1000, 3)
+    val server = TestingServer(2181, true)
+    val jobId = "job_${System.currentTimeMillis()}"
+
+
+    given("an action") {
+
+        val data = ActionData(ActionStatus.Pending, "test_action", "start.scala", "", "spark", "scala", "0000001", hashMapOf(), mutableListOf())
+        val queue = LinkedBlockingQueue<ActionData>()
+
+        val client = CuratorFrameworkFactory.newClient(server.connectString, retryPolicy)
+        client.start()
+
+        client.create().withMode(CreateMode.PERSISTENT).forPath("/$jobId")
+        val action = SequentialAction(data.name, data.src, "", data.groupId, data.typeId, mapOf(), jobId, queue, client, 1)
+
+        it("should queue it's ActionData int the job queue when executed") {
+            action.execute()
+            assertEquals(queue.peek().name, data.name)
+            assertEquals(queue.peek().src, data.src)
+        }
+
+        it("should also create a sequential znode for the task with the value of Queued") {
+            val taskStatus = client.data.forPath("/$jobId/task-0000000000")
+
+            assertNotNull(taskStatus)
+            assertEquals(String(taskStatus), "Queued")
+
+        }
+    }
+})
\ No newline at end of file
diff --git a/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/dsl/GitUtilTests.kt b/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/dsl/GitUtilTests.kt
index 3b9a229..db36abf 100644
--- a/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/dsl/GitUtilTests.kt
+++ b/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/dsl/GitUtilTests.kt
@@ -1,3 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package org.apache.amaterasu.leader.common.dsl
 
 import org.jetbrains.spek.api.Spek
diff --git a/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/dsl/JobExecutionTests.kt b/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/dsl/JobExecutionTests.kt
new file mode 100644
index 0000000..af3c9d9
--- /dev/null
+++ b/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/dsl/JobExecutionTests.kt
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.amaterasu.leader.common.dsl
+
+import org.apache.amaterasu.common.dataobjects.ActionData
+import org.apache.curator.framework.CuratorFrameworkFactory
+import org.apache.curator.retry.ExponentialBackoffRetry
+import org.apache.curator.test.TestingServer
+import org.apache.zookeeper.CreateMode
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.given
+import org.jetbrains.spek.api.dsl.it
+import org.jetbrains.spek.api.dsl.on
+import java.util.concurrent.LinkedBlockingQueue
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+class JobExecutionTests : Spek({
+    val retryPolicy = ExponentialBackoffRetry(1000, 3)
+    val server = TestingServer(2183, true)
+    val client = CuratorFrameworkFactory.newClient(server.connectString, retryPolicy)
+    client.start()
+
+    val jobId = "job_${System.currentTimeMillis()}"
+
+    val yaml = this::class.java.getResource("/simple-maki.yml").readText()
+    val queue = LinkedBlockingQueue<ActionData>()
+    client.create().withMode(CreateMode.PERSISTENT).forPath("/$jobId")
+
+    given("a job parsed in to a JobManager") {
+        val job = JobParser.parse(jobId, yaml, queue, client, 1)
+
+        it("queue the first action when the JobManager.start method is called ") {
+            job.start()
+
+            assertEquals(queue.peek().name, "start")
+
+            // making sure that the status is reflected in zk
+            val actionStatus = client.data.forPath("/$jobId/task-0000000000")
+            assertEquals(String(actionStatus), "Queued")
+        }
+
+        it("return the start action when calling getNextAction and dequeue it") {
+            assertEquals(job.nextActionData!!.name, "start")
+            assertEquals(queue.size, 0)
+        }
+
+        it("also changes the status of start to started") {
+            val actionStatus = client.data.forPath("/$jobId/task-0000000000")
+            assertEquals(String(actionStatus), "Started")
+        }
+
+        it("is not out of actions when an action is still Pending") {
+            assertFalse { job.outOfActions }
+        }
+
+        it("will requeue a task when calling JobManager.requeueAction") {
+            job.requeueAction("0000000000")
+            val actionStatus = client.data.forPath("/$jobId/task-0000000000")
+            assertEquals(String(actionStatus), "Queued")
+        }
+
+        it("will restart the task") {
+            val data = job.nextActionData
+            assertEquals(data!!.name, "start")
+
+            // making sure that the status is reflected in zk
+            val actionStatus = client.data.forPath("/$jobId/task-0000000000")
+            assertEquals(String(actionStatus), "Started")
+        }
+
+        it("will mark the action as Complete when the actionComplete method is called") {
+            job.actionComplete("0000000000")
+            // making sure that the status is reflected in zk
+            val actionStatus = client.data.forPath("/$jobId/task-0000000000")
+
+            assertEquals(String(actionStatus), "Complete")
+        }
+
+        on("completion of start, the next action step2 is queued") {
+            assertEquals(queue.peek().name, "step2")
+        }
+
+        it("should also be registered as queued in zk") {
+            val actionStatus = client.data.forPath("/$jobId/task-0000000001")
+            assertEquals(String(actionStatus), "Queued")
+        }
+
+        it("is marked as Started when JobManager.nextActionData is called") {
+            val data = job.nextActionData
+            assertEquals(data!!.name, "step2")
+        }
+
+        it("will start an error action JobManager.actionFailed is called") {
+            job.actionFailed("0000000001", "test failure")
+
+            assertEquals(queue.peek().name, "error-action")
+        }
+
+        it("marks the error action as Complete when the actionComplete method is called"){
+            job.actionComplete("0000000001-error")
+
+            // making sure that the status is reflected in zk
+            val actionStatus = client.data.forPath("/$jobId/task-0000000001-error")
+            assertEquals(String(actionStatus) ,"Complete")
+        }
+
+        it("will be out of actions when all actions have been executed"){
+            assertTrue { job.outOfActions }
+        }
+    }
+
+})
\ No newline at end of file
diff --git a/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/dsl/JobParserTests.kt b/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/dsl/JobParserTests.kt
new file mode 100644
index 0000000..9e4f53b
--- /dev/null
+++ b/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/dsl/JobParserTests.kt
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.amaterasu.leader.common.dsl
+
+import org.apache.amaterasu.common.dataobjects.ActionData
+import org.apache.curator.framework.CuratorFrameworkFactory
+import org.apache.curator.retry.ExponentialBackoffRetry
+import org.apache.curator.test.TestingServer
+import org.apache.zookeeper.CreateMode
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.given
+import org.jetbrains.spek.api.dsl.it
+import java.util.concurrent.LinkedBlockingQueue
+import kotlin.test.assertEquals
+
+class JobParserTests : Spek({
+
+    val retryPolicy = ExponentialBackoffRetry(1000, 3)
+    val server = TestingServer(2187, true)
+    val client = CuratorFrameworkFactory.newClient(server.connectString, retryPolicy)
+    client.start()
+
+    val jobId = "job_${System.currentTimeMillis()}"
+    val queue = LinkedBlockingQueue<ActionData>()
+
+    // this will be performed by the job bootstrapper
+    client.create().withMode(CreateMode.PERSISTENT).forPath("/$jobId")
+
+    given("a simple-maki.yaml that is passed to a JobParser") {
+
+        val yaml = this::class.java.getResource("/simple-maki.yml").readText()
+        val job = JobParser.parse(jobId, yaml, queue, client, 1)
+
+        it("parse the job details correctly") {
+            assertEquals(job.name, "amaterasu-test")
+        }
+
+        it("also have two actions in the right order") {
+
+            assertEquals(job.registeredActions.size, 3)
+
+            assertEquals(job.registeredActions["0000000000"]!!.data.name, "start")
+            assertEquals(job.registeredActions["0000000001"]!!.data.name, "step2")
+            assertEquals(job.registeredActions["0000000001-error"]!!.data.name, "error-action")
+
+        }
+
+        it("also also parse action 'config' successfully") {
+            assertEquals(job.registeredActions["0000000000"]!!.data.config, "start-cfg.yaml")
+            assertEquals(job.registeredActions["0000000001"]!!.data.config, "step2-cfg.yaml")
+            assertEquals(job.registeredActions["0000000001-error"]!!.data.config, "error-cfg.yaml")
+        }
+    }
+})
\ No newline at end of file
diff --git a/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/utilities/HttpServerTests.kt b/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/utilities/HttpServerTests.kt
new file mode 100644
index 0000000..c79a819
--- /dev/null
+++ b/leader-common/src/test/kotlin/org/apache/amaterasu/leader/common/utilities/HttpServerTests.kt
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.amaterasu.leader.common.utilities
+
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.given
+import org.jetbrains.spek.api.dsl.it
+import org.jsoup.Jsoup
+import kotlin.test.assertEquals
+
+class HttpServerTests : Spek({
+    val resources = this::class.java.getResource("/jetty-test-data.txt").path.removeSuffix("/jetty-test-data.txt")
+
+    given("an Amaterasu Web server is started") {
+        val server = HttpServer()
+        server.start("8002", resources)
+
+        it("serve content and stop successfully") {
+            lateinit var data: String
+            try {
+                val html = Jsoup.connect("http://localhost:8002/jetty-test-data.txt").get().select("body").text()
+                data = html.toString()
+            } finally {
+                server.stop()
+            }
+            assertEquals(data, "This is a test file to download from Jetty webserver")
+        }
+    }
+
+    given("an Amaterasu Web server is started pointing to a sub dir with two files") {
+        val server = HttpServer()
+        server.start("8001", "$resources/dist")
+
+        it("list the files correctly") {
+            try {
+                val urls = server.getFilesInDirectory("localhost", "8001", "")
+                assertEquals(urls.size, 2)
+            } finally {
+                server.stop()
+            }
+        }
+    }
+})
\ No newline at end of file
diff --git a/leader/src/test/resources/dist/jetty-fileserver-test-file-1.txt b/leader-common/src/test/resources/dist/jetty-fileserver-test-file-1.txt
similarity index 100%
rename from leader/src/test/resources/dist/jetty-fileserver-test-file-1.txt
rename to leader-common/src/test/resources/dist/jetty-fileserver-test-file-1.txt
diff --git a/leader/src/test/resources/dist/jetty-fileserver-test-file-2.txt b/leader-common/src/test/resources/dist/jetty-fileserver-test-file-2.txt
similarity index 100%
rename from leader/src/test/resources/dist/jetty-fileserver-test-file-2.txt
rename to leader-common/src/test/resources/dist/jetty-fileserver-test-file-2.txt
diff --git a/leader/src/test/resources/jetty-test-data.txt b/leader-common/src/test/resources/jetty-test-data.txt
similarity index 100%
rename from leader/src/test/resources/jetty-test-data.txt
rename to leader-common/src/test/resources/jetty-test-data.txt
diff --git a/leader/src/test/resources/simple-maki.yml b/leader-common/src/test/resources/simple-maki.yml
similarity index 100%
rename from leader/src/test/resources/simple-maki.yml
rename to leader-common/src/test/resources/simple-maki.yml
diff --git a/leader-mesos/build.gradle b/leader-mesos/build.gradle
index cb08d6a..e0ade9f 100644
--- a/leader-mesos/build.gradle
+++ b/leader-mesos/build.gradle
@@ -14,7 +14,93 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+buildscript {
 
+    repositories {
+        mavenCentral()
+        maven {
+            url 'http://repository.jetbrains.com/all'
+        }
+        maven {
+            url "https://jetbrains.jfrog.io/jetbrains/spek-snapshots"
+        }
+    }
 
+    dependencies {
+        classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:$kotlin_version"
+        classpath 'org.junit.platform:junit-platform-gradle-plugin:1.0.0'
+    }
+}
 
+plugins {
+    id "com.github.johnrengelman.shadow" version "2.0.4"
+}
 
+apply plugin: 'kotlin'
+apply plugin: 'org.junit.platform.gradle.plugin'
+
+junitPlatform {
+    filters {
+        engines {
+            include 'spek'
+        }
+    }
+}
+
+sourceCompatibility = 1.8
+targetCompatibility = 1.8
+
+shadowJar {
+    zip64 true
+}
+
+repositories {
+    maven { url "https://plugins.gradle.org/m2/" }
+    maven { url 'http://repository.jetbrains.com/all' }
+    maven { url "https://jetbrains.jfrog.io/jetbrains/spek-snapshots" }
+    maven { url "http://dl.bintray.com/jetbrains/spek" }
+    maven { url "http://oss.jfrog.org/artifactory/oss-snapshot-local" }
+
+    mavenCentral()
+    jcenter()
+}
+
+dependencies {
+    compile project(':leader-common')
+    compile project(':amaterasu-sdk')
+
+    compile "org.jetbrains.kotlin:kotlin-stdlib-jdk8:$kotlin_version"
+    compile "org.jetbrains.kotlin:kotlin-reflect"
+    compile group: 'org.jetbrains.kotlinx', name: 'kotlinx-coroutines-core', version: '1.1.1'
+
+    testCompile 'org.jetbrains.spek:spek-api:1.1.5'
+    testCompile "org.jetbrains.kotlin:kotlin-test-junit:$kotlin_version"
+    testRuntime 'org.jetbrains.spek:spek-junit-platform-engine:1.1.5'
+
+    // Spek requires kotlin-reflect, can be omitted if already in the classpath
+    testRuntimeOnly "org.jetbrains.kotlin:kotlin-reflect:$kotlin_version"
+}
+
+task copyToHomeRoot(type: Copy) {
+    from 'src/main/scripts'
+    into '../build/amaterasu/'
+}
+
+task copyToHomeBin(type: Copy) {
+    dependsOn shadowJar
+    from 'build/libs'
+    into '../build/amaterasu/bin'
+}
+
+task copyToHome() {
+    dependsOn copyToHomeRoot
+    dependsOn copyToHomeBin
+}
+
+compileKotlin{
+    kotlinOptions.jvmTarget = "1.8"
+}
+
+compileTestKotlin {
+    kotlinOptions.jvmTarget = "1.8"
+}
diff --git a/leader-mesos/src/main/kotlin/org/apache/amaterasu/leader/mesos/Client.kt b/leader-mesos/src/main/kotlin/org/apache/amaterasu/leader/mesos/Client.kt
new file mode 100644
index 0000000..d2209cf
--- /dev/null
+++ b/leader-mesos/src/main/kotlin/org/apache/amaterasu/leader/mesos/Client.kt
@@ -0,0 +1,8 @@
+package org.apache.amaterasu.leader.mesos
+
+object Client {
+
+    @Throws(Exception::class)
+    @JvmStatic
+    fun main(args: Array<String>) = ClientArgsParser().main(args)
+}
\ No newline at end of file
diff --git a/leader-mesos/src/main/kotlin/org/apache/amaterasu/leader/mesos/ClientArgsParser.kt b/leader-mesos/src/main/kotlin/org/apache/amaterasu/leader/mesos/ClientArgsParser.kt
new file mode 100644
index 0000000..6b9a73e
--- /dev/null
+++ b/leader-mesos/src/main/kotlin/org/apache/amaterasu/leader/mesos/ClientArgsParser.kt
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.amaterasu.leader.mesos
+
+import org.apache.amaterasu.common.configuration.ClusterConfig
+import org.apache.amaterasu.leader.common.launcher.AmaOpts
+import org.apache.amaterasu.leader.common.launcher.ArgsParser
+import org.apache.log4j.LogManager
+import org.apache.mesos.MesosSchedulerDriver
+import org.apache.mesos.Protos
+import java.io.FileInputStream
+
+class ClientArgsParser : ArgsParser() {
+
+    override fun run() {
+
+        val opts = AmaOpts(repo, branch, env, name, jobId, newJobId, report, home)
+
+        val config = ClusterConfig.apply(FileInputStream("${opts.home}/amaterasu.properties"))
+        val resume = opts.jobId.isNotEmpty()
+
+        LogManager.resetConfiguration()
+
+        val frameworkBuilder = Protos.FrameworkInfo.newBuilder()
+                .setName("${opts.name} - Amaterasu Job")
+                .setFailoverTimeout(config.timeout())
+                .setUser(config.user())
+
+        if (resume) {
+            frameworkBuilder.setId(Protos.FrameworkID.newBuilder().setValue(opts.jobId))
+        }
+
+        val framework = frameworkBuilder.build()
+
+        val masterAddress = "${config.master()}:${config.masterPort()}"
+
+        val scheduler = JobScheduler(
+                opts.repo,
+                opts.branch,
+                opts.userName,
+                opts.password,
+                opts.env,
+                resume,
+                config,
+                opts.report,
+                opts.home
+        )
+
+        val driver = MesosSchedulerDriver(scheduler, framework, masterAddress)
+
+        driver.run()
+
+    }
+}
+
diff --git a/leader-mesos/src/main/kotlin/org/apache/amaterasu/leader/mesos/JobScheduler.kt b/leader-mesos/src/main/kotlin/org/apache/amaterasu/leader/mesos/JobScheduler.kt
new file mode 100644
index 0000000..f19d0fa
--- /dev/null
+++ b/leader-mesos/src/main/kotlin/org/apache/amaterasu/leader/mesos/JobScheduler.kt
@@ -0,0 +1,384 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.amaterasu.leader.mesos
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory
+import com.fasterxml.jackson.module.kotlin.readValue
+import com.fasterxml.jackson.module.kotlin.registerKotlinModule
+import org.apache.amaterasu.common.configuration.ClusterConfig
+import org.apache.amaterasu.common.configuration.ConfigManager
+import org.apache.amaterasu.common.dataobjects.ActionData
+import org.apache.amaterasu.common.execution.actions.Notification
+import org.apache.amaterasu.common.execution.actions.enums.NotificationLevel
+import org.apache.amaterasu.common.execution.actions.enums.NotificationType
+import org.apache.amaterasu.common.logging.KLogging
+import org.apache.amaterasu.leader.common.execution.JobLoader
+import org.apache.amaterasu.leader.common.execution.JobManager
+import org.apache.amaterasu.leader.common.execution.frameworks.FrameworkProvidersFactory
+import org.apache.amaterasu.leader.common.utilities.ActiveReportListener
+import org.apache.amaterasu.leader.common.utilities.DataLoader
+import org.apache.amaterasu.leader.common.utilities.HttpServer
+import org.apache.amaterasu.sdk.frameworks.FrameworkSetupProvider
+import org.apache.amaterasu.sdk.frameworks.RunnerSetupProvider
+import org.apache.commons.io.FileUtils
+import org.apache.curator.framework.CuratorFramework
+import org.apache.curator.framework.CuratorFrameworkFactory
+import org.apache.curator.retry.ExponentialBackoffRetry
+import org.apache.log4j.LogManager
+import org.apache.mesos.Protos
+import org.apache.mesos.Scheduler
+import org.apache.mesos.SchedulerDriver
+import org.apache.mesos.protobuf.ByteString
+import java.io.File
+import java.nio.file.Path
+import java.nio.file.Paths
+import java.util.*
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.LinkedBlockingQueue
+
+class JobScheduler(private val src: String,
+                   private val branch: String,
+                   private val username: String,
+                   private val password: String,
+                   private val env: String,
+                   private val resume: Boolean,
+                   private val config: ClusterConfig,
+                   private val report: String,
+                   private val home: String) : Scheduler, KLogging() {
+
+    private var client: CuratorFramework
+    private val server = HttpServer()
+    private val listener = ActiveReportListener()
+    private lateinit var jobManager: JobManager
+    private val mapper = ObjectMapper().registerKotlinModule()
+    private var reportLevel: NotificationLevel
+    private val offersToTaskIds = ConcurrentHashMap<String, String>()
+    private val taskIdsToActions = ConcurrentHashMap<Protos.TaskID, String>()
+    private lateinit var frameworkFactory: FrameworkProvidersFactory
+    private lateinit var configManager: ConfigManager
+    private val yamlMapper = ObjectMapper(YAMLFactory()).registerKotlinModule()
+
+    private val jarFile = File(this::class.java.protectionDomain.codeSource.location.path)
+    private val amaDist = File("${File(jarFile.parent).parent}/dist")
+
+
+    init {
+        LogManager.resetConfiguration()
+        server.start(config.webserver().Port(), "$home/${config.webserver().Root()}")
+
+        reportLevel = NotificationLevel.valueOf(report.capitalize())
+        val retryPolicy = ExponentialBackoffRetry(1000, 3)
+        client = CuratorFrameworkFactory.newClient(config.zk(), retryPolicy)
+        client.start()
+    }
+
+    /**
+     * creates a working dir for a job
+     * @param jobId: the id of the job (will be used as the jobs directory name as well)
+     */
+    private fun createJobDir(jobId: String) {
+        val amaHome = File(jarFile.parent).parent
+        val jobDir = "$amaHome/dist/$jobId/"
+
+        val dir = File(jobDir)
+        if (!dir.exists()) {
+            dir.mkdir()
+        }
+    }
+
+    override fun registered(driver: SchedulerDriver?, frameworkId: Protos.FrameworkID?, masterInfo: Protos.MasterInfo?) {
+        jobManager = if (!resume) {
+            JobLoader.loadJob(src,
+                    branch,
+                    frameworkId!!.value,
+                    username,
+                    password,
+                    client,
+                    config.jobs().tasks().attempts(),
+                    LinkedBlockingQueue<ActionData>())
+        } else {
+            JobLoader.reloadJob(frameworkId!!.value,
+                    username,
+                    password,
+                    client,
+                    config.jobs().tasks().attempts(),
+                    LinkedBlockingQueue<ActionData>())
+        }
+
+        jobManager.start()
+        createJobDir(jobManager.jobId)
+    }
+
+    override fun disconnected(driver: SchedulerDriver?) {}
+
+    override fun reregistered(driver: SchedulerDriver?, masterInfo: Protos.MasterInfo?) {}
+
+    override fun error(driver: SchedulerDriver?, message: String?) {
+        log.error(message)
+    }
+
+    override fun slaveLost(driver: SchedulerDriver?, slaveId: Protos.SlaveID?) {}
+
+    override fun executorLost(driver: SchedulerDriver?, executorId: Protos.ExecutorID?, slaveId: Protos.SlaveID?, status: Int) {}
+
+    override fun statusUpdate(driver: SchedulerDriver?, status: Protos.TaskStatus?) {
+        status?.let {
+
+            val actionName = taskIdsToActions[it.taskId]
+            when (it.state) {
+                Protos.TaskState.TASK_STARTING -> log.info("Task starting ...")
+                Protos.TaskState.TASK_RUNNING -> {
+                    jobManager.actionStarted(it.taskId.value)
+                    listener.printNotification(Notification("", "created container for $actionName created", NotificationType.Info, NotificationLevel.Execution))
+                }
+                Protos.TaskState.TASK_FINISHED -> {
+                    jobManager.actionComplete(it.taskId.value)
+                    listener.printNotification(Notification("", "Container ${it.executorId.value} Complete with task ${it.taskId.value} with success.", NotificationType.Info, NotificationLevel.Execution))
+                }
+                Protos.TaskState.TASK_FAILED,
+                Protos.TaskState.TASK_KILLED,
+                Protos.TaskState.TASK_ERROR,
+                Protos.TaskState.TASK_LOST -> {
+                    jobManager.actionFailed(it.taskId.value, it.message)
+                    listener.printNotification(Notification("", "error launching container with ${it.message} in ${it.data.toStringUtf8()}", NotificationType.Error, NotificationLevel.Execution))
+                }
+                else -> log.warn("WTF? just got unexpected task state: " + it.state)
+            }
+        }
+    }
+
+    override fun frameworkMessage(driver: SchedulerDriver?, executorId: Protos.ExecutorID?, slaveId: Protos.SlaveID?, data: ByteArray?) {
+
+        data?.let {
+            val notification: Notification = mapper.readValue(it)
+
+            when (reportLevel) {
+                NotificationLevel.Code -> listener.printNotification(notification)
+                NotificationLevel.Execution ->
+                    if (notification.notLevel != NotificationLevel.Code)
+                        listener.printNotification(notification)
+            }
+        }
+    }
+
+    override fun resourceOffers(driver: SchedulerDriver?, offers: MutableList<Protos.Offer>?) {
+        offers?.forEach {
+            when {
+                validateOffer(it) -> synchronized(jobManager) {
+                    val actionData = jobManager.nextActionData
+
+                    // if we got a new action to process
+                    actionData?.let { ad ->
+
+                        frameworkFactory = FrameworkProvidersFactory(env, config)
+                        val items = frameworkFactory.providers.values.flatMap { x -> x.configurationItems }
+                        configManager = ConfigManager(env, "repo", items)
+
+                        val taskId = Protos.TaskID.newBuilder().setValue(actionData.id).build()
+                        taskIdsToActions[taskId] = actionData.name
+
+                        createTaskConfiguration(ad)
+                        listener.printNotification(Notification("", "looking for ${actionData.groupId} provider", NotificationType.Info, NotificationLevel.Execution))
+                        val frameworkProvider = frameworkFactory.providers[actionData.groupId]
+                        listener.printNotification(Notification("", "found $frameworkProvider provider", NotificationType.Info, NotificationLevel.Execution))
+
+                        val runnerProvider = frameworkProvider!!.getRunnerProvider(actionData.typeId)
+
+                        listener.printNotification(Notification("", "provider ${runnerProvider::class.qualifiedName}", NotificationType.Info, NotificationLevel.Execution))
+                        val execData = DataLoader.getExecutorDataBytes(env, config)
+                        val executorId = taskId.value + "-" + UUID.randomUUID()
+
+                        val envConf = configManager.getActionConfiguration(ad.name, ad.config)
+                        val commandStr = runnerProvider.getCommand(jobManager.jobId, actionData, envConf, executorId, "")
+                        listener.printNotification(Notification("", "container command $commandStr", NotificationType.Info, NotificationLevel.Execution))
+
+                        val command = Protos.CommandInfo
+                                .newBuilder()
+                                .setValue(commandStr)
+
+                        // setting the container's resources
+                        val resources = setupContainerResources(frameworkProvider, runnerProvider, actionData)
+                        resources.forEach { r ->
+                            command.addUris(Protos.CommandInfo.URI.newBuilder()
+                                    .setValue(r)
+                                    .setExecutable(false)
+                                    .setExtract(true)
+                                    .build())
+                        }
+
+                        // setting the container's executable
+                        val executable: Path = getExecutable(runnerProvider, actionData)
+                        command.addUris(Protos.CommandInfo.URI.newBuilder()
+                                .setValue(toServingUri(executable.toString()))
+                                .setExecutable(false)
+                                .setExtract(false)
+                                .build())
+
+                        // setting the processes environment variables
+                        val envVarsList = frameworkProvider.environmentVariables.map { e ->
+                            Protos.Environment.Variable.newBuilder()
+                                    .setName(e.key)
+                                    .setValue(e.value)
+                                    .build()
+                        }
+                        command.setEnvironment(Protos.Environment.newBuilder().addAllVariables(envVarsList))
+
+                        val driverConfiguration = frameworkProvider.getDriverConfiguration(configManager)
+
+                        var actionTask: Protos.TaskInfo = if (runnerProvider.hasExecutor) {
+                            val executor = Protos.ExecutorInfo
+                                    .newBuilder()
+                                    .setData(ByteString.copyFrom(execData))
+                                    .setName(taskId.value)
+                                    .setExecutorId(Protos.ExecutorID.newBuilder().setValue(executorId))
+                                    .setCommand(command)
+                                    .build()
+
+                            Protos.TaskInfo.newBuilder()
+                                    .setName(taskId.value)
+                                    .setTaskId(taskId)
+                                    .setExecutor(executor)
+                                    .setData(ByteString.copyFrom(DataLoader.getTaskDataBytes(actionData, env)))
+                                    .addResources(createScalarResource("cpus", driverConfiguration.cpus.toDouble()))
+                                    .addResources(createScalarResource("mem", driverConfiguration.memory.toDouble()))
+                                    .addResources(createScalarResource("disk", config.jobs().repoSize().toDouble()))
+                                    .setSlaveId(it.slaveId)
+                                    .build()
+                            //slavesExecutors.put(offer.getSlaveId.getValue, executor)
+                        } else {
+                            Protos.TaskInfo.newBuilder()
+                                    .setName(taskId.value)
+                                    .setTaskId(taskId)
+                                    .setCommand(command)
+                                    .addResources(createScalarResource("cpus", driverConfiguration.cpus.toDouble()))
+                                    .addResources(createScalarResource("mem", driverConfiguration.memory.toDouble()))
+                                    .addResources(createScalarResource("disk", config.jobs().repoSize().toDouble()))
+                                    .setSlaveId(it.slaveId)
+                                    .build()
+                        }
+
+                        listener.printNotification(Notification("", "requesting container for ${actionData.name}", NotificationType.Info, NotificationLevel.Execution))
+                        driver?.launchTasks(Collections.singleton(it.id), listOf(actionTask))
+
+                    } ?: run {
+                        if (jobManager.outOfActions) {
+                            log.info("framework ${jobManager.jobId} execution finished")
+
+                            val repo = File("repo/")
+                            repo.delete()
+
+                            server.stop()
+                            driver?.declineOffer(it.id)
+                            driver?.stop()
+                            System.exit(0)
+                        } else {
+                            log.info("Declining offer, no sufficient resources")
+                            driver!!.declineOffer(it.id)
+                        }
+                    }
+                }
+
+            }
+        }
+    }
+
+    private fun getExecutable(runnerProvider: RunnerSetupProvider, actionData: ActionData): Path {
+        // setting up action executable
+        val sourcePath = File(runnerProvider.getActionExecutable(jobManager.jobId, actionData))
+        val executable: Path = if (actionData.hasArtifact) {
+            val relativePath = amaDist.toPath().root.relativize(sourcePath.toPath())
+            relativePath.subpath(amaDist.toPath().nameCount, relativePath.nameCount)
+        } else {
+            val dest = File("dist/${jobManager.jobId}/$sourcePath")
+            FileUtils.copyFile(sourcePath, dest)
+            Paths.get(jobManager.jobId, sourcePath.toPath().toString())
+        }
+        return executable
+    }
+
+    override fun offerRescinded(driver: SchedulerDriver?, offerId: Protos.OfferID?) {
+        offerId?.let {
+            val actionId = offersToTaskIds[it.value]
+            jobManager.requeueAction(actionId!!)
+        }
+    }
+
+    private fun validateOffer(offer: Protos.Offer): Boolean {
+        val resources = offer.resourcesList
+
+        return resources.count { it.name == "cpus" && it.scalar.value >= config.jobs().tasks().cpus() } > 0 &&
+                resources.count { it.name == "mem" && it.scalar.value >= config.jobs().tasks().mem() } > 0
+    }
+
+    private fun createTaskConfiguration(actionData: ActionData) {
+
+        // setting up the configuration files for the container
+        val envYaml = configManager.getActionConfigContent(actionData.name, actionData.config)
+        writeConfigFile(envYaml, jobManager.jobId, actionData.name, "env.yaml")
+
+        val dataStores = DataLoader.getTaskData(actionData, env).exports
+        val dataStoresYaml = yamlMapper.writeValueAsString(dataStores)
+        writeConfigFile(dataStoresYaml, jobManager.jobId, actionData.name, "datastores.yaml")
+
+        val datasets = DataLoader.getDatasets(env)
+        writeConfigFile(datasets, jobManager.jobId, actionData.name, "datasets.yaml")
+
+        writeConfigFile("jobId: ${jobManager.jobId}\nactionName: ${actionData.name}", jobManager.jobId, actionData.name, "runtime.yaml")
+
+    }
+
+    private fun writeConfigFile(configuration: String, jobId: String, actionName: String, fileName: String) {
+        val jarFile = File(this::class.java.protectionDomain.codeSource.location.path)
+        val amaHome = File(jarFile.parent).parent
+        val envLocation = "$amaHome/dist/$jobId/$actionName/"
+
+        val dir = File(envLocation)
+        if (!dir.exists()) {
+            dir.mkdirs()
+        }
+
+        File("$envLocation/$fileName").writeText(configuration)
+    }
+
+    private fun setupContainerResources(framework: FrameworkSetupProvider, runnerProvider: RunnerSetupProvider, actionData: ActionData): List<String> {
+        val result = mutableListOf<String>()
+
+        result.addAll(framework.groupResources.map { toServingUri(it.name) })
+        result.addAll(runnerProvider.runnerResources.map { toServingUri(it) })
+
+        val actionDeps = runnerProvider.getActionDependencies(jobManager.jobId, actionData).map {
+            FileUtils.copyFile(File(it), File("dist/$it"))
+            toServingUri(it)
+        }
+        result.addAll(actionDeps)
+
+        val actionResources = runnerProvider.getActionResources(jobManager.jobId, actionData).map { toServingUri(it) }
+        result.addAll(actionResources
+        )
+        return result
+    }
+
+    private fun toServingUri(file: String): String = "http://${System.getenv("AMA_NODE")}:${config.webserver().Port()}/$file"
+
+    private fun createScalarResource(name: String, value: Double): Protos.Resource {
+        return Protos.Resource.newBuilder()
+                .setName(name)
+                .setType(Protos.Value.Type.SCALAR)
+                .setScalar(Protos.Value.Scalar.newBuilder().setValue(value)).build()
+    }
+}
\ No newline at end of file
diff --git a/leader/src/main/scripts/ama-start-mesos.sh b/leader-mesos/src/main/scripts/ama-start-mesos.sh
similarity index 98%
rename from leader/src/main/scripts/ama-start-mesos.sh
rename to leader-mesos/src/main/scripts/ama-start-mesos.sh
index f12ad5c..5b9e59c 100755
--- a/leader/src/main/scripts/ama-start-mesos.sh
+++ b/leader-mesos/src/main/scripts/ama-start-mesos.sh
@@ -108,7 +108,7 @@
 done
 
 echo "repo: ${REPO} "
-CMD="java -cp ${CP} -Djava.library.path=/usr/lib org.apache.amaterasu.leader.mesos.MesosJobLauncher --home ${BASEDIR}"
+CMD="java -cp ${CP} -Djava.library.path=/usr/lib org.apache.amaterasu.leader.mesos.Client --home ${BASEDIR}"
 
 if [ -n "$REPO" ]; then
     CMD+=" --repo ${REPO}"
diff --git a/leader-yarn/src/main/kotlin/org/apache/amaterasu/leader/yarn/ApplicationMaster.kt b/leader-yarn/src/main/kotlin/org/apache/amaterasu/leader/yarn/ApplicationMaster.kt
index e89345b..b94135b 100644
--- a/leader-yarn/src/main/kotlin/org/apache/amaterasu/leader/yarn/ApplicationMaster.kt
+++ b/leader-yarn/src/main/kotlin/org/apache/amaterasu/leader/yarn/ApplicationMaster.kt
@@ -136,8 +136,8 @@
 
         val items = mutableListOf<FrameworkSetupProvider>()
 
-        for (p in frameworkFactory.providers().values()) {
-            items.add(p)
+        for (provider in frameworkFactory.providers.values) {
+            items.add(provider)
         }
         val configItems = items.flatMap { it.configurationItems.asIterable() }
         configManager = ConfigManager(env, "repo", configItems)
@@ -178,7 +178,7 @@
     private fun initJob(opts: AmaOpts) {
 
         this.env = opts.env
-        frameworkFactory = FrameworkProvidersFactory.apply(env, config)
+        frameworkFactory = FrameworkProvidersFactory(env, config)
 
         try {
             val retryPolicy = ExponentialBackoffRetry(1000, 3)
diff --git a/leader-yarn/src/main/kotlin/org/apache/amaterasu/leader/yarn/Client.kt b/leader-yarn/src/main/kotlin/org/apache/amaterasu/leader/yarn/Client.kt
index 0a0e8d7..ba5b3e6 100644
--- a/leader-yarn/src/main/kotlin/org/apache/amaterasu/leader/yarn/Client.kt
+++ b/leader-yarn/src/main/kotlin/org/apache/amaterasu/leader/yarn/Client.kt
@@ -138,8 +138,8 @@
                 }
 
                 // setup frameworks
-                val frameworkFactory = FrameworkProvidersFactory.apply(opts.env, config)
-                for (group in frameworkFactory.groups()) {
+                val frameworkFactory = FrameworkProvidersFactory(opts.env, config)
+                for (group in frameworkFactory.groups) {
                     val framework = frameworkFactory.getFramework(group)
                     for (file in framework.groupResources) {
                         if (file.exists())
diff --git a/leader-yarn/src/main/kotlin/org/apache/amaterasu/leader/yarn/ClientArgsParser.kt b/leader-yarn/src/main/kotlin/org/apache/amaterasu/leader/yarn/ClientArgsParser.kt
index b9f1e67..bc3c321 100644
--- a/leader-yarn/src/main/kotlin/org/apache/amaterasu/leader/yarn/ClientArgsParser.kt
+++ b/leader-yarn/src/main/kotlin/org/apache/amaterasu/leader/yarn/ClientArgsParser.kt
@@ -22,7 +22,7 @@
 class ClientArgsParser(val args: Array<String>): ArgsParser() {
 
     override fun run() {
-        var opts = AmaOpts(repo, branch, env, name, jobId, newJobId, report, home)
+        val opts = AmaOpts(repo, branch, env, name, jobId, newJobId, report, home)
         val client = Client()
         client.run(opts, args)
     }
diff --git a/leader/src/main/scripts/ama-start-yarn.sh b/leader-yarn/src/main/scripts/ama-start-yarn.sh
similarity index 100%
rename from leader/src/main/scripts/ama-start-yarn.sh
rename to leader-yarn/src/main/scripts/ama-start-yarn.sh
diff --git a/leader/build.gradle b/leader/build.gradle
deleted file mode 100644
index fd545c6..0000000
--- a/leader/build.gradle
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-plugins {
-    id "com.github.johnrengelman.shadow" version "1.2.4"
-    id 'com.github.maiflai.scalatest' version '0.22'
-    id 'scala'
-    id 'org.jetbrains.kotlin.jvm'
-    id 'java'
-}
-
-sourceCompatibility = 1.8
-targetCompatibility = 1.8
-
-shadowJar {
-    zip64 true
-}
-
-repositories {
-    maven {
-        url "https://plugins.gradle.org/m2/"
-    }
-    mavenCentral()
-}
-
-dependencies {
-    compile 'org.scala-lang:scala-library:2.11.8'
-    compile group: 'org.scala-lang', name: 'scala-reflect', version: '2.11.8'
-
-    compile project(':common')
-    compile project(':leader-common')
-    compile project(':amaterasu-sdk')
-    compile group: 'com.github.scopt', name: 'scopt_2.11', version: '3.3.0'
-    compile group: 'com.github.nscala-time', name: 'nscala-time_2.11', version: '2.2.0'
-    compile group: 'org.apache.curator', name: 'curator-framework', version: '2.13.0'
-    compile group: 'com.fasterxml.jackson.module', name: 'jackson-module-scala_2.11', version: '2.9.4'
-    compile group: 'com.fasterxml.jackson.core', name: 'jackson-core', version: '2.9.8'
-    compile group: 'com.fasterxml.jackson.core', name: 'jackson-annotations', version: '2.9.8'
-    compile group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: '2.9.8'
-    compile group: 'com.fasterxml.jackson.dataformat', name: 'jackson-dataformat-yaml', version: '2.9.8'
-    compile group: 'org.eclipse.jetty', name: 'jetty-plus', version: '9.4.14.v20181114'
-    compile group: 'org.eclipse.jetty', name: 'jetty-server', version: '9.4.14.v20181114'
-    compile group: 'org.eclipse.jetty', name: 'jetty-http', version: '9.4.14.v20181114'
-    compile group: 'org.eclipse.jetty', name: 'jetty-io', version: '9.4.14.v20181114'
-    compile group: 'org.eclipse.jetty', name: 'jetty-servlet', version: '9.4.14.v20181114'
-    compile group: 'org.eclipse.jetty.toolchain', name: 'jetty-test-helper', version: '4.0'
-    compile group: 'org.yaml', name: 'snakeyaml', version: '1.23'
-    compile group: 'commons-cli', name: 'commons-cli', version: '1.2'
-    compile group: 'org.jsoup', name: 'jsoup', version: '1.10.2'
-    compile group: 'org.scala-lang.modules', name: 'scala-async_2.11', version: '0.9.6'
-    compile group: 'org.jsoup', name: 'jsoup', version: '1.10.2'
-    compile group: 'net.liftweb', name: 'lift-json_2.11', version: '3.2.0'
-    compile "org.jetbrains.kotlin:kotlin-stdlib-jdk8"
-    compile "org.jetbrains.kotlin:kotlin-reflect"
-
-    testCompile project(':common')
-    testCompile "gradle.plugin.com.github.maiflai:gradle-scalatest:0.14"
-    testRuntime 'org.pegdown:pegdown:1.1.0'
-    testCompile 'junit:junit:4.11'
-    testCompile 'org.scalatest:scalatest_2.11:3.0.2'
-    testCompile 'org.scala-lang:scala-library:2.11.8'
-    testCompile 'org.apache.curator:curator-test:2.13.0'
-
-}
-
-
-sourceSets {
-    test {
-        resources.srcDirs += [file('src/test/resources')]
-    }
-
-    // this is done so Scala will compile before Java
-    main {
-        kotlin {
-            srcDirs = ['src/main/kotlin']
-        }
-        scala {
-            srcDirs = ['src/main/kotlin','src/main/java', 'src/main/scala']
-        }
-        java {
-            srcDirs = ['src/main/java']
-        }
-    }
-}
-
-test {
-    maxParallelForks = 1
-}
-
-task copyToHomeRoot(type: Copy) {
-    from 'src/main/scripts'
-    into '../build/amaterasu/'
-}
-
-task copyToHomeBin(type: Copy) {
-    dependsOn shadowJar
-    from 'build/libs'
-    into '../build/amaterasu/bin'
-}
-
-task copyToHome() {
-    dependsOn copyToHomeRoot
-    dependsOn copyToHomeBin
-}
-
-compileKotlin{
-    kotlinOptions.jvmTarget = "1.8"
-}
-
-compileTestKotlin {
-    kotlinOptions.jvmTarget = "1.8"
-}
-
-compileTestScala {
-    dependsOn compileScala
-}
-
-compileScala {
-    dependsOn compileJava
-    classpath += files(compileJava.destinationDir) + files(compileKotlin.destinationDir)
-}
-
-compileJava {
-    dependsOn compileKotlin
-}
\ No newline at end of file
diff --git a/leader/src/main/resources/log4j.properties b/leader/src/main/resources/log4j.properties
deleted file mode 100644
index 7ba668f..0000000
--- a/leader/src/main/resources/log4j.properties
+++ /dev/null
@@ -1,23 +0,0 @@
-#  Licensed to the Apache Software Foundation (ASF) under one or more
-#  contributor license agreements.  See the NOTICE file distributed with
-#  this work for additional information regarding copyright ownership.
-#  The ASF licenses this file to You under the Apache License, Version 2.0
-#  (the "License"); you may not use this file except in compliance with
-#  the License.  You may obtain a copy of the License at
-#
-#       http://www.apache.org/licenses/LICENSE-2.0
-#
-#  Unless required by applicable law or agreed to in writing, software
-#  distributed under the License is distributed on an "AS IS" BASIS,
-#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-#  See the License for the specific language governing permissions and
-#  limitations under the License.
-
-# Root logger option
-log4j.rootLogger=DEBUG, stdout, file
-
-# Redirect log messages to console
-log4j.appender.stdout=org.apache.log4j.ConsoleAppender
-log4j.appender.stdout.Target=System.out
-log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
-log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n
diff --git a/leader/src/main/scala/org/apache/amaterasu/leader/mesos/MesosJobLauncher.scala b/leader/src/main/scala/org/apache/amaterasu/leader/mesos/MesosJobLauncher.scala
deleted file mode 100755
index 98d2afa..0000000
--- a/leader/src/main/scala/org/apache/amaterasu/leader/mesos/MesosJobLauncher.scala
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.leader.mesos
-
-import org.apache.amaterasu.common.configuration.ClusterConfig
-import org.apache.amaterasu.leader.mesos.schedulers.JobScheduler
-import org.apache.amaterasu.leader.utilities.{Args, BaseJobLauncher}
-import org.apache.log4j.LogManager
-import org.apache.mesos.Protos.FrameworkID
-import org.apache.mesos.{MesosSchedulerDriver, Protos}
-
-/**
-  * The JobLauncher allows the execution of a single job, without creating a full
-  * Amaterasu cluster (no cluster scheduler).
-  */
-object MesosJobLauncher extends BaseJobLauncher {
-
-  override def run(arguments: Args, config: ClusterConfig, resume: Boolean): Unit = {
-    LogManager.resetConfiguration()
-    val frameworkBuilder = Protos.FrameworkInfo.newBuilder()
-      .setName(s"${arguments.name} - Amaterasu Job")
-      .setFailoverTimeout(config.timeout)
-      .setUser(config.user)
-
-    // TODO: test this
-    if (resume) {
-      frameworkBuilder.setId(FrameworkID.newBuilder().setValue(arguments.jobId))
-    }
-
-    val framework = frameworkBuilder.build()
-
-    val masterAddress = s"${config.master}:${config.masterPort}"
-
-    val scheduler = JobScheduler(
-      arguments.repo,
-      arguments.branch,
-      arguments.username,
-      arguments.password,
-      arguments.env,
-      resume,
-      config,
-      arguments.report,
-      arguments.home
-    )
-
-    val driver = new MesosSchedulerDriver(scheduler, framework, masterAddress)
-
-    log.debug(s"Connecting to master on: $masterAddress")
-    driver.run()
-  }
-}
diff --git a/leader/src/main/scala/org/apache/amaterasu/leader/mesos/executors/JobExecutor.scala b/leader/src/main/scala/org/apache/amaterasu/leader/mesos/executors/JobExecutor.scala
deleted file mode 100755
index 2adff07..0000000
--- a/leader/src/main/scala/org/apache/amaterasu/leader/mesos/executors/JobExecutor.scala
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.leader.mesos.executors
-
-import org.apache.amaterasu.common.logging.Logging
-import org.apache.mesos.Protos._
-import org.apache.mesos.{Executor, ExecutorDriver}
-
-object JobExecutor extends Logging with Executor {
-
-  override def shutdown(driver: ExecutorDriver): Unit = {}
-
-  override def disconnected(driver: ExecutorDriver): Unit = {}
-
-  override def killTask(driver: ExecutorDriver, taskId: TaskID): Unit = {}
-
-  override def reregistered(driver: ExecutorDriver, slaveInfo: SlaveInfo): Unit = {}
-
-  override def error(driver: ExecutorDriver, message: String): Unit = {}
-
-  override def frameworkMessage(driver: ExecutorDriver, data: Array[Byte]): Unit = {}
-
-  override def registered(driver: ExecutorDriver, executorInfo: ExecutorInfo, frameworkInfo: FrameworkInfo, slaveInfo: SlaveInfo): Unit = {}
-
-  override def launchTask(driver: ExecutorDriver, task: TaskInfo): Unit = {
-
-    //val data = mapper.readValue(task.getData.toStringUtf8, JobData.getClass)
-
-  }
-
-  def main(args: Array[String]) {
-
-    val repo = args(0)
-
-  }
-}
diff --git a/leader/src/main/scala/org/apache/amaterasu/leader/mesos/schedulers/AmaterasuScheduler.scala b/leader/src/main/scala/org/apache/amaterasu/leader/mesos/schedulers/AmaterasuScheduler.scala
deleted file mode 100755
index 13b2413..0000000
--- a/leader/src/main/scala/org/apache/amaterasu/leader/mesos/schedulers/AmaterasuScheduler.scala
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.leader.mesos.schedulers
-
-import org.apache.amaterasu.common.logging.Logging
-import org.apache.mesos.Protos.{Resource, Value}
-import org.apache.mesos.Scheduler
-
-trait AmaterasuScheduler extends Logging with Scheduler {
-
-  def createScalarResource(name: String, value: Double): Resource = {
-    Resource.newBuilder
-      .setName(name)
-      .setType(Value.Type.SCALAR)
-      .setScalar(Value.Scalar.newBuilder().setValue(value)).build()
-  }
-
-}
\ No newline at end of file
diff --git a/leader/src/main/scala/org/apache/amaterasu/leader/mesos/schedulers/JobScheduler.scala b/leader/src/main/scala/org/apache/amaterasu/leader/mesos/schedulers/JobScheduler.scala
deleted file mode 100755
index 866d9e2..0000000
--- a/leader/src/main/scala/org/apache/amaterasu/leader/mesos/schedulers/JobScheduler.scala
+++ /dev/null
@@ -1,526 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.leader.mesos.schedulers
-
-import java.io.{File, PrintWriter, StringWriter}
-import java.nio.file.{Files, Path, Paths, StandardCopyOption}
-import java.util
-import java.util.{Collections, UUID}
-import java.util.concurrent.locks.ReentrantLock
-import java.util.concurrent.{ConcurrentHashMap, LinkedBlockingQueue}
-
-import com.fasterxml.jackson.databind.ObjectMapper
-import com.fasterxml.jackson.dataformat.yaml.YAMLFactory
-import com.fasterxml.jackson.module.scala.DefaultScalaModule
-import org.apache.amaterasu.common.configuration.{ClusterConfig, ConfigManager}
-import org.apache.amaterasu.common.configuration.enums.ActionStatus
-import org.apache.amaterasu.common.dataobjects.ActionData
-import org.apache.amaterasu.common.execution.actions.Notification
-import org.apache.amaterasu.common.execution.actions.enums.{NotificationLevel, NotificationType}
-import org.apache.amaterasu.leader.common.execution.{JobLoader, JobManager}
-import org.apache.amaterasu.leader.common.execution.frameworks.FrameworkProvidersFactory
-import org.apache.amaterasu.leader.common.utilities.DataLoader
-import org.apache.amaterasu.leader.utilities.HttpServer
-import org.apache.commons.io.FileUtils
-import org.apache.curator.framework.{CuratorFramework, CuratorFrameworkFactory}
-import org.apache.curator.retry.ExponentialBackoffRetry
-import org.apache.log4j.LogManager
-import org.apache.mesos.Protos.CommandInfo.URI
-import org.apache.mesos.Protos.Environment.Variable
-import org.apache.mesos.Protos._
-import org.apache.mesos.protobuf.ByteString
-import org.apache.mesos.{Protos, SchedulerDriver}
-
-import scala.collection.JavaConverters._
-import scala.collection.concurrent
-import scala.collection.concurrent.TrieMap
-
-/**
-  * The JobScheduler is a Mesos implementation. It is in charge of scheduling the execution of
-  * Amaterasu actions for a specific job
-  */
-class JobScheduler extends AmaterasuScheduler {
-
-  /*private val props: Properties = new Properties(new File(""))
-  private val version = props.getProperty("version")
-  println(s"===> version  $version")*/
-  LogManager.resetConfiguration()
-  private var frameworkFactory: FrameworkProvidersFactory = _
-  private var configManager: ConfigManager = _
-  private var jobManager: JobManager = _
-  private var client: CuratorFramework = _
-  private var config: ClusterConfig = _
-  private var src: String = _
-  private var env: String = _
-  private var branch: String = _
-  private var userName: String = _
-  private var password: String = _
-  private var resume: Boolean = false
-  private var reportLevel: NotificationLevel = _
-
-  private val jarFile = new File(this.getClass.getProtectionDomain.getCodeSource.getLocation.getPath)
-  private val amaDist = new File(s"${new File(jarFile.getParent).getParent}/dist")
-
-  val slavesExecutors = new TrieMap[String, ExecutorInfo]
-  private var awsEnv: String = ""
-
-  // this map holds the following structure:
-  // slaveId
-  //  |
-  //  +-> taskId, actionStatus)
-  private val executionMap: concurrent.Map[String, concurrent.Map[String, ActionStatus]] = new ConcurrentHashMap[String, concurrent.Map[String, ActionStatus]].asScala
-  private val lock = new ReentrantLock()
-  private val offersToTaskIds: concurrent.Map[String, String] = new ConcurrentHashMap[String, String].asScala
-  private val taskIdsToActions: concurrent.Map[Protos.TaskID, String] = new ConcurrentHashMap[Protos.TaskID, String].asScala
-
-  private val mapper = new ObjectMapper()
-  mapper.registerModule(DefaultScalaModule)
-
-  private val yamlMapper = new ObjectMapper(new YAMLFactory())
-  yamlMapper.registerModule(DefaultScalaModule)
-
-  def error(driver: SchedulerDriver, message: String): Unit = {
-    log.error(s"===> $message")
-  }
-
-  def executorLost(driver: SchedulerDriver, executorId: ExecutorID, slaveId: SlaveID, status: Int) {}
-
-  def slaveLost(driver: SchedulerDriver, slaveId: SlaveID) {}
-
-  def disconnected(driver: SchedulerDriver) {}
-
-  def frameworkMessage(driver: SchedulerDriver, executorId: ExecutorID, slaveId: SlaveID, data: Array[Byte]): Unit = {
-
-    val notification = mapper.readValue(data, classOf[Notification])
-
-    reportLevel match {
-      case NotificationLevel.Code => printNotification(notification)
-      case NotificationLevel.Execution =>
-        if (notification.getNotLevel != NotificationLevel.Code)
-          printNotification(notification)
-      case _ =>
-    }
-
-  }
-
-  def statusUpdate(driver: SchedulerDriver, status: TaskStatus): Unit = {
-
-    val actionName = taskIdsToActions(status.getTaskId)
-    status.getState match {
-      case TaskState.TASK_STARTING => log.info("Task starting ...")
-      case TaskState.TASK_RUNNING => {
-        jobManager.actionStarted(status.getTaskId.getValue)
-        printNotification(new Notification("", s"created container for $actionName created", NotificationType.Info, NotificationLevel.Execution))
-
-      }
-      case TaskState.TASK_FINISHED => {
-        jobManager.actionComplete(status.getTaskId.getValue)
-        printNotification(new Notification("", s"Container ${status.getExecutorId.getValue} Complete with task ${status.getTaskId.getValue} with success.", NotificationType.Info, NotificationLevel.Execution))
-      }
-      case TaskState.TASK_FAILED |
-           TaskState.TASK_KILLED |
-           TaskState.TASK_ERROR |
-           TaskState.TASK_LOST => {
-        jobManager.actionFailed(status.getTaskId.getValue, status.getMessage)
-        printNotification(new Notification("", s"error launching container with ${status.getMessage} in ${status.getData.toStringUtf8}", NotificationType.Error, NotificationLevel.Execution))
-
-      }
-      case _ => log.warn("WTF? just got unexpected task state: " + status.getState)
-    }
-
-  }
-
-  def validateOffer(offer: Offer): Boolean = {
-
-    val resources = offer.getResourcesList.asScala
-
-    resources.count(r => r.getName == "cpus" && r.getScalar.getValue >= config.jobs.tasks.cpus) > 0 &&
-      resources.count(r => r.getName == "mem" && r.getScalar.getValue >= config.jobs.tasks.mem) > 0
-  }
-
-  def offerRescinded(driver: SchedulerDriver, offerId: OfferID): Unit = {
-
-    val actionId = offersToTaskIds(offerId.getValue)
-    jobManager.reQueueAction(actionId)
-
-  }
-
-  def resourceOffers(driver: SchedulerDriver, offers: util.List[Offer]): Unit = {
-
-    println(jobManager.toString)
-
-    for (offer <- offers.asScala) {
-
-      if (validateOffer(offer)) {
-
-        log.info(s"Accepting offer, id=${offer.getId}")
-
-        // this is done to avoid the processing the same action
-        // multiple times
-        lock.lock()
-
-        try {
-          val actionData = jobManager.getNextActionData
-          if (actionData != null) {
-
-            frameworkFactory = FrameworkProvidersFactory(env, config)
-            val items = frameworkFactory.providers.values.flatMap(_.getConfigurationItems).toList.asJava
-            configManager = new ConfigManager(env, "repo", items)
-
-            val taskId = Protos.TaskID.newBuilder().setValue(actionData.getId).build()
-            taskIdsToActions.put(taskId, actionData.getName)
-            // setting up the configuration files for the container
-            val envYaml = configManager.getActionConfigContent(actionData.getName, actionData.getConfig)
-            writeConfigFile(envYaml, jobManager.getJobId, actionData.getName, "env.yaml")
-
-            val envConf = configManager.getActionConfiguration(actionData.getName, actionData.getConfig)
-            val dataStores = DataLoader.getTaskData(actionData, env).getExports
-            val writer = new StringWriter()
-            yamlMapper.writeValue(writer, dataStores)
-            val dataStoresYaml = writer.toString
-            writeConfigFile(dataStoresYaml, jobManager.getJobId, actionData.getName, "datastores.yaml")
-
-            writeConfigFile(s"jobId: ${jobManager.getJobId}\nactionName: ${actionData.getName}", jobManager.getJobId, actionData.getName, "runtime.yaml")
-
-            val datasets = DataLoader.getDatasets(env)
-            writeConfigFile(datasets, jobManager.getJobId, actionData.getName, "datasets.yaml")
-            offersToTaskIds.put(offer.getId.getValue, taskId.getValue)
-
-            // atomically adding a record for the slave, I'm storing all the actions
-            // on a slave level to efficiently handle slave loses
-            executionMap.putIfAbsent(offer.getSlaveId.toString, new ConcurrentHashMap[String, ActionStatus].asScala)
-
-            val slaveActions = executionMap(offer.getSlaveId.toString)
-            slaveActions.put(taskId.getValue, ActionStatus.Started)
-
-            val frameworkProvider = frameworkFactory.providers(actionData.getGroupId)
-
-            val runnerProvider = frameworkProvider.getRunnerProvider(actionData.getTypeId)
-
-            printNotification(new Notification("", s"provider ${runnerProvider.getClass.getName}", NotificationType.Info, NotificationLevel.Execution))
-            // searching for an executor that already exist on the slave, if non exist
-            // we create a new one
-            var executor: ExecutorInfo = null
-
-            //            val slaveId = offer.getSlaveId.getValue
-            //            slavesExecutors.synchronized {
-
-            val execData = DataLoader.getExecutorDataBytes(env, config)
-            val executorId = taskId.getValue + "-" + UUID.randomUUID()
-            //creating the command
-
-            //                          // TODO: move this into the runner provider somehow
-            //                          if(!actionData.getSrc.isEmpty){
-            //                            copy(get(s"repo/src/${actionData.getSrc}"), get(s"dist/${jobManager.getJobId}/${actionData.getName}/${actionData.getSrc}"), REPLACE_EXISTING)
-            //                          }
-            val commandStr = runnerProvider.getCommand(jobManager.getJobId, actionData, envConf, executorId, "")
-            printNotification(new Notification("", s"container command $commandStr", NotificationType.Info, NotificationLevel.Execution))
-
-            val command = CommandInfo
-              .newBuilder
-              .setValue(commandStr)
-            //              .addUris(URI.newBuilder
-            //                .setValue(s"http://${sys.env("AMA_NODE")}:${config.webserver.Port}/executor-${config.version}-all.jar")
-            //                .setExecutable(false)
-            //                .setExtract(false)
-            //                .build())
-
-            // Getting framework (group) resources
-            log.info(s"===> groupResources: ${frameworkProvider.getGroupResources}")
-            frameworkProvider.getGroupResources.foreach(f => command.addUris(URI.newBuilder
-              .setValue(s"http://${sys.env("AMA_NODE")}:${config.webserver.Port}/${f.getName}")
-              .setExecutable(false)
-              .setExtract(true)
-              .build()
-            ))
-
-            // Getting runner resources
-            runnerProvider.getRunnerResources.foreach(r => {
-              command.addUris(URI.newBuilder
-                .setValue(s"http://${sys.env("AMA_NODE")}:${config.webserver.Port}/$r")
-                .setExecutable(false)
-                .setExtract(false)
-                .build())
-            })
-
-            // Getting action dependencies
-            runnerProvider.getActionDependencies(jobManager.getJobId, actionData).foreach(r => {
-
-              FileUtils.copyFile(new File(r), new File(s"dist/$r"))
-              command.addUris(URI.newBuilder
-                .setValue(s"http://${sys.env("AMA_NODE")}:${config.webserver.Port}/$r")
-                .setExecutable(false)
-                .setExtract(false)
-                .build())
-            })
-
-            // Getting action specific resources
-            runnerProvider.getActionResources(jobManager.getJobId, actionData).foreach(r => command.addUris(URI.newBuilder
-              .setValue(s"http://${sys.env("AMA_NODE")}:${config.webserver.Port}/$r")
-              .setExecutable(false)
-              .setExtract(false)
-              .build()))
-
-            // setting up action executable
-            val sourcePath = new File(runnerProvider.getActionExecutable(jobManager.getJobId, actionData))
-            var executable: Path = null
-            if (actionData.getHasArtifact) {
-              val relativePath = amaDist.toPath.getRoot.relativize(sourcePath.toPath)
-              executable = relativePath.subpath(amaDist.toPath.getNameCount, relativePath.getNameCount)
-            } else {
-              val dest = new File(s"dist/${jobManager.getJobId}/${sourcePath.toString}")
-              FileUtils.copyFile(sourcePath, dest)
-              executable = Paths.get(jobManager.getJobId, sourcePath.toPath.toString)
-            }
-
-            println(s"===> executable $executable")
-            command.addUris(URI.newBuilder
-              .setValue(s"http://${sys.env("AMA_NODE")}:${config.webserver.Port}/$executable")
-              .setExecutable(false)
-              .setExtract(false)
-              .build())
-
-            command
-              .addUris(URI.newBuilder()
-                .setValue(s"http://${sys.env("AMA_NODE")}:${config.webserver.Port}/amaterasu.properties")
-                .setExecutable(false)
-                .setExtract(false)
-                .build())
-
-            // setting the processes environment variables
-            val envVarsList = frameworkProvider.getEnvironmentVariables.asScala.toList.map(x => Variable.newBuilder().setName(x._1).setValue(x._2).build()).asJava
-            command.setEnvironment(Environment.newBuilder().addAllVariables(envVarsList))
-
-            executor = ExecutorInfo
-              .newBuilder
-              .setData(ByteString.copyFrom(execData))
-              .setName(taskId.getValue)
-              .setExecutorId(ExecutorID.newBuilder().setValue(executorId))
-              .setCommand(command)
-
-              .build()
-
-            slavesExecutors.put(offer.getSlaveId.getValue, executor)
-
-
-            val driverConfiguration = frameworkProvider.getDriverConfiguration(configManager)
-
-            var actionTask: TaskInfo = null
-
-            if (runnerProvider.getHasExecutor) {
-              actionTask = TaskInfo
-                .newBuilder
-                .setName(taskId.getValue)
-                .setTaskId(taskId)
-                .setExecutor(executor)
-
-                .setData(ByteString.copyFrom(DataLoader.getTaskDataBytes(actionData, env)))
-                .addResources(createScalarResource("cpus", driverConfiguration.getCpus))
-                .addResources(createScalarResource("mem", driverConfiguration.getMemory))
-                .addResources(createScalarResource("disk", config.jobs.repoSize))
-                .setSlaveId(offer.getSlaveId)
-                .build()
-
-              //driver.launchTasks(Collections.singleton(offer.getId), List(actionTask).asJava)
-            }
-            else {
-              actionTask = TaskInfo
-                .newBuilder
-                .setName(taskId.getValue)
-                .setTaskId(taskId)
-                .setCommand(command)
-
-                //.setData(ByteString.copyFrom(DataLoader.getTaskDataBytes(actionData, env)))
-                .addResources(createScalarResource("cpus", driverConfiguration.getCpus))
-                .addResources(createScalarResource("mem", driverConfiguration.getMemory))
-                .addResources(createScalarResource("disk", config.jobs.repoSize))
-                .setSlaveId(offer.getSlaveId)
-                .build()
-
-              //driver.launchTasks(Collections.singleton(offer.getId), List(actionTask).asJava)
-            }
-
-            printNotification(new Notification("", s"requesting container for ${actionData.getName}", NotificationType.Info, NotificationLevel.Execution))
-            driver.launchTasks(Collections.singleton(offer.getId), List(actionTask).asJava)
-
-          }
-          else if (jobManager.getOutOfActions) {
-            log.info(s"framework ${jobManager.getJobId} execution finished")
-
-            val repo = new File("repo/")
-            repo.delete()
-
-            HttpServer.stop()
-            driver.declineOffer(offer.getId)
-            driver.stop()
-            sys.exit()
-          }
-          else {
-            log.info("Declining offer, no action ready for execution")
-            driver.declineOffer(offer.getId)
-          }
-        }
-        finally {
-          lock.unlock()
-        }
-      }
-      else {
-        log.info("Declining offer, no sufficient resources")
-        driver.declineOffer(offer.getId)
-      }
-
-    }
-
-  }
-
-  def registered(driver: SchedulerDriver, frameworkId: FrameworkID, masterInfo: MasterInfo): Unit = {
-
-    if (!resume) {
-
-      jobManager = JobLoader.loadJob(
-        src,
-        branch,
-        frameworkId.getValue,
-        userName,
-        password,
-        client,
-        config.jobs.tasks.attempts,
-        new LinkedBlockingQueue[ActionData]()
-      )
-    }
-    else {
-
-      JobLoader.reloadJob(
-        frameworkId.getValue,
-        userName,
-        password,
-        client,
-        config.jobs.tasks.attempts,
-        new LinkedBlockingQueue[ActionData]()
-      )
-
-    }
-
-
-
-
-    jobManager.start()
-
-    createJobDir(jobManager.getJobId)
-
-  }
-
-  def reregistered(driver: SchedulerDriver, masterInfo: Protos.MasterInfo) {}
-
-  def printNotification(notification: Notification): Unit = {
-
-    var color = Console.WHITE
-
-    notification.getNotType match {
-
-      case NotificationType.Info =>
-        color = Console.WHITE
-        println(s"$color${Console.BOLD}===> ${notification.getMsg} ${Console.RESET}")
-      case NotificationType.Success =>
-        color = Console.GREEN
-        println(s"$color${Console.BOLD}===> ${notification.getLine} ${Console.RESET}")
-      case NotificationType.Error =>
-        color = Console.RED
-        println(s"$color${Console.BOLD}===> ${notification.getLine} ${Console.RESET}")
-        println(s"$color${Console.BOLD}===> ${notification.getMsg} ${Console.RESET}")
-
-    }
-
-  }
-
-  private def createJobDir(jobId: String): Unit = {
-    val jarFile = new File(this.getClass.getProtectionDomain.getCodeSource.getLocation.getPath)
-    val amaHome = new File(jarFile.getParent).getParent
-    val jobDir = s"$amaHome/dist/$jobId/"
-
-    val dir = new File(jobDir)
-    if (!dir.exists()) {
-      dir.mkdir()
-    }
-  }
-
-  /**
-    * This function creates an action specific env.yml file int the dist folder with the following path:
-    * dist/{jobId}/{actionName}/env.yml to be added to the container
-    *
-    * @param configuration A YAML string to be written to the env file
-    * @param jobId         the jobId
-    * @param actionName    the name of the action
-    */
-  def writeConfigFile(configuration: String, jobId: String, actionName: String, fileName: String): Unit = {
-    val jarFile = new File(this.getClass.getProtectionDomain.getCodeSource.getLocation.getPath)
-    val amaHome = new File(jarFile.getParent).getParent
-    val envLocation = s"$amaHome/dist/$jobId/$actionName/"
-
-    val dir = new File(envLocation)
-    if (!dir.exists()) {
-      dir.mkdirs()
-    }
-
-    new PrintWriter(s"$envLocation/$fileName") {
-      write(configuration)
-      close
-    }
-  }
-}
-
-object JobScheduler {
-
-  def apply(src: String,
-            branch: String,
-            username: String,
-            password: String,
-            env: String,
-            resume: Boolean,
-            config: ClusterConfig,
-            report: String,
-            home: String): JobScheduler = {
-
-    LogManager.resetConfiguration()
-    val scheduler = new JobScheduler()
-
-    HttpServer.start(config.webserver.Port, s"$home/${config.webserver.Root}")
-
-    if (sys.env.get("AWS_ACCESS_KEY_ID").isDefined &&
-      sys.env.get("AWS_SECRET_ACCESS_KEY").isDefined) {
-
-      scheduler.awsEnv = s"env AWS_ACCESS_KEY_ID=${sys.env("AWS_ACCESS_KEY_ID")} env AWS_SECRET_ACCESS_KEY=${sys.env("AWS_SECRET_ACCESS_KEY")}"
-    }
-
-    scheduler.resume = resume
-    scheduler.src = src
-    scheduler.branch = branch
-    scheduler.userName = username
-    scheduler.password = password
-    scheduler.env = env
-    scheduler.reportLevel = NotificationLevel.valueOf(report.capitalize)
-
-    val retryPolicy = new ExponentialBackoffRetry(1000, 3)
-    scheduler.client = CuratorFrameworkFactory.newClient(config.zk, retryPolicy)
-    scheduler.client.start()
-    scheduler.config = config
-
-    scheduler
-
-  }
-
-}
\ No newline at end of file
diff --git a/leader/src/main/scala/org/apache/amaterasu/leader/utilities/Args.scala b/leader/src/main/scala/org/apache/amaterasu/leader/utilities/Args.scala
deleted file mode 100644
index 8364456..0000000
--- a/leader/src/main/scala/org/apache/amaterasu/leader/utilities/Args.scala
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.leader.utilities
-
-case class Args(
-                 repo: String = "",
-                 branch: String = "master",
-                 env: String = "default",
-                 name: String = "amaterasu-job",
-                 jobId: String = null,
-                 report: String = "code",
-                 home: String = "",
-                 newJobId: String = "",
-                 username: String = "",
-                 password: String = ""
-               ) {
-  def toCmdString: String = {
-    var cmd = s""" --repo $repo --branch $branch --env $env --name $name --report $report --home $home"""
-    if (!username.isEmpty && !password.isEmpty) {
-      cmd += s" --user-name $username --password $password"
-    }
-    if(jobId != null && !jobId.isEmpty) {
-      cmd += s" --job-id $jobId"
-    }
-    cmd
-  }
-
-  override def toString: String = {
-    toCmdString
-  }
-}
-
-object Args {
-  def getParser: scopt.OptionParser[Args] = {
-    val pack = this.getClass.getPackage
-    new scopt.OptionParser[Args]("amaterasu job") {
-
-      head("amaterasu job", if(pack == null) "DEVELOPMENT" else pack.getImplementationVersion)
-
-      opt[String]('r', "repo") action { (x, c) =>
-        c.copy(repo = x)
-      } text "The git repo containing the job"
-
-      opt[String]('b', "branch") action { (x, c) =>
-        c.copy(branch = x)
-      } text "The branch to be executed (default is master)"
-
-      opt[String]('e', "env") action { (x, c) =>
-        c.copy(env = x)
-      } text "The environment to be executed (test, prod, etc. values from the default env are taken if np env specified)"
-
-      opt[String]('n', "name") action { (x, c) =>
-        c.copy(name = x)
-      } text "The name of the job"
-
-       opt[String]('i', "job-id") action { (x, c) =>
-        c.copy(jobId = x)
-      } text "The jobId - should be passed only when resuming a job"
-
-      opt[String]('j', "new-job-id") action { (x, c) =>
-        c.copy(newJobId = x)
-      } text "A new jobId - should never be passed by a user"
-
-      opt[String]('r', "report") action { (x, c) =>
-        c.copy(report = x)
-      }  text "The level of reporting"
-
-      opt[String]('h', "home") action { (x, c) =>
-        c.copy(home = x)
-      }
-
-      opt[String]('u', "user-name") action { (x, c) =>
-        c.copy(username = x)
-      }
-
-      opt[String]('p', "password") action { (x, c) =>
-        c.copy(password = x)
-      }
-    }
-  }
-}
diff --git a/leader/src/main/scala/org/apache/amaterasu/leader/utilities/BaseJobLauncher.scala b/leader/src/main/scala/org/apache/amaterasu/leader/utilities/BaseJobLauncher.scala
deleted file mode 100644
index 38c90c7..0000000
--- a/leader/src/main/scala/org/apache/amaterasu/leader/utilities/BaseJobLauncher.scala
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.leader.utilities
-
-import java.io.FileInputStream
-
-import org.apache.amaterasu.common.configuration.ClusterConfig
-import org.apache.amaterasu.common.logging.Logging
-
-
-abstract class BaseJobLauncher extends Logging with App {
-
-  def run(args: Args, config: ClusterConfig, resume: Boolean): Unit = ???
-
-  val parser = Args.getParser
-  parser.parse(args, Args()) match {
-
-    case Some(arguments: Args) =>
-
-      val config = ClusterConfig(new FileInputStream(s"${arguments.home}/amaterasu.properties"))
-      val resume = arguments.jobId != null
-
-      run(arguments, config, resume)
-
-    case None =>
-    // arguments are bad, error message will have been displayed
-  }
-}
diff --git a/leader/src/main/scala/org/apache/amaterasu/leader/utilities/HttpServer.scala b/leader/src/main/scala/org/apache/amaterasu/leader/utilities/HttpServer.scala
deleted file mode 100644
index 4aacd38..0000000
--- a/leader/src/main/scala/org/apache/amaterasu/leader/utilities/HttpServer.scala
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.leader.utilities
-
-import org.apache.amaterasu.common.logging.Logging
-import org.apache.log4j.{BasicConfigurator, Level, Logger}
-import org.eclipse.jetty.server.handler._
-import org.eclipse.jetty.server.{Server, ServerConnector}
-import org.eclipse.jetty.util.log.StdErrLog
-import org.jsoup.Jsoup
-import org.jsoup.select.Elements
-
-import scala.collection.JavaConverters._
-import scala.io.{BufferedSource, Source}
-
-/**
-  * Implementation of Jetty Web server to server Amaterasu libraries and other distribution files
-  */
-object HttpServer extends Logging {
-
-  var server: Server = _
-
-  def start(port: String, serverRoot: String): Unit = {
-
-    BasicConfigurator.configure()
-    initLogging()
-
-    server = new Server()
-    val connector = new ServerConnector(server)
-    connector.setPort(port.toInt)
-    server.addConnector(connector)
-
-    val handler = new ResourceHandler()
-    handler.setDirectoriesListed(true)
-    handler.setWelcomeFiles(Array[String]("index.html"))
-    handler.setResourceBase(serverRoot)
-    val handlers = new HandlerList()
-    handlers.setHandlers(Array(handler, new DefaultHandler()))
-
-    server.setHandler(handlers)
-    server.start()
-
-  }
-
-  def stop() {
-    if (server == null) throw new IllegalStateException("Server not Started")
-
-    server.stop()
-    server = null
-  }
-
-  def initLogging(): Unit = {
-    System.setProperty("org.eclipse.jetty.util.log.class", classOf[StdErrLog].getName)
-    Logger.getLogger("org.eclipse.jetty").setLevel(Level.ALL)
-    Logger.getLogger("org.eclipse.jetty.websocket").setLevel(Level.OFF)
-  }
-
-  /*
-  Method: getFilesInDirectory
-  Description: provides a list of files in the given directory URL.
-  @Params: amaNode: Hostname of the URL, port: Port # of the host, directory: destination directory to fetch files
-  Note: Should the files in URL root be fetched, provide an empty value to directory.
-   */
-  def getFilesInDirectory(amaNode: String, port: String, directory: String = ""): Array[String] = {
-    println("http://" + amaNode + ":" + port + "/" + directory)
-    val html: BufferedSource = Source.fromURL("http://" + amaNode + ":" + port + "/" + directory)
-    println(html)
-    val htmlDoc = Jsoup.parse(html.mkString)
-    val htmlElement: Elements = htmlDoc.body().select("a")
-    val files = htmlElement.asScala
-    val fileNames = files.map(url => url.attr("href")).filter(file => !file.contains("..")).map(name => name.replace("/", "")).toArray
-    fileNames
-  }
-}
\ No newline at end of file
diff --git a/leader/src/test/resources/amaterasu.properties b/leader/src/test/resources/amaterasu.properties
deleted file mode 100755
index 96e3724..0000000
--- a/leader/src/test/resources/amaterasu.properties
+++ /dev/null
@@ -1,32 +0,0 @@
-#  Licensed to the Apache Software Foundation (ASF) under one or more
-#  contributor license agreements.  See the NOTICE file distributed with
-#  this work for additional information regarding copyright ownership.
-#  The ASF licenses this file to You under the Apache License, Version 2.0
-#  (the "License"); you may not use this file except in compliance with
-#  the License.  You may obtain a copy of the License at
-#
-#       http://www.apache.org/licenses/LICENSE-2.0
-#
-#  Unless required by applicable law or agreed to in writing, software
-#  distributed under the License is distributed on an "AS IS" BASIS,
-#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-#  See the License for the specific language governing permissions and
-#  limitations under the License.
-zk=127.0.0.1
-version=0.2.0-incubating-rc4
-master=10.0.0.31
-user=root
-mode=yarn
-webserver.port=8000
-webserver.root=dist
-spark.version=2.2.1-bin-hadoop2.7
-yarn.queue=default
-yarn.jarspath=hdfs:///apps/amaterasu
-spark.home=/usr/lib/spark
-#spark.home=/opt/cloudera/parcels/SPARK2-2.1.0.cloudera2-1.cdh5.7.0.p0.171658/lib/spark2
-yarn.hadoop.home.dir=/etc/hadoop
-spark.opts.spark.yarn.am.extraJavaOptions="-Dhdp.version=2.6.1.0-129"
-spark.opts.spark.driver.extraJavaOptions="-Dhdp.version=2.6.1.0-129"
-yarn.master.memoryMB=2048
-yarn.worker.memoryMB=2048
-mesos.libPath=/usr/local/lib/libmesos.dylib
\ No newline at end of file
diff --git a/leader/src/test/scala/org/apache/amaterasu/common/execution/ActionStatusTests.scala b/leader/src/test/scala/org/apache/amaterasu/common/execution/ActionStatusTests.scala
deleted file mode 100755
index ca9186c..0000000
--- a/leader/src/test/scala/org/apache/amaterasu/common/execution/ActionStatusTests.scala
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.common.execution
-
-import java.util
-import java.util.concurrent.LinkedBlockingQueue
-
-import org.apache.amaterasu.common.configuration.enums.ActionStatus
-import org.apache.amaterasu.common.dataobjects.ActionData
-import org.apache.amaterasu.leader.common.execution.actions.SequentialAction
-import org.apache.curator.framework.CuratorFrameworkFactory
-import org.apache.curator.retry.ExponentialBackoffRetry
-import org.apache.curator.test.TestingServer
-import org.apache.zookeeper.CreateMode
-import org.scalatest.{DoNotDiscover, FlatSpec, Matchers}
-
-import scala.collection.JavaConverters._
-
-class ActionStatusTests extends FlatSpec with Matchers {
-
-  // setting up a testing zookeeper server (curator TestServer)
-  val retryPolicy = new ExponentialBackoffRetry(1000, 3)
-  val server = new TestingServer(2181, true)
-  val jobId = s"job_${System.currentTimeMillis}"
-  val data = new ActionData(ActionStatus.Pending, "test_action", "start.scala", "", "spark","scala", "0000001", new util.HashMap() , List[String]().asJava)
-
-  "an Action" should "queue it's ActionData int the job queue when executed" in {
-
-    val queue = new LinkedBlockingQueue[ActionData]()
-    // val config = ClusterConfig()
-
-    val client = CuratorFrameworkFactory.newClient(server.getConnectString, retryPolicy)
-    client.start()
-
-    client.create().withMode(CreateMode.PERSISTENT).forPath(s"/$jobId")
-    val action = new SequentialAction(data.getName, data.getSrc, "", data.getGroupId, data.getTypeId, Map.empty[String, String].asJava, jobId, queue, client, 1)
-
-    action.execute()
-    queue.peek().getName should be(data.getName)
-    queue.peek().getSrc should be(data.getSrc)
-
-  }
-
-  it should "also create a sequential znode for the task with the value of Queued" in {
-
-    val client = CuratorFrameworkFactory.newClient(server.getConnectString, retryPolicy)
-    client.start()
-
-    val taskStatus = client.getData.forPath(s"/$jobId/task-0000000000")
-
-    taskStatus should not be null
-    new String(taskStatus) should be("Queued")
-
-  }
-
-}
\ No newline at end of file
diff --git a/leader/src/test/scala/org/apache/amaterasu/common/execution/JobExecutionTests.scala b/leader/src/test/scala/org/apache/amaterasu/common/execution/JobExecutionTests.scala
deleted file mode 100755
index 28b1ce3..0000000
--- a/leader/src/test/scala/org/apache/amaterasu/common/execution/JobExecutionTests.scala
+++ /dev/null
@@ -1,158 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.common.execution
-
-import java.util.concurrent.LinkedBlockingQueue
-
-import org.apache.amaterasu.common.dataobjects.ActionData
-import org.apache.amaterasu.leader.common.dsl.JobParser
-import org.apache.amaterasu.leader.common.execution.actions.Action
-import org.apache.curator.framework.CuratorFrameworkFactory
-import org.apache.curator.retry.ExponentialBackoffRetry
-import org.apache.curator.test.TestingServer
-import org.apache.zookeeper.CreateMode
-import org.scalatest.{FlatSpec, Matchers}
-
-import scala.io.Source
-
-class JobExecutionTests extends FlatSpec with Matchers {
-
-  val retryPolicy = new ExponentialBackoffRetry(1000, 3)
-  val server = new TestingServer(2183, true)
-  val client = CuratorFrameworkFactory.newClient(server.getConnectString, retryPolicy)
-  client.start()
-
-  val jobId = s"job_${System.currentTimeMillis}"
-  val yaml = Source.fromURL(getClass.getResource("/simple-maki.yml")).mkString
-  val queue = new LinkedBlockingQueue[ActionData]()
-
-  // this will be performed by the job bootstraper
-
-  client.create().withMode(CreateMode.PERSISTENT).forPath(s"/$jobId")
-  //  client.setData().forPath(s"/$jobId/src",src.getBytes)
-  //  client.setData().forPath(s"/$jobId/branch", branch.getBytes)
-
-
-  val job = JobParser.parse(jobId, yaml, queue, client, 1)
-
-  "a job" should "queue the first action when the JobManager.start method is called " in {
-
-    job.start
-
-    queue.peek.getName should be("start")
-
-    // making sure that the status is reflected in zk
-    val actionStatus = client.getData.forPath(s"/$jobId/task-0000000000")
-    new String(actionStatus) should be("Queued")
-
-  }
-
-  it should "return the start action when calling getNextAction and dequeue it" in {
-
-    job.getNextActionData.getName should be("start")
-    queue.size should be(0)
-
-    // making sure that the status is reflected in zk
-    val actionStatus = client.getData.forPath(s"/$jobId/task-0000000000")
-    new String(actionStatus) should be("Started")
-  }
-
-  it should "not be out of actions when an action is still Pending" in {
-    job.getOutOfActions should be(false)
-  }
-
-  it should "be requeued correctly" in {
-    job.reQueueAction("0000000000")
-    val actionStatus = client.getData.forPath(s"/$jobId/task-0000000000")
-    new String(new String(actionStatus)) should be("Queued")
-  }
-
-  it should "should be able to restart" in {
-
-    val data = job.getNextActionData
-
-    data.getName should be("start")
-
-    // making sure that the status is reflected in zk
-    val actionStatus = client.getData.forPath(s"/$jobId/task-0000000000")
-    new String(actionStatus) should be("Started")
-  }
-
-  it should "be marked as Complete when the actionComplete method is called" in {
-
-    job.actionComplete("0000000000")
-
-    // making sure that the status is reflected in zk
-    val actionStatus = client.getData.forPath(s"/$jobId/task-0000000000")
-
-    new String(new String(actionStatus)) should be("Complete")
-
-  }
-
-  "the next step2 job" should "be Queued as a result of the completion" in {
-
-    queue.peek.getName should be("step2")
-
-    // making sure that the status is reflected in zk
-    val actionStatus = client.getData.forPath(s"/$jobId/task-0000000001")
-    new String(actionStatus) should be("Queued")
-
-  }
-
-  it should "be marked as Started when JobManager.getNextActionData is called" in {
-
-    val data = job.getNextActionData
-
-    data.getName should be("step2")
-
-    // making sure that the status is reflected in zk
-    val actionStatus = client.getData.forPath(s"/$jobId/task-0000000001")
-    new String(actionStatus) should be("Started")
-  }
-
-  it should "be marked as Failed when JobManager.actionFailed is called" in {
-
-    job.actionFailed("0000000001", "test failure")
-    queue.peek.getName should be("error-action")
-  }
-
-  "an ErrorAction" should "be queued if one exist" in {
-    // making sure that the status is reflected in zk
-    val actionStatus = client.getData.forPath(s"/$jobId/task-0000000001-error")
-    new String(actionStatus) should be("Queued")
-
-    // and returned by getNextActionData
-    val data = job.getNextActionData
-
-  }
-
-  it should "be marked as Complete when the actionComplete method is called" in {
-
-    job.actionComplete("0000000001-error")
-
-    // making sure that the status is reflected in zk
-    val actionStatus = client.getData.forPath(s"/$jobId/task-0000000001-error")
-
-    new String(new String(actionStatus)) should be("Complete")
-
-  }
-
-  it should " be out of actions when all actions have been executed" in {
-
-    job.getOutOfActions should be(true)
-  }
-}
diff --git a/leader/src/test/scala/org/apache/amaterasu/common/execution/JobParserTests.scala b/leader/src/test/scala/org/apache/amaterasu/common/execution/JobParserTests.scala
deleted file mode 100755
index 5987b35..0000000
--- a/leader/src/test/scala/org/apache/amaterasu/common/execution/JobParserTests.scala
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.common.execution
-
-import java.util.concurrent.LinkedBlockingQueue
-
-import org.apache.amaterasu.common.dataobjects.ActionData
-import org.apache.amaterasu.leader.common.dsl.JobParser
-import org.apache.curator.framework.CuratorFrameworkFactory
-import org.apache.curator.retry.ExponentialBackoffRetry
-import org.apache.curator.test.TestingServer
-import org.apache.zookeeper.CreateMode
-import org.scalatest.{FlatSpec, Matchers}
-
-import scala.io.Source
-
-class JobParserTests extends FlatSpec with Matchers {
-
-  val retryPolicy = new ExponentialBackoffRetry(1000, 3)
-  val server = new TestingServer(2187, true)
-  val client = CuratorFrameworkFactory.newClient(server.getConnectString, retryPolicy)
-  client.start()
-
-  private val jobId = s"job_${System.currentTimeMillis}"
-  private val yaml = Source.fromURL(getClass.getResource("/simple-maki.yml")).mkString
-  private val queue = new LinkedBlockingQueue[ActionData]()
-
-  // this will be performed by the job bootstrapper
-  client.create().withMode(CreateMode.PERSISTENT).forPath(s"/$jobId")
-
-  private val job = JobParser.parse(jobId, yaml, queue, client, 1)
-
-  "JobParser" should "parse the simple-maki.yml" in {
-
-    job.getName should be("amaterasu-test")
-
-  }
-
-  //TODO: I suspect this test is not indicative, and that order is assured need to verify this
-  it should "also have two actions in the right order" in {
-
-    job.getRegisteredActions.size should be(3)
-
-    job.getRegisteredActions.get("0000000000").data.getName should be("start")
-    job.getRegisteredActions.get("0000000001").data.getName should be("step2")
-    job.getRegisteredActions.get("0000000001-error").data.getName should be("error-action")
-
-  }
-
-  it should "Action 'config' is parsed successfully" in {
-
-    job.getRegisteredActions.size should be(3)
-
-    job.getRegisteredActions.get("0000000000").data.getConfig should be("start-cfg.yaml")
-    job.getRegisteredActions.get("0000000001").data.getConfig should be("step2-cfg.yaml")
-    job.getRegisteredActions.get("0000000001-error").data.getConfig should be("error-cfg.yaml")
-
-  }
-
-}
diff --git a/leader/src/test/scala/org/apache/amaterasu/common/execution/JobRestoreTests.scala b/leader/src/test/scala/org/apache/amaterasu/common/execution/JobRestoreTests.scala
deleted file mode 100755
index 235eccb..0000000
--- a/leader/src/test/scala/org/apache/amaterasu/common/execution/JobRestoreTests.scala
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.common.execution
-
-import java.util.concurrent.LinkedBlockingQueue
-
-import org.apache.amaterasu.common.configuration.enums.ActionStatus
-import org.apache.amaterasu.common.dataobjects.ActionData
-import org.apache.amaterasu.leader.common.execution.{JobLoader, JobManager}
-import org.apache.curator.framework.{CuratorFramework, CuratorFrameworkFactory}
-import org.apache.curator.retry.ExponentialBackoffRetry
-import org.apache.curator.test.TestingServer
-import org.apache.zookeeper.CreateMode
-import org.scalatest.{BeforeAndAfterEach, FlatSpec, Matchers}
-
-import scala.io.Source
-
-class JobRestoreTests extends FlatSpec with Matchers with BeforeAndAfterEach {
-
-  val retryPolicy = new ExponentialBackoffRetry(1000, 3)
-  val server = new TestingServer(2184, true)
-  var client: CuratorFramework = null
-
-  val jobId = s"job_${System.currentTimeMillis}"
-  val maki = Source.fromURL(getClass.getResource("/simple-maki.yml")).mkString
-  val queue = new LinkedBlockingQueue[ActionData]()
-
-  var manager: JobManager = null
-
-  client = CuratorFrameworkFactory.newClient(server.getConnectString, retryPolicy)
-  client.start()
-
-  override def beforeEach(): Unit = {
-
-    // creating the jobs znode and storing the source repo and branch
-    client.create().withMode(CreateMode.PERSISTENT).forPath(s"/$jobId")
-    client.create().withMode(CreateMode.PERSISTENT).forPath(s"/$jobId/src", "".getBytes)
-    client.create().withMode(CreateMode.PERSISTENT).forPath(s"/$jobId/branch", "".getBytes)
-
-    manager = JobLoader.createJobManager(maki, jobId, client, 3, queue)
-
-  }
-
-  override def afterEach(): Unit = {
-
-    client.delete().deletingChildrenIfNeeded().forPath(s"/$jobId")
-
-  }
-
-  "a restored job" should "have all Queued actions in the executionQueue" in {
-
-    // setting task-0000000002 as Queued
-    client.setData().forPath(s"/${jobId}/task-0000000002", ActionStatus.Queued.toString.getBytes)
-
-    JobLoader.restoreJobState(manager, jobId, client)
-
-    queue.peek.getName should be("start")
-  }
-
-  "a restored job" should "have all Started actions in the executionQueue" in {
-
-    // setting task-0000000002 as Queued
-    client.setData().forPath(s"/${jobId}/task-0000000002", ActionStatus.Started.toString.getBytes)
-
-    JobLoader.restoreJobState(manager, jobId, client)
-
-    queue.peek.getName should be("start")
-  }
-}
diff --git a/leader/src/test/scala/org/apache/amaterasu/integration/GitTests.scala b/leader/src/test/scala/org/apache/amaterasu/integration/GitTests.scala
deleted file mode 100755
index 1e9b48b..0000000
--- a/leader/src/test/scala/org/apache/amaterasu/integration/GitTests.scala
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.integration
-
-import org.apache.amaterasu.leader.common.dsl.GitUtil
-import org.scalatest.{FlatSpec, Matchers}
-
-import scala.reflect.io.Path
-
-class GitTests extends FlatSpec with Matchers {
-
-  "GitUtil.cloneRepo" should "clone the sample job git repo" in {
-
-    val path = Path("repo")
-    path.deleteRecursively()
-
-    GitUtil.cloneRepo("https://github.com/shintoio/amaterasu-job-sample.git", "master", "", "")
-
-    val exists = new java.io.File("repo/maki.yml").exists
-    exists should be(true)
-  }
-}
diff --git a/leader/src/test/scala/org/apache/amaterasu/leader/mesos/MesosTestUtil.scala b/leader/src/test/scala/org/apache/amaterasu/leader/mesos/MesosTestUtil.scala
deleted file mode 100755
index 0397b87..0000000
--- a/leader/src/test/scala/org/apache/amaterasu/leader/mesos/MesosTestUtil.scala
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.leader.mesos
-
-import java.util
-import java.util.UUID
-
-import org.apache.mesos.Protos._
-
-
-object MesosTestUtil {
-  def createOffer(mem: Double, disk: Double, cpus: Int): Offer = {
-
-    val resources = new util.ArrayList[Resource]()
-    resources.add(createScalarResource("mem", mem))
-    resources.add(createScalarResource("disk", disk))
-    resources.add(createScalarResource("cpus", cpus))
-
-    Offer.newBuilder()
-      .setId(OfferID.newBuilder().setValue(UUID.randomUUID.toString))
-      .setFrameworkId(FrameworkID.newBuilder().setValue("Amaterasu"))
-      .setSlaveId(SlaveID.newBuilder().setValue(UUID.randomUUID.toString))
-      .setHostname("localhost")
-      .addAllResources(resources)
-
-      .build()
-
-  }
-
-  def createScalarResource(name: String, value: Double): org.apache.mesos.Protos.Resource = {
-
-    org.apache.mesos.Protos.Resource.newBuilder().setName(name).setType(Value.Type.SCALAR)
-      .setScalar(Value.Scalar.newBuilder.setValue(value).build()).build()
-
-  }
-}
diff --git a/leader/src/test/scala/org/apache/amaterasu/utilities/HttpServerTests.scala b/leader/src/test/scala/org/apache/amaterasu/utilities/HttpServerTests.scala
deleted file mode 100644
index 7eba1da..0000000
--- a/leader/src/test/scala/org/apache/amaterasu/utilities/HttpServerTests.scala
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.amaterasu.utilities
-
-
-import java.io.File
-
-import org.scalatest.{FlatSpec, Matchers}
-
-
-class HttpServerTests extends FlatSpec with Matchers {
-
-  // this is an ugly hack, getClass.getResource("/").getPath should have worked but
-  // stopped working when we moved to gradle :(
-  val resources: String = new File(getClass.getResource("/simple-maki.yml").getPath).getParent
-
-  //  "Jetty Web server" should "start HTTP server, serve content and stop successfully" in {
-  //
-  //    var data = ""
-  //    try {
-  //      HttpServer.start("8000", resources)
-  //      val html = Source.fromURL("http://localhost:8000/jetty-test-data.txt")
-  //      data = html.mkString
-  //    }
-  //    finally {
-  //      HttpServer.stop()
-  //    }
-  //    data should equal("This is a test file to download from Jetty webserver")
-  //  }
-
-//  "Jetty File server with '/' as root" should "start HTTP server, serve content and stop successfully" in {
-//
-//    var urlCount: Int = 0
-//    println("resource location" + resources)
-//    try {
-//      HttpServer.start("8000", resources)
-//      val urls = HttpServer.getFilesInDirectory("127.0.0.1", "8000", "dist")
-//      urls.foreach(println)
-//      urlCount = urls.length
-//    } catch {
-//      case e: Exception => println(s"++++>> ${e.getMessage}")
-//    }
-//    finally {
-//      HttpServer.stop()
-//    }
-//    urlCount should equal(2)
-//  }
-//
-//  "Jetty File server with 'dist' as root" should "start HTTP server, serve content and stop successfully" in {
-//    var data = ""
-//    var urlCount: Int = 0
-//    println("resource location" + resources)
-//    try {
-//      HttpServer.start("8000", resources + "/dist")
-//      val urls = HttpServer.getFilesInDirectory("localhost", "8000", "")
-//      urls.foreach(println)
-//      urlCount = urls.length
-//    }
-//    finally {
-//      HttpServer.stop()
-//    }
-//    urlCount should equal(2)
-//  }
-}
diff --git a/sdk/src/main/kotlin/org/apache/amaterasu/sdk/frameworks/FrameworkSetupProvider.kt b/sdk/src/main/kotlin/org/apache/amaterasu/sdk/frameworks/FrameworkSetupProvider.kt
index f68922e..6b0859f 100644
--- a/sdk/src/main/kotlin/org/apache/amaterasu/sdk/frameworks/FrameworkSetupProvider.kt
+++ b/sdk/src/main/kotlin/org/apache/amaterasu/sdk/frameworks/FrameworkSetupProvider.kt
@@ -26,11 +26,11 @@
 
     val groupIdentifier: String
 
-    val groupResources: Array<File>
+    val groupResources: List<File>
 
     val environmentVariables: Map<String, String>
 
-    val configurationItems: Array<String>
+    val configurationItems: List<String>
 
     fun init(env: String, conf: ClusterConfig)
 
diff --git a/settings.gradle b/settings.gradle
index 9e757dc..cc5161b 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -16,8 +16,8 @@
  */
 
 // Core
-include 'leader'
-project(':leader')
+//include 'leader'
+//project(':leader')
 
 include 'leader-common'
 project(':leader-common')
@@ -32,9 +32,6 @@
 include 'common'
 project(':common')
 
-include 'executor'
-project(':executor')
-
 include 'sdk'
 findProject(':sdk')?.name = 'amaterasu-sdk'