Don't persist additional job arguments. (#797)

Job arguments which are generated just before a job is launched should
not be persisted to the state store.

This resolves #779 and #794.

Author: brndnmtthws

src/main/scala/org/apache/mesos/chronos/scheduler/jobs/TaskManager.scala

@@ -173,14 +173,7 @@ class TaskManager @Inject()(
         jobsObserver.apply(JobExpired(jobOption.get, taskId))
         None
       } else {
-        val jobArguments = TaskUtils.getJobArgumentsForTaskId(taskId)
-        var job = jobOption.get
-
-        if (jobArguments != null && !jobArguments.isEmpty) {
-          job = JobUtils.getJobWithArguments(job, jobArguments)
-        }
-
-        Some(taskId, job)
+        Some(taskId, jobOption.get)
       }
     }
   }

src/main/scala/org/apache/mesos/chronos/scheduler/mesos/MesosTaskBuilder.scala

@@ -9,12 +9,7 @@ import org.apache.mesos.Protos.ContainerInfo.DockerInfo
 import org.apache.mesos.Protos.Environment.Variable
 import org.apache.mesos.Protos._
 import org.apache.mesos.chronos.scheduler.config.SchedulerConfiguration
-import org.apache.mesos.chronos.scheduler.jobs.{
-  BaseJob,
-  ContainerType,
-  Fetch,
-  TaskUtils
-}
+import org.apache.mesos.chronos.scheduler.jobs.{Volume => _, _}
 
 import scala.collection.JavaConverters._
 import scala.collection.Map
@@ -86,22 +81,30 @@ class MesosTaskBuilder @Inject()(val conf: SchedulerConfiguration) {
     if (job.executor.nonEmpty) {
       appendExecutorData(taskInfo, job, environment, uriCommand)
     } else {
+      val jobArguments = TaskUtils.getJobArgumentsForTaskId(taskId.getValue)
+      val jobWithCommand = if (jobArguments != null && !jobArguments.isEmpty) {
+        JobUtils.getJobWithArguments(job, jobArguments)
+      } else {
+        job
+      }
+
       val command = CommandInfo.newBuilder()
-      if (job.command.startsWith("http") || job.command.startsWith("ftp")) {
+      if (jobWithCommand.command.startsWith("http") || jobWithCommand.command
+            .startsWith("ftp")) {
         val uri1 = CommandInfo.URI
           .newBuilder()
-          .setValue(job.command)
+          .setValue(jobWithCommand.command)
           .setExecutable(true)
           .build()
 
         command
           .addUris(uri1)
-          .setValue(
-            "\"." + job.command.substring(job.command.lastIndexOf("/")) + "\"")
+          .setValue("\"." + jobWithCommand.command.substring(
+            jobWithCommand.command.lastIndexOf("/")) + "\"")
           .setEnvironment(environment)
       } else {
-        val jobHasCmd = !job.command.isEmpty
-        if (jobHasCmd) command.setValue(job.command)
+        val jobHasCmd = !jobWithCommand.command.isEmpty
+        if (jobHasCmd) command.setValue(jobWithCommand.command)
 
         command
           .setShell(job.shell)