[ZEPPELIN-6546] Restore Math.max accumulation for NoteJobInfo.unixTimeLastRun

zeppelin-server/src/main/java/org/apache/zeppelin/service/JobManagerService.java

@@ -194,7 +194,7 @@ public NoteJobInfo(Note note) {
         }
         // get data for the job manager.
         ParagraphJobInfo paragraphItem = new ParagraphJobInfo(paragraph);
-        lastRunningUnixTime = getUnixTimeLastRunParagraph(paragraph);
+        lastRunningUnixTime = Math.max(lastRunningUnixTime, getUnixTimeLastRunParagraph(paragraph));
         paragraphs.add(paragraphItem);
       }
 

zeppelin-server/src/test/java/org/apache/zeppelin/service/JobManagerServiceTest.java

@@ -18,6 +18,7 @@
 package org.apache.zeppelin.service;
 
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -28,10 +29,16 @@
 import static org.mockito.Mockito.when;
 
 import java.io.IOException;
+import java.util.Collections;
+import java.util.Date;
 import java.util.List;
 import org.apache.zeppelin.conf.ZeppelinConfiguration;
 import org.apache.zeppelin.notebook.AuthorizationService;
+import org.apache.zeppelin.notebook.Note;
+import org.apache.zeppelin.notebook.NoteInfo;
 import org.apache.zeppelin.notebook.Notebook;
+import org.apache.zeppelin.notebook.Paragraph;
+import org.apache.zeppelin.scheduler.Job;
 import org.apache.zeppelin.service.JobManagerService.NoteJobInfo;
 import org.apache.zeppelin.service.exception.JobManagerForbiddenException;
 import org.apache.zeppelin.user.AuthenticationInfo;
@@ -111,4 +118,80 @@ void removeNoteJobInfo_doesNothing() {
     }
   }
 
+  @Nested
+  class WhenJobManagerIsEnabled {
+
+    private static final long LAST_RUN_TIME = 200_000L;
+    private static final long NEVER_RUN_CREATED_TIME = 100_000L;
+
+    private Note mockNote;
+
+    @BeforeEach
+    void enableJobManager() throws IOException {
+      when(zConf.isJobManagerEnabled()).thenReturn(true);
+
+      mockNote = mock(Note.class);
+      when(mockNote.getId()).thenReturn("note1");
+      when(mockNote.getName()).thenReturn("note1");
+      when(mockNote.getConfig()).thenReturn(Collections.emptyMap());
+      when(mockNote.getDefaultInterpreterGroup()).thenReturn("spark");
+
+      Paragraph lastRunParagraph = mock(Paragraph.class);
+      when(lastRunParagraph.isTerminated()).thenReturn(true);
+      when(lastRunParagraph.getDateFinished()).thenReturn(new Date(LAST_RUN_TIME));
+      when(lastRunParagraph.getStatus()).thenReturn(Job.Status.FINISHED);
+      when(lastRunParagraph.getId()).thenReturn("p1");
+      when(lastRunParagraph.getTitle()).thenReturn(null);
+
+      Paragraph neverRunParagraph = mock(Paragraph.class);
+      when(neverRunParagraph.isTerminated()).thenReturn(false);
+      when(neverRunParagraph.isRunning()).thenReturn(false);
+      when(neverRunParagraph.getDateCreated()).thenReturn(new Date(NEVER_RUN_CREATED_TIME));
+      when(neverRunParagraph.getStatus()).thenReturn(Job.Status.READY);
+      when(neverRunParagraph.getId()).thenReturn("p2");
+      when(neverRunParagraph.getTitle()).thenReturn(null);
+
+      when(mockNote.getParagraphs()).thenReturn(List.of(lastRunParagraph, neverRunParagraph));
+
+      when(mockNotebook.getNotesInfo()).thenReturn(List.of(new NoteInfo("note1", "note1.zpln")));
+      when(mockAuthorizationService.isOwner(any(), eq("note1"))).thenReturn(true);
+      when(mockNotebook.processNote(eq("note1"), any())).thenAnswer(invocation -> {
+        Notebook.NoteProcessor<Object> noteProcessor = invocation.getArgument(1);
+        return noteProcessor.process(mockNote);
+      });
+    }
+
+    @Test
+    void getNoteJobInfoByUnixTime_usesMaxParagraphTimestamp_notLastParagraph() throws IOException {
+      ServiceCallback<List<NoteJobInfo>> callback = new SimpleServiceCallback<>();
+
+      List<NoteJobInfo> result = jobManagerService.getNoteJobInfoByUnixTime(
+          (NEVER_RUN_CREATED_TIME + LAST_RUN_TIME) / 2,
+          serviceContext,
+          callback
+      );
+
+      assertEquals(1, result.size());
+    }
+
+    @Test
+    void getNoteJobInfoByUnixTime_boundaryIsExactlyMaxTimestamp() throws IOException {
+      ServiceCallback<List<NoteJobInfo>> callback = new SimpleServiceCallback<>();
+
+      List<NoteJobInfo> includedResult = jobManagerService.getNoteJobInfoByUnixTime(
+          LAST_RUN_TIME - 1,
+          serviceContext,
+          callback
+      );
+      List<NoteJobInfo> excludedResult = jobManagerService.getNoteJobInfoByUnixTime(
+          LAST_RUN_TIME,
+          serviceContext,
+          callback
+      );
+
+      assertEquals(1, includedResult.size());
+      assertTrue(excludedResult.isEmpty());
+    }
+  }
+
 }