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
commit a6542f220c2b1c1be5a4e10100321c1751a8dbc1 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]
