This is an automated email from the ASF dual-hosted git repository.
liaoxin01 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 1d132b6a46b [fix](fe) Avoid Meta Service queries during streaming job
replay (#69498)
1d132b6a46b is described below
commit 1d132b6a46ba18d3601e38ee0dc6686cda9d5c78
Author: wudi <[email protected]>
AuthorDate: Fri Oct 23 16:57:18 2026 +0800
[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 +--
.../StreamingInsertJobOffsetPersistenceTest.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 0b99218be95..add7761ddc2 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
@@ -1157,7 +1157,7 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
modifyPropertiesInternal(replayJob.getProperties());
if (replayPersistInfo != null) {
setOffsetProviderPersist(replayPersistInfo);
- if (tvfType != null && !Config.isCloudMode()) {
+ if (tvfType != null) {
offsetProvider.restoreFromPersistInfo(offsetProviderPersist);
}
} else {
@@ -1169,9 +1169,8 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
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 b0b30708725..63bce5df853 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
@@ -386,6 +386,50 @@ public class StreamingInsertJobOffsetPersistenceTest {
Assertions.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<String,
String>());
+ 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> 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);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
