Fix restartFn bug

Author: kjanecki

functions/kubernetes/amqpConnector.js

@@ -9,7 +9,7 @@ async function initialize(queue_name) {
         conn = await amqplib.connect(`amqp://${process.env.RABBIT_HOSTNAME}`, "heartbeat=60");
     }
     let ch = await conn.createChannel()
-    await ch.assertQueue(queue_name, {durable: false, expires: 600000}); // TODO: implement dynamic queue creation & cleanup
+    await ch.assertQueue(queue_name, {durable: false, expires: 6000000}); // TODO: implement dynamic queue creation & cleanup
     channels[queue_name] = ch
 
 }

functions/kubernetes/k8sCommand.js

@@ -129,9 +129,9 @@ async function k8sCommandGroup(bufferItems) {
     if (getExecutorType(context) === "WORKER_POOL") {
       await amqpEnqueueJobs(jobArr, taskIdArr, contextArr, customParams)
     } else {
-      await submitK8sJob(kubeconfig, jobArr, taskIdArr, contextArr, customParams, restartFn)
+      await submitK8sJob(kubeconfig, jobArr, taskIdArr, contextArr, customParams)
     }
-    jobExitCodes = await synchronizeJobs(jobArr, taskIdArr, contextArr, customParams);
+    jobExitCodes = await synchronizeJobs(jobArr, taskIdArr, contextArr, customParams, restartFn);
   } catch (err) {
     console.log("Error when submitting job:", err);
     throw err;

functions/kubernetes/k8sJobSubmit.js

@@ -105,7 +105,7 @@ var createK8sJobYaml = (job, taskIds, context, jobYamlTemplate, customParams) =>
 //
 //
 // Returns: job exit code
-var submitK8sJob = async(kubeconfig, jobArr, taskIdArr, contextArr, customParams, restartFn) => {
+var submitK8sJob = async(kubeconfig, jobArr, taskIdArr, contextArr, customParams) => {
 
   // Load definition of the the worker job pod
   // File 'job-template.yaml' should be provided externally during deployment