This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 08f19f63ae2 branch-4.2: [fix](fe) Avoid Meta Service queries during
streaming job replay #69498 (#69537)
08f19f63ae2 is described below
commit 08f19f63ae2c735ae3cd5614740302d8f4789472
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Oct 26 14:21:44 2026 +0800
branch-4.2: [fix](fe) Avoid Meta Service queries during streaming job
replay #69498 (#69537)
Cherry-picked from #69498
---------
Co-authored-by: wudi <[email protected]>
Co-authored-by: JNSimba <[email protected]>
---
.../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 71f94723427..7343a7b723d 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 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 {
@@ -1137,9 +1137,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 65d94c8e672..9ee3668ca92 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 class StreamingInsertJobOffsetPersistenceTest {
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<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();
+ 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();
+ 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();
+ }
+ }
+
@Test
public void testReplayUpdatedRestoresFinalStateAndRemovesCallback() {
TestStreamingInsertJob job = newJob(new JdbcSourceOffsetProvider(),
1013L);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
