Fixed data job status in case of OOM

projects/control-service/projects/pipelines_control_service/src/integration-test/java/com/vmware/taurus/datajobs/it/DataJobDeploymentCrudIT.java

@@ -92,7 +92,8 @@ public void testDataJobDeploymentCrud() throws Exception {
 
     // Take the job zip as byte array
     byte[] jobZipBinary =
-        IOUtils.toByteArray(getClass().getClassLoader().getResourceAsStream("simple_job.zip"));
+        IOUtils.toByteArray(
+            getClass().getClassLoader().getResourceAsStream("data_jobs/simple_job.zip"));
 
     // Execute job upload with no user
     mockMvc
@@ -376,7 +377,8 @@ public void cleanUp() throws Exception {
   @Test
   public void testDataJobDeleteSource() throws Exception {
     byte[] jobZipBinary =
-        IOUtils.toByteArray(getClass().getClassLoader().getResourceAsStream("simple_job.zip"));
+        IOUtils.toByteArray(
+            getClass().getClassLoader().getResourceAsStream("data_jobs/simple_job.zip"));
 
     mockMvc
         .perform(

projects/control-service/projects/pipelines_control_service/src/integration-test/java/com/vmware/taurus/datajobs/it/DataJobEphemeralStorageIT.java

@@ -5,8 +5,6 @@
 
 package com.vmware.taurus.datajobs.it;
 
-import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.vmware.taurus.ControlplaneApplication;
 import com.vmware.taurus.controlplane.model.data.DataJobExecution;
 import com.vmware.taurus.datajobs.it.common.BaseIT;
@@ -29,10 +27,6 @@
     classes = ControlplaneApplication.class)
 public class DataJobEphemeralStorageIT extends BaseIT {
 
-  private final ObjectMapper objectMapper =
-      new ObjectMapper()
-          .registerModule(new JavaTimeModule()); // Used for converting to OffsetDateTime;
-
   @RegisterExtension
   static DataJobDeploymentExtension dataJobDeploymentExtension =
       new DataJobDeploymentExtension("job_ephemeral_storage.zip");

projects/control-service/projects/pipelines_control_service/src/integration-test/java/com/vmware/taurus/datajobs/it/DataJobOOMIT.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2021 VMware, Inc.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package com.vmware.taurus.datajobs.it;
+
+import com.vmware.taurus.ControlplaneApplication;
+import com.vmware.taurus.controlplane.model.data.DataJobExecution;
+import com.vmware.taurus.datajobs.it.common.BaseIT;
+import com.vmware.taurus.datajobs.it.common.DataJobDeploymentExtension;
+import com.vmware.taurus.datajobs.it.common.JobExecutionUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.springframework.boot.test.context.SpringBootTest;
+
+
+@Slf4j
+@SpringBootTest(
+    webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+    classes = ControlplaneApplication.class)
+public class DataJobOOMIT extends BaseIT {
+
+  @RegisterExtension
+  static DataJobDeploymentExtension dataJobDeploymentExtension =
+      new DataJobDeploymentExtension("oom_job.zip");
+
+  @Test
+  public void testDataJob_causesOOM_shouldCompleteWithUserError(
+      String jobName, String teamName, String username, String deploymentId) throws Exception {
+    // manually start job execution
+    ImmutablePair<String, String> executeDataJobResult =
+        JobExecutionUtil.executeDataJob(jobName, teamName, username, deploymentId, mockMvc);
+    String opId = executeDataJobResult.getLeft();
+    String executionId = executeDataJobResult.getRight();
+
+    // Check the data job execution status
+    JobExecutionUtil.checkDataJobExecutionStatus(
+        executionId,
+        DataJobExecution.StatusEnum.USER_ERROR,
+        opId,
+        jobName,
+        teamName,
+        username,
+        mockMvc);
+  }
+}

projects/control-service/projects/pipelines_control_service/src/integration-test/java/com/vmware/taurus/datajobs/it/common/DataJobDeploymentExtension.java

@@ -70,6 +70,8 @@ public class DataJobDeploymentExtension
   private static final String DEPLOYMENT_ID = "NOT_USED";
   private static final String TEAM_NAME = "test-team";
 
+  private static final String JOB_SOURCE_PATH = "data_jobs/";
+
   protected final ObjectMapper MAPPER = new ObjectMapper();
 
   private String jobName =
@@ -124,7 +126,8 @@ public void beforeEach(ExtensionContext context) throws Exception {
 
     if (!initialized) {
       byte[] jobZipBinary =
-          IOUtils.toByteArray(getClass().getClassLoader().getResourceAsStream(jobSource));
+          IOUtils.toByteArray(
+              getClass().getClassLoader().getResourceAsStream(JOB_SOURCE_PATH + jobSource));
 
       // Upload the data job
       MvcResult uploadResult =

projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/KubernetesService.java

@@ -1652,6 +1652,12 @@ Optional<JobExecution> getJobExecutionStatus(V1Job job, JobStatusCondition jobSt
     jobExecutionStatusBuilder.succeeded(
         Optional.ofNullable(jobStatusCondition).map(JobStatusCondition::isSuccess).orElse(null));
 
+    // omits events that come after the Data Job completion
+    if (jobExecutionStatusBuilder.succeeded != null
+        && StringUtils.isBlank(jobExecutionStatusBuilder.containerTerminationReason)) {
+      return Optional.empty();
+    }
+
     return Optional.of(jobExecutionStatusBuilder.build());
   }