[b/420369667] fix NPE for Oozie end date (#910)

* fix NPE for Oozie end date

* add test for filtering jobs with null end date

* fix code style

* remove empty line

Author: vladislav-sidorovich

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/connector/hadoop/oozie/AbstractOozieJobsTask.java

@@ -34,6 +34,7 @@
 import java.util.List;
 import javax.annotation.CheckForNull;
 import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 import org.apache.commons.beanutils.PropertyUtils;
 import org.apache.commons.csv.CSVFormat;
 import org.apache.commons.csv.CSVPrinter;
@@ -100,7 +101,8 @@ protected Void doRun(TaskRunContext context, @Nonnull ByteSink sink, @Nonnull Ha
         for (J job : jobs) {
           Date currentJobEndTime = getJobEndTime(job);
           boolean inDateRange =
-              minJobEndTimeTimestamp <= currentJobEndTime.getTime()
+              currentJobEndTime != null
+                  && minJobEndTimeTimestamp <= currentJobEndTime.getTime()
                   && currentJobEndTime.getTime() < maxJobEndTimeTimestamp;
           if (!inDateRange) {
             // It's client side filtering. It's inefficient.
@@ -120,10 +122,10 @@ protected Void doRun(TaskRunContext context, @Nonnull ByteSink sink, @Nonnull Ha
 
         J lastJob = jobs.get(jobs.size() - 1);
         Date endTime = getJobEndTime(lastJob);
-        if (endTime == null) {
-          break;
+        boolean isLastJobInProgress = endTime == null;
+        if (!isLastJobInProgress) {
+          latestFetchedJobEndTimestamp = endTime.getTime();
         }
-        latestFetchedJobEndTimestamp = endTime.getTime();
         offset += jobs.size();
       }
 
@@ -147,6 +149,7 @@ CSVFormat createJobSpecificCSVFormat() {
   abstract List<J> fetchJobsWithFilter(
       XOozieClient oozieClient, String oozieFilter, int start, int len) throws OozieClientException;
 
+  @Nullable
   abstract Date getJobEndTime(J job);
 
   private static String toISO(ZonedDateTime dateTime) {

dumper/app/src/test/java/com/google/edwmigration/dumper/application/dumper/connector/hadoop/oozie/OozieBundleJobsTaskTest.java

@@ -18,6 +18,7 @@
 
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
@@ -61,6 +62,13 @@ public void getJobEndTime_success() throws Exception {
     verifyNoMoreInteractions(job);
   }
 
+  @Test
+  public void getJobEndTime_nullable() {
+    BundleJob job = mock(BundleJob.class);
+
+    assertNull("null is expected value for end time", task.getJobEndTime(job));
+  }
+
   @Test
   public void csvHeadersContainRequiredFields() {
     String[] header = task.createJobSpecificCSVFormat().getHeader();

dumper/app/src/test/java/com/google/edwmigration/dumper/application/dumper/connector/hadoop/oozie/OozieCoordinatorJobsTaskTest.java

@@ -18,6 +18,7 @@
 
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
@@ -61,6 +62,13 @@ public void getJobEndTime_success() throws Exception {
     verifyNoMoreInteractions(job);
   }
 
+  @Test
+  public void getJobEndTime_nullable() {
+    CoordinatorJob job = mock(CoordinatorJob.class);
+
+    assertNull("null is expected value for end time", task.getJobEndTime(job));
+  }
+
   @Test
   public void csvHeadersContainRequiredFields() {
     String[] header = task.createJobSpecificCSVFormat().getHeader();

dumper/app/src/test/java/com/google/edwmigration/dumper/application/dumper/connector/hadoop/oozie/OozieWorkflowJobsTaskTest.java

@@ -24,6 +24,7 @@
 import static com.google.common.net.HttpHeaders.CONTENT_TYPE;
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
@@ -189,6 +190,54 @@ public void doRun_requestBatchesFilterOnClient_ok() throws Exception {
     assertEquals(expected, actual);
   }
 
+  @Test
+  public void doRun_requestBatchesFilterOnClient_nullDate_success() throws Exception {
+    final ZonedDateTime endTime =
+        ZonedDateTime.ofInstant(Instant.ofEpochMilli(timestampInMockResponses), UTC).plusHours(1);
+    final ZonedDateTime startTime = endTime.minusDays(7);
+    Date lastCapturedDate = Date.from(startTime.toInstant());
+    when(context.getArguments()).thenReturn(new ConnectorArguments("--connector", "oozie"));
+    MemoryByteSink sink = new MemoryByteSink();
+    stubOozieVersionsCall();
+    server.stubFor(
+        get(urlEqualTo("/v2/jobs?filter=sortby%3DendTime%3B&jobtype=wf&offset=1&len=1000"))
+            .willReturn(
+                okJsonWithBodyFile("oozie/jobs-one-item-template.json")
+                    .withTransformers("response-template")
+                    .withTransformerParameter(
+                        // filter out this job because endTime is null (job is in progress)
+                        "endTime", null)));
+    server.stubFor(
+        get(urlEqualTo("/v2/jobs?filter=sortby%3DendTime%3B&jobtype=wf&offset=2&len=1000"))
+            .willReturn(
+                okJsonWithBodyFile("oozie/jobs-one-item-template.json")
+                    .withTransformers("response-template")
+                    .withTransformerParameter(
+                        // include this job
+                        "endTime", JsonUtils.formatDateRfc822(lastCapturedDate))));
+    server.stubFor(
+        get(urlEqualTo("/v2/jobs?filter=sortby%3DendTime%3B&jobtype=wf&offset=3&len=1000"))
+            .willReturn(
+                okJsonWithBodyFile("oozie/jobs-one-item-template.json")
+                    .withTransformers("response-template")
+                    .withTransformerParameter(
+                        // filter out this job, because out of range
+                        // before start
+                        "endTime",
+                        JsonUtils.formatDateRfc822(
+                            Date.from(startTime.minusSeconds(1).toInstant())))));
+
+    OozieWorkflowJobsTask task = new OozieWorkflowJobsTask(startTime, endTime);
+
+    // Act
+    task.doRun(context, sink, new OozieHandle(oozieClient));
+
+    // Assert
+    String actual = sink.getContent();
+    String expected = readFileAsString("/oozie/expected-jobs-one-job-from-template.csv");
+    assertEquals(expected, actual);
+  }
+
   @Test
   public void doRun_requestBatchesUntilEmptyResponse_ok() throws Exception {
     when(context.getArguments()).thenReturn(new ConnectorArguments("--connector", "oozie"));
@@ -284,6 +333,14 @@ public void getJobEndTime_success() throws Exception {
     verifyNoMoreInteractions(job);
   }
 
+  @Test
+  public void getJobEndTime_nullable() {
+    OozieWorkflowJobsTask task = newOozieWorkflowJobTaskForOneDay();
+    WorkflowJob job = mock(WorkflowJob.class);
+
+    assertNull("null is expected value for end time", task.getJobEndTime(job));
+  }
+
   @Test
   public void csvHeadersContainRequiredFields() {
     OozieWorkflowJobsTask task = newOozieWorkflowJobTaskForOneDay();

dumper/app/src/test/resources/oozie/expected-jobs-one-job-from-template.csv

@@ -0,0 +1,3 @@
+acl,actions,appName,appPath,conf,consoleUrl,createdTime,endTime,externalId,group,id,lastModifiedTime,parentId,run,startTime,status,user
+,[],Empty workflow,,,https://oozie-host:11443/oozie?job=0000128-250317102116043-oozie-oozi-W,1741701600000,1741618860000,,,0000128-250317102116043-oozie-oozi-W,1742220060000,0000002-250307121251646-oozie-oozi-C@473,0,1742220060000,SUCCEEDED,csso_dev
+