From 615bc8b2253a0605ba7fdd70227de43ffbb46919 Mon Sep 17 00:00:00 2001 From: HwangRock Date: Sat, 18 Jul 2026 18:41:46 +0900 Subject: [PATCH 1/2] [ZEPPELIN-6546] Restore Math.max accumulation for NoteJobInfo.unixTimeLastRun The NoteJobInfo constructor overwrote lastRunningUnixTime on every loop iteration, so a note's last-run time always came from its last paragraph instead of its most recently run one. This breaks Job Manager sorting and the incremental LIST_UPDATE_NOTE_JOBS filter, which can silently drop a note that just ran. ZEPPELIN-2860 fixed exactly this with Math.max in 2017 (commit 4a369f100). The ZEPPELIN-3737 refactor (commit 001c621c7) moved the logic into JobManagerService and dropped the Math.max, silently reintroducing the bug. Restore the max accumulation. --- .../java/org/apache/zeppelin/service/JobManagerService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/service/JobManagerService.java b/zeppelin-server/src/main/java/org/apache/zeppelin/service/JobManagerService.java index 6b65ef58d65..e699d58f490 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/service/JobManagerService.java +++ b/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); } From 0011ca3f3be334e6c8e1d0f45178add9f4e69626 Mon Sep 17 00:00:00 2001 From: HwangRock Date: Sat, 18 Jul 2026 18:41:55 +0900 Subject: [PATCH 2/2] [ZEPPELIN-6546] Add regression tests for unixTimeLastRun computation Two tests build a note whose first paragraph finished recently while the last paragraph was created earlier and never ran, then pin the computed unixTimeLastRun through the getNoteJobInfoByUnixTime filter boundary: the note must be included for thresholds below the newest paragraph timestamp and excluded at exactly that timestamp. Both tests fail without the Math.max restoration. --- .../service/JobManagerServiceTest.java | 83 +++++++++++++++++++ 1 file changed, 83 insertions(+) diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/service/JobManagerServiceTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/service/JobManagerServiceTest.java index e8d69d604b0..b56c8473081 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/service/JobManagerServiceTest.java +++ b/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 noteProcessor = invocation.getArgument(1); + return noteProcessor.process(mockNote); + }); + } + + @Test + void getNoteJobInfoByUnixTime_usesMaxParagraphTimestamp_notLastParagraph() throws IOException { + ServiceCallback> callback = new SimpleServiceCallback<>(); + + List result = jobManagerService.getNoteJobInfoByUnixTime( + (NEVER_RUN_CREATED_TIME + LAST_RUN_TIME) / 2, + serviceContext, + callback + ); + + assertEquals(1, result.size()); + } + + @Test + void getNoteJobInfoByUnixTime_boundaryIsExactlyMaxTimestamp() throws IOException { + ServiceCallback> callback = new SimpleServiceCallback<>(); + + List includedResult = jobManagerService.getNoteJobInfoByUnixTime( + LAST_RUN_TIME - 1, + serviceContext, + callback + ); + List excludedResult = jobManagerService.getNoteJobInfoByUnixTime( + LAST_RUN_TIME, + serviceContext, + callback + ); + + assertEquals(1, includedResult.size()); + assertTrue(excludedResult.isEmpty()); + } + } + }