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 1ff9059a46c [fix](routine-load) Persist and replay cancel reason, and
fix cleanup stall from corrupted final jobs (#66173)
1ff9059a46c is described below
commit 1ff9059a46c688d4bcaadbc7458497aedb58b54a
Author: re20052 <[email protected]>
AuthorDate: Mon Sep 21 16:21:29 2026 +0800
[fix](routine-load) Persist and replay cancel reason, and fix cleanup stall
from corrupted final jobs (#66173)
### What problem does this PR solve?
Problem Summary:
There are several related issues around how a routine load job's
state-change reason
is persisted and how a corrupted job is cleaned up:
1. **`pauseReason` / `cancelReason` are not persisted in the Image.**
Both fields lack the
`@SerializedName` annotation, so `GsonUtils.GSON` (which applies
`HiddenAnnotationExclusionStrategy`) strips them during Image
serialization. After an FE
restart, `SHOW ALL ROUTINE LOAD` shows an empty `ReasonOfStateChanged`
for any job that was
`PAUSED` or `CANCELLED`, hiding the real failure cause. This aligns with
how
`LoadJob#failMsg` is already persisted (`@SerializedName("fm")`).
2. **The incremental EditLog path drops the reason for CANCELLED.** In
`unprotectUpdateState`,
only `PAUSED` logged a `RoutineLoadOperation` carrying the reason;
`CANCELLED` logged one
without it. On follower replay / failover before the next checkpoint,
the cancel reason was
lost even though the Image annotation above is now present.
3. **Replay parsing failure produced a corrupted final job.** When
`gsonPostProcess` fails to
parse the original SQL (e.g. the table was dropped, syntax
incompatibility after upgrade),
the job was force-transitioned to `CANCELLED` without setting
`endTimestamp`, leaving it at
the default `-1`. `isExpired()` then hits
`Preconditions.checkState(endTimestamp != -1)` and
throws, which breaks the cleanup loop and effectively leaks these jobs
in memory (they are
never recycled).
This PR:
- Adds `@SerializedName("pauseReason")` /
`@SerializedName("cancelReason")` so both reasons
survive FE restarts, consistent with BrokerLoad's `FailMsg`.
- In `unprotectUpdateState`, also carries the reason in the
`RoutineLoadOperation` for
`CANCELLED` (not just `PAUSED`), so followers/replay keep it.
- In the `gsonPostProcess` catch, terminalizes the unusable job as
`CANCELLED`, sets
`endTimestamp` only when it was still unset (so repeated image loads
before the state is
checkpointed don't keep refreshing it and postponing expiration), and
fills `cancelReason`
only when it was previously null (avoid overwriting a real historical
reason) with a message
like `FE restart deserialize failed at <time>: <exception>`.
- Makes `isExpired()` tolerate a missing `endTimestamp`: instead of
throwing and stalling the
cleanup thread, it logs a warning and treats the corrupted final job as
expired so it can be
recycled. This also drains any such jobs already accumulated on running
clusters.
- Uses `Optional#orElseThrow` on `getDb(dbId)` to surface a clear
`Database <id> does not exist`
message instead of a bare `NoSuchElementException: No value present`.
---
.../doris/load/routineload/RoutineLoadJob.java | 32 ++++++--
.../doris/load/routineload/RoutineLoadJobTest.java | 85 ++++++++++++++++++++++
2 files changed, 112 insertions(+), 5 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
index 9873368f405..4cd4434a73e 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
@@ -242,7 +242,9 @@ public abstract class RoutineLoadJob
protected long autoResumeCount;
// some other msg which need to show to user;
protected String otherMsg = "";
+ @SerializedName("pauseReason")
protected ErrorReason pauseReason;
+ @SerializedName("cancelReason")
protected ErrorReason cancelReason;
@SerializedName("cts")
@@ -1513,7 +1515,7 @@ public abstract class RoutineLoadJob
}
if (!isReplay && jobState != JobState.RUNNING) {
- if (jobState == JobState.PAUSED) {
+ if (jobState == JobState.PAUSED || jobState == JobState.CANCELLED)
{
Env.getCurrentEnv().getEditLog().logOpRoutineLoadJob(new
RoutineLoadOperation(id, jobState, reason));
} else {
Env.getCurrentEnv().getEditLog().logOpRoutineLoadJob(new
RoutineLoadOperation(id, jobState));
@@ -1943,8 +1945,15 @@ public abstract class RoutineLoadJob
if (!isFinal()) {
return false;
}
- Preconditions.checkState(endTimestamp != -1, endTimestamp);
- return (System.currentTimeMillis() - endTimestamp) >
Config.label_keep_max_second * 1000;
+ try {
+ Preconditions.checkState(endTimestamp != -1, endTimestamp);
+ return (System.currentTimeMillis() - endTimestamp) >
Config.label_keep_max_second * 1000;
+ } catch (Exception e) {
+ LOG.warn("routine load job {} is in final state {} but has no
endTimestamp, "
+ + "skip expiring it this round (may race with an
in-progress cancel/stop).",
+ id, state, e);
+ return false;
+ }
}
public boolean isFinal() {
@@ -2000,7 +2009,8 @@ public abstract class RoutineLoadJob
});
try {
ConnectContext ctx = new ConnectContext();
-
ctx.setDatabase(Env.getCurrentEnv().getInternalCatalog().getDb(dbId).get().getName());
+
ctx.setDatabase(Env.getCurrentEnv().getInternalCatalog().getDb(dbId)
+ .orElseThrow(() -> new Exception("Database " + dbId + "
does not exist")).getName());
StatementContext statementContext = new StatementContext();
statementContext.setConnectContext(ctx);
ctx.setStatementContext(statementContext);
@@ -2032,7 +2042,19 @@ public abstract class RoutineLoadJob
ctx.cleanup();
}
} catch (Exception e) {
- this.state = JobState.CANCELLED;
+ // Terminalize this unusable job as CANCELLED. Set endTimestamp
only if unset (avoid
+ // refreshing on every image load) and keep the existing cancel
reason if present.
+ state = JobState.CANCELLED;
+ routineLoadTaskInfoList.clear();
+ long failureTimestamp = System.currentTimeMillis();
+ if (endTimestamp == -1) {
+ endTimestamp = failureTimestamp;
+ }
+ if (cancelReason == null) {
+ cancelReason = new ErrorReason(InternalErrorCode.INTERNAL_ERR,
+ "FE restart deserialize failed at " +
TimeUtils.longToTimeString(failureTimestamp)
+ + ": " + e.getMessage());
+ }
LOG.warn("error happens when parsing create routine load stmt: " +
origStmt.originStmt, e);
}
if (userIdentity != null) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
index 1ce644f8ab4..e1f25b14dd0 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
@@ -34,6 +34,8 @@ import
org.apache.doris.load.routineload.kafka.KafkaRoutineLoadJob;
import org.apache.doris.load.routineload.kafka.KafkaTaskInfo;
import
org.apache.doris.nereids.trees.plans.commands.info.CreateRoutineLoadInfo;
import org.apache.doris.persist.EditLog;
+import org.apache.doris.persist.RoutineLoadOperation;
+import org.apache.doris.qe.OriginStatement;
import org.apache.doris.thrift.TKafkaRLTaskProgress;
import org.apache.doris.thrift.TLoadSourceType;
import org.apache.doris.thrift.TRLTaskTxnCommitAttachment;
@@ -50,6 +52,7 @@ import com.google.common.collect.Maps;
import org.apache.kafka.common.PartitionInfo;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
@@ -499,4 +502,86 @@ public class RoutineLoadJobTest {
Assertions.assertFalse(isPartialUpdate);
}
+ // When a job is cancelled, the CANCELLED operation persisted to edit log
must carry the error reason,
+ // so that followers replaying the log (and a promoted new master) keep
the cancel reason.
+ @Test
+ public void testCancelledOperationCarriesReason() throws UserException {
+ Env env = Mockito.mock(Env.class);
+ InternalCatalog catalog = Mockito.mock(InternalCatalog.class);
+ EditLog editLog = Mockito.mock(EditLog.class);
+ GlobalTransactionMgrIface globalTxnMgr =
Mockito.mock(GlobalTransactionMgrIface.class);
+ TxnStateCallbackFactory callbackFactory =
Mockito.mock(TxnStateCallbackFactory.class);
+
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+ envStatic.when(Env::getCurrentEnv).thenReturn(env);
+ envStatic.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+
envStatic.when(Env::getCurrentGlobalTransactionMgr).thenReturn(globalTxnMgr);
+ Mockito.when(env.getInternalCatalog()).thenReturn(catalog);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+
Mockito.when(globalTxnMgr.getCallbackFactory()).thenReturn(callbackFactory);
+ // db has been deleted, update() will cancel the job with a DB_ERR
reason
+
Mockito.doReturn(null).when(catalog).getDbNullable(Mockito.anyLong());
+
+ RoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob();
+ routineLoadJob.update();
+
+ Assertions.assertEquals(RoutineLoadJob.JobState.CANCELLED,
routineLoadJob.getState());
+
+ ArgumentCaptor<RoutineLoadOperation> captor =
ArgumentCaptor.forClass(RoutineLoadOperation.class);
+ Mockito.verify(editLog).logOpRoutineLoadJob(captor.capture());
+ RoutineLoadOperation operation = captor.getValue();
+ Assertions.assertEquals(RoutineLoadJob.JobState.CANCELLED,
operation.getJobState());
+ Assertions.assertNotNull(operation.getErrorReason(),
+ "cancel reason must be carried in the edit log operation");
+ }
+ }
+
+ // On deserialize failure in gsonPostProcess, an active job should be
terminalized via executeCancel:
+ // state becomes CANCELLED, endTimestamp gets set (not -1) and a cancel
reason is recorded.
+ @Test
+ public void testGsonPostProcessCancelOnDeserializeFailure() throws
Exception {
+ Env env = Mockito.mock(Env.class);
+ InternalCatalog catalog = Mockito.mock(InternalCatalog.class);
+
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+ envStatic.when(Env::getCurrentEnv).thenReturn(env);
+ Mockito.when(env.getInternalCatalog()).thenReturn(catalog);
+ // return empty so the orElseThrow inside gsonPostProcess fires
and drops into the catch block
+
Mockito.doReturn(java.util.Optional.empty()).when(catalog).getDb(Mockito.anyLong());
+
+ RoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob();
+ Deencapsulation.setField(routineLoadJob, "state",
RoutineLoadJob.JobState.RUNNING);
+ // a non-parsable origin statement guarantees the parsing fails
and falls into the catch block,
+ // and also keeps origStmt non-null so the catch block's logging
does not NPE
+ Deencapsulation.setField(routineLoadJob, "origStmt", new
OriginStatement("invalid stmt", 0));
+ routineLoadJob.gsonPostProcess();
+
+ Assertions.assertEquals(RoutineLoadJob.JobState.CANCELLED,
routineLoadJob.getState());
+ Assertions.assertTrue(routineLoadJob.getEndTimestamp() != -1,
+ "endTimestamp must be set when terminalizing the job");
+ Assertions.assertNotNull(Deencapsulation.getField(routineLoadJob,
"cancelReason"));
+ }
+ }
+
+ // A final job whose endTimestamp is not yet set (e.g. cleanup racing with
an in-progress
+ // cancel/stop) must not be expired this round, and must not throw and
break cleanup.
+ @Test
+ public void testIsExpiredSkipsWhenEndTimestampMissing() {
+ RoutineLoadJob missingEndTs = new KafkaRoutineLoadJob();
+ Deencapsulation.setField(missingEndTs, "state",
RoutineLoadJob.JobState.CANCELLED);
+ // endTimestamp keeps its default value -1
+ Assertions.assertFalse(missingEndTs.isExpired());
+
+ // a non-final job is never expired
+ RoutineLoadJob running = new KafkaRoutineLoadJob();
+ Deencapsulation.setField(running, "state",
RoutineLoadJob.JobState.RUNNING);
+ Assertions.assertFalse(running.isExpired());
+
+ // a final job that ended long ago (epoch) is expired via the normal
path
+ RoutineLoadJob oldJob = new KafkaRoutineLoadJob();
+ Deencapsulation.setField(oldJob, "state",
RoutineLoadJob.JobState.CANCELLED);
+ Deencapsulation.setField(oldJob, "endTimestamp", 0L);
+ Assertions.assertTrue(oldJob.isExpired());
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
