This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new d91f703cfb1 branch-4.1: [fix](fe) Persist routine load job before
scheduling #67681- #68560 (#68773)
d91f703cfb1 is described below
commit d91f703cfb19cd7a30a4140338c5091ca5c2abac
Author: Refrain <[email protected]>
AuthorDate: Tue Oct 13 12:10:03 2026 +0800
branch-4.1: [fix](fe) Persist routine load job before scheduling #67681-
#68560 (#68773)
### What problem does this PR solve?
Issue Number: None
Related PR: #67681, #68560
Problem Summary:
Manual backport of #67681 to branch-4.1, replacing the closed automatic
backport #68560.
Persist the routine load create journal before publishing the job to the
scheduler. This keeps the initial NEED_SCHEDULE state stable during
serialization and prevents replay from restoring a RUNNING job without
tasks.
Adapt RoutineLoadJobPersistenceTest to branch-4.1: use the Kafka classes
in their existing routineload package and remove unsupported Kinesis
references. The Kafka test retains assertions for
journal-before-publication ordering and successful scheduling after
publication. This addresses the test compilation failure in TeamCity
build 1062738.
### Release note
Fix routine load jobs potentially becoming stuck after replay due to
scheduling before the create journal is persisted.
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test
- Unit test adapted for branch-4.1. Local compilation and tests were not
run, as requested. `git diff --check` passed.
- Behavior changed:
- [ ] No.
- [x] Yes. The create journal is persisted before the job becomes
visible to the scheduler.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../doris/load/routineload/RoutineLoadManager.java | 3 +-
.../routineload/RoutineLoadJobPersistenceTest.java | 132 +++++++++++++++++++++
2 files changed, 134 insertions(+), 1 deletion(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
index 6ef436e0b37..42cb12c4786 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
@@ -223,8 +223,9 @@ public class RoutineLoadManager implements Writable {
+ " routine load jobs are running. exceed limit.");
}
- unprotectedAddJob(routineLoadJob);
+ // Persist the initial job before the scheduler can change its
state or create tasks.
Env.getCurrentEnv().getEditLog().logCreateRoutineLoadJob(routineLoadJob);
+ unprotectedAddJob(routineLoadJob);
} finally {
writeUnlock();
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobPersistenceTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobPersistenceTest.java
new file mode 100644
index 00000000000..365a8a210c9
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobPersistenceTest.java
@@ -0,0 +1,132 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.load.routineload;
+
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.load.routineload.RoutineLoadJob.JobState;
+import org.apache.doris.persist.EditLog;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.transaction.GlobalTransactionMgrIface;
+import org.apache.doris.transaction.TxnStateCallbackFactory;
+
+import com.google.common.collect.Lists;
+import com.google.gson.JsonParser;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.EnumSet;
+
+public class RoutineLoadJobPersistenceTest {
+ private MockedStatic<Env> envMock;
+ private ConnectContext previousContext;
+ private EditLog editLog;
+ private RoutineLoadManager manager;
+ private TxnStateCallbackFactory callbackFactory;
+
+ @BeforeEach
+ public void setUp() {
+ previousContext = ConnectContext.get();
+ ConnectContext.remove();
+ envMock = Mockito.mockStatic(Env.class);
+ Env env = Mockito.mock(Env.class);
+ GlobalTransactionMgrIface transactionMgr =
Mockito.mock(GlobalTransactionMgrIface.class);
+ editLog = Mockito.mock(EditLog.class);
+ manager = new RoutineLoadManager();
+ callbackFactory = new TxnStateCallbackFactory();
+ envMock.when(Env::getCurrentEnv).thenReturn(env);
+
envMock.when(Env::getCurrentGlobalTransactionMgr).thenReturn(transactionMgr);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+ Mockito.when(env.getRoutineLoadManager()).thenReturn(manager);
+
Mockito.when(env.getRoutineLoadTaskScheduler()).thenReturn(Mockito.mock(RoutineLoadTaskScheduler.class));
+
Mockito.when(transactionMgr.getCallbackFactory()).thenReturn(callbackFactory);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ envMock.close();
+ ConnectContext.remove();
+ if (previousContext != null) {
+ previousContext.setThreadLocalInfo();
+ }
+ }
+
+ @Test
+ public void testCreateJournalBeforePublishingJob() throws Exception {
+ RoutineLoadJob job = createJob();
+ Mockito.doAnswer(invocation -> {
+ // Exercise scheduling at the old race window, before the create
record is serialized.
+ for (RoutineLoadJob visibleJob :
manager.getRoutineLoadJobByState(EnumSet.of(JobState.NEED_SCHEDULE))) {
+ setSourceProgress(visibleJob);
+ visibleJob.divideRoutineLoadJob(1);
+ }
+ byte[] record = serialize(invocation.getArgument(0));
+ Assertions.assertEquals(JobState.NEED_SCHEDULE.name(),
serializedState(record));
+ Assertions.assertNull(manager.getJob(job.getId()));
+ Assertions.assertNull(callbackFactory.getCallback(job.getId()));
+ Assertions.assertEquals(0, job.getSizeOfRoutineLoadTaskInfoList());
+ return null;
+ }).when(editLog).logCreateRoutineLoadJob(job);
+
+ manager.addRoutineLoadJob(job, "db", "tbl");
+
+ Mockito.verify(editLog).logCreateRoutineLoadJob(job);
+ Assertions.assertSame(job, manager.getJob(job.getId()));
+ Assertions.assertSame(job, callbackFactory.getCallback(job.getId()));
+ setSourceProgress(job);
+ job.divideRoutineLoadJob(1);
+ Assertions.assertEquals(JobState.RUNNING, job.getState());
+ Assertions.assertEquals(1, job.getSizeOfRoutineLoadTaskInfoList());
+ }
+
+ private RoutineLoadJob createJob() {
+ return new KafkaRoutineLoadJob(1L, "job", 1L, 1L, "127.0.0.1:9092",
"topic", UserIdentity.ADMIN);
+ }
+
+ private void setSourceProgress(RoutineLoadJob job) {
+ job.progress = new KafkaProgress(Collections.singletonMap(0, 100L));
+ Deencapsulation.setField(job, "currentKafkaPartitions",
Lists.newArrayList(0));
+ }
+
+ private byte[] serialize(Writable value) throws IOException {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (DataOutputStream out = new DataOutputStream(bytes)) {
+ value.write(out);
+ }
+ return bytes.toByteArray();
+ }
+
+ private String serializedState(byte[] record) throws IOException {
+ try (DataInputStream in = new DataInputStream(new
ByteArrayInputStream(record))) {
+ return
JsonParser.parseString(Text.readString(in)).getAsJsonObject().get("st").getAsString();
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
