From bc1ea421a54c1570572b2660fdd95bada115e32c Mon Sep 17 00:00:00 2001 From: wudi Date: Fri, 23 Oct 2026 16:57:18 +0800 Subject: [PATCH 1/2] [fix](fe) Avoid Meta Service queries during streaming job replay (#69498) ### What problem does this PR solve? Issue Number: N/A Related PR: #57264 Problem Summary: Cloud streaming jobs query Meta Service for progress while replaying each job update journal, making FE replay depend on Meta Service availability and RPC latency. ALTER already persists the updated offset provider state and job statistics in the journal. Restore the provider state and cloud statistics locally during replay instead of querying Meta Service. A paused job therefore retains its altered offset and journaled statistics after an FE restart, before RESUME. PENDING scheduling still fetches the authoritative committed progress from Meta Service before dispatching tasks, and pauses the job if that fetch fails. ### Release note Cloud streaming jobs restore journaled offsets and statistics during FE replay without querying Meta Service. Progress is still synchronized from Meta Service before consumption resumes. --- .../insert/streaming/StreamingInsertJob.java | 5 +-- ...reamingInsertJobOffsetPersistenceTest.java | 44 +++++++++++++++++++ 2 files changed, 46 insertions(+), 3 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java index 71f94723427436..7343a7b723d023 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java @@ -1125,7 +1125,7 @@ public void replayOnUpdated(StreamingInsertJob replayJob) { modifyPropertiesInternal(replayJob.getProperties()); if (replayPersistInfo != null) { setOffsetProviderPersist(replayPersistInfo); - if (tvfType != null && !Config.isCloudMode()) { + if (tvfType != null) { offsetProvider.restoreFromPersistInfo(offsetProviderPersist); } } else { @@ -1137,9 +1137,8 @@ public void replayOnUpdated(StreamingInsertJob replayJob) { offsetProviderPersist, offsetProvider.getPersistInfo()); } } - // When the pause state is restarted, it also needs to be updated if (Config.isCloudMode()) { - replayOnCloudMode(); + this.jobStatistic = replayJob.getJobStatistic(); } } catch (Exception e) { // should not happen diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java index 65d94c8e672691..68249928d8b962 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java @@ -387,6 +387,50 @@ public void testCloudReplayRefreshesPersistedOffset() { Assert.assertFalse(recovered.hasMoreDataToConsume()); } + @Test + public void testCloudReplayDoesNotFetchProgressAndPendingFailurePreventsDispatch() throws Exception { + TestStreamingInsertJob job = Mockito.spy(newJob(new S3SourceOffsetProvider(), 1021L)); + Deencapsulation.setField(job, "tvfType", S3TableValuedFunction.NAME); + Deencapsulation.setField(job, "runningStreamTask", null); + job.offsetProvider.restoreFromPersistInfo("{\"endFile\":\"data/a.csv\"}"); + StreamingInsertJob replayJob = new StreamingInsertJob(); + Deencapsulation.setField(replayJob, "jobId", 9001L); + Deencapsulation.setField(replayJob, "properties", new HashMap()); + Deencapsulation.setField(replayJob, "tvfType", S3TableValuedFunction.NAME); + replayJob.setJobStatus(JobStatus.PAUSED); + replayJob.getProperties().put(StreamingJobProperties.OFFSET_PROPERTY, "{\"fileName\":\"data/b.csv\"}"); + replayJob.setOffsetProviderPersist("{\"endFile\":\"data/b.csv\"}"); + replayJob.getJobStatistic().setScannedRows(10); + replayJob.getJobStatistic().setLoadBytes(218); + replayJob.getJobStatistic().setFileNumber(1); + replayJob.getJobStatistic().setFileSize(138); + replayJob.getJobStatistic().setFilteredRows(2); + StreamingInsertJob replayRecord = GsonUtils.GSON.fromJson( + GsonUtils.GSON.toJson(replayJob, StreamingInsertJob.class), StreamingInsertJob.class); + Mockito.doThrow(new JobException("MS unavailable")).when(job).replayOnCloudMode(); + + try (MockedStatic config = Mockito.mockStatic(Config.class)) { + config.when(Config::isCloudMode).thenReturn(true); + + job.replayOnUpdated(replayRecord); + + Mockito.verify(job, Mockito.never()).replayOnCloudMode(); + Assertions.assertEquals(JobStatus.PAUSED, job.getJobStatus()); + Assertions.assertEquals("{\"endFile\":\"data/b.csv\"}", job.getOffsetProviderPersist()); + Assertions.assertEquals("{\"fileName\":\"data/b.csv\"}", job.offsetProvider.getShowCurrentOffset()); + Assertions.assertEquals(replayJob.getJobStatistic().toJson(), job.getJobStatistic().toJson()); + + job.setJobStatus(JobStatus.PENDING); + new StreamingJobSchedulerTask(job).run(); + + Mockito.verify(job).replayOnCloudMode(); + Assertions.assertEquals(JobStatus.PAUSED, job.getJobStatus()); + Assertions.assertTrue(job.getFailureReason().getMsg().contains("MS unavailable")); + Mockito.verify(job, Mockito.never()).replayOffsetProviderIfNeed(); + Mockito.verify(job, Mockito.never()).createStreamingTask(); + } + } + @Test public void testReplayUpdatedRestoresFinalStateAndRemovesCallback() { TestStreamingInsertJob job = newJob(new JdbcSourceOffsetProvider(), 1013L); From d0e32c02ec118e0671b8fa7de9f4ed3b98053e3a Mon Sep 17 00:00:00 2001 From: JNSimba <676366545@qq.com> Date: Mon, 26 Oct 2026 10:18:54 +0800 Subject: [PATCH 2/2] [fix](fe) Use JUnit 4 assertions in streaming replay backport test Fix the test compilation failure in PR #69537 targeting branch-4.2. The cherry-picked test uses Assertions without an import, while the existing test class uses org.junit.Assert. Use Assert for all six added assertions. Validation: Maven FE reactor test compilation and Checkstyle passed; StreamingInsertJobOffsetPersistenceTest ran 18 tests with no failures or errors. --- .../StreamingInsertJobOffsetPersistenceTest.java | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java index 68249928d8b962..9ee3668ca92868 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java @@ -415,17 +415,17 @@ public void testCloudReplayDoesNotFetchProgressAndPendingFailurePreventsDispatch job.replayOnUpdated(replayRecord); Mockito.verify(job, Mockito.never()).replayOnCloudMode(); - Assertions.assertEquals(JobStatus.PAUSED, job.getJobStatus()); - Assertions.assertEquals("{\"endFile\":\"data/b.csv\"}", job.getOffsetProviderPersist()); - Assertions.assertEquals("{\"fileName\":\"data/b.csv\"}", job.offsetProvider.getShowCurrentOffset()); - Assertions.assertEquals(replayJob.getJobStatistic().toJson(), job.getJobStatistic().toJson()); + Assert.assertEquals(JobStatus.PAUSED, job.getJobStatus()); + Assert.assertEquals("{\"endFile\":\"data/b.csv\"}", job.getOffsetProviderPersist()); + Assert.assertEquals("{\"fileName\":\"data/b.csv\"}", job.offsetProvider.getShowCurrentOffset()); + Assert.assertEquals(replayJob.getJobStatistic().toJson(), job.getJobStatistic().toJson()); job.setJobStatus(JobStatus.PENDING); new StreamingJobSchedulerTask(job).run(); Mockito.verify(job).replayOnCloudMode(); - Assertions.assertEquals(JobStatus.PAUSED, job.getJobStatus()); - Assertions.assertTrue(job.getFailureReason().getMsg().contains("MS unavailable")); + Assert.assertEquals(JobStatus.PAUSED, job.getJobStatus()); + Assert.assertTrue(job.getFailureReason().getMsg().contains("MS unavailable")); Mockito.verify(job, Mockito.never()).replayOffsetProviderIfNeed(); Mockito.verify(job, Mockito.never()).createStreamingTask(); }