Fix batch bugs (#1471)

Author: vishalbollu

pkg/operator/resources/batchapi/enqueue.go

@@ -236,8 +236,13 @@ func enqueueS3FileContents(jobSpec *spec.Job, delimitedFiles *schema.DelimitedFi
 		s3Path := awslib.S3Path(bucket, *s3Obj.Key)
 		writeToJobLogStream(jobSpec.JobKey, fmt.Sprintf("enqueuing contents from file %s", s3Path))
 
+		awsClientForBucket, err := awslib.NewFromClientS3Path(s3Path, config.AWS)
+		if err != nil {
+			return false, err
+		}
+
 		itemIndex := 0
-		err := config.AWS.S3FileIterator(bucket, s3Obj, _s3DownloadChunkSize, func(readCloser io.ReadCloser, isLastChunk bool) (bool, error) {
+		err = awsClientForBucket.S3FileIterator(bucket, s3Obj, _s3DownloadChunkSize, func(readCloser io.ReadCloser, isLastChunk bool) (bool, error) {
 			_, err := bytesBuffer.ReadFrom(readCloser)
 			if err != nil {
 				return false, err

pkg/operator/resources/batchapi/job.go

@@ -191,15 +191,15 @@ func deployJob(apiSpec *spec.API, jobSpec *spec.Job, submission *schema.JobSubmi
 		return
 	}
 
-	err = setRunningStatus(jobSpec.JobKey)
+	err = createK8sJob(apiSpec, jobSpec)
 	if err != nil {
 		handleJobSubmissionError(jobSpec.JobKey, err)
-		return
 	}
 
-	err = createK8sJob(apiSpec, jobSpec)
+	err = setRunningStatus(jobSpec.JobKey)
 	if err != nil {
 		handleJobSubmissionError(jobSpec.JobKey, err)
+		return
 	}
 }
 

pkg/operator/resources/batchapi/manage_resources_cron.go

@@ -132,10 +132,12 @@ func ManageJobResources() error {
 			continue
 		}
 
-		err = checkIfJobCompleted(jobKey, *queueURL, k8sJob)
-		if err != nil {
-			telemetry.Error(err)
-			errors.PrintError(err)
+		if jobState.Status == status.JobRunning {
+			err = checkIfJobCompleted(jobKey, *queueURL, k8sJob)
+			if err != nil {
+				telemetry.Error(err)
+				errors.PrintError(err)
+			}
 		}
 	}
 

pkg/operator/resources/batchapi/s3_iterator.go

@@ -52,7 +52,12 @@ func s3IteratorFromLister(s3Lister schema.S3Lister, fn func(string, *s3.Object)
 			return err
 		}
 
-		err = config.AWS.S3Iterator(bucket, key, false, nil, func(s3Obj *s3.Object) (bool, error) {
+		awsClientForBucket, err := aws.NewFromClientS3Path(s3Path, config.AWS)
+		if err != nil {
+			return err
+		}
+
+		err = awsClientForBucket.S3Iterator(bucket, key, false, nil, func(s3Obj *s3.Object) (bool, error) {
 			s3FilePath := aws.S3Path(bucket, *s3Obj.Key)
 
 			shouldSkip := false