This is an automated email from the ASF dual-hosted git repository.
hello-stephen 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 d1bcea2962c [fix](fe) Stabilize RollupJobV2Test task execution (#65204)
d1bcea2962c is described below
commit d1bcea2962cd9c7d144e010349339c665d2e37e2
Author: shuke <[email protected]>
AuthorDate: Mon Jul 6 14:33:06 2026 +0800
[fix](fe) Stabilize RollupJobV2Test task execution (#65204)
Problem Summary:
`RollupJobV2Test.testSchemaChange1` manually drives a rollup job through
`WAITING_TXN`, `RUNNING`, and `FINISHED`. The test also allowed
`AgentTaskExecutor.submit()` to schedule generated alter tasks
asynchronously. In FE UT, that background executor can observe fake
backend/system-info state and mark tasks failed, which can cancel the
rollup job before the manually driven success path finishes.
The flaky CI symptom was `expected:<RUNNING> but was:<CANCELLED>`. This
happens when the async `AgentBatchTask` marks generated alter tasks
failed with `backend ... is not alive`; the next catalog-ready cycle
then sees the rollup task failure threshold and cancels the rollup job.
This PR mocks `AgentTaskExecutor.submit()` in the fixture so the
test-owned state transitions remain deterministic. It also adds
`testSchemaChangeCancelWhenRollupTasksFailed` to keep explicit coverage
for the cancellation branch: two rollup tasks on the same tablet are
marked failed, and the job is asserted to become `CANCELLED`.
---
.../org/apache/doris/alter/RollupJobV2Test.java | 58 ++++++++++++++++++++++
1 file changed, 58 insertions(+)
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/alter/RollupJobV2Test.java
b/fe/fe-core/src/test/java/org/apache/doris/alter/RollupJobV2Test.java
index 56b9b9f2ce6..f46afdc56fd 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/alter/RollupJobV2Test.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/alter/RollupJobV2Test.java
@@ -47,7 +47,9 @@ import
org.apache.doris.nereids.trees.plans.commands.info.AddRollupOp;
import org.apache.doris.nereids.trees.plans.commands.info.AlterOp;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.OriginStatement;
+import org.apache.doris.task.AgentBatchTask;
import org.apache.doris.task.AgentTask;
+import org.apache.doris.task.AgentTaskExecutor;
import org.apache.doris.task.AgentTaskQueue;
import org.apache.doris.thrift.TStorageFormat;
import org.apache.doris.thrift.TTaskType;
@@ -59,6 +61,8 @@ import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
import java.io.DataInputStream;
import java.io.DataOutputStream;
@@ -86,6 +90,7 @@ public class RollupJobV2Test {
private FakeEnv fakeEnv;
private FakeEditLog fakeEditLog;
+ private MockedStatic<AgentTaskExecutor> agentTaskExecutorMock;
@Before
public void setUp() throws InstantiationException, IllegalAccessException,
IllegalArgumentException,
@@ -115,6 +120,9 @@ public class RollupJobV2Test {
AgentTaskQueue.clearAllTasks();
FakeEnv.setEnv(masterEnv);
+ agentTaskExecutorMock = Mockito.mockStatic(AgentTaskExecutor.class);
+ agentTaskExecutorMock.when(() ->
AgentTaskExecutor.submit(Mockito.any(AgentBatchTask.class)))
+ .thenAnswer(invocation -> null);
}
@After
@@ -130,6 +138,9 @@ public class RollupJobV2Test {
if (fakeTransactionIDGenerator != null) {
fakeTransactionIDGenerator.close();
}
+ if (agentTaskExecutorMock != null) {
+ agentTaskExecutorMock.close();
+ }
}
@Test
@@ -269,6 +280,53 @@ public class RollupJobV2Test {
Assert.assertEquals(JobState.FINISHED, rollupJob.getJobState());
}
+ @Test
+ public void testSchemaChangeCancelWhenRollupTasksFailed() throws Exception
{
+ if (fakeEnv != null) {
+ fakeEnv.close();
+ }
+ fakeEnv = new FakeEnv();
+ if (fakeEditLog != null) {
+ fakeEditLog.close();
+ }
+ fakeEditLog = new FakeEditLog();
+ FakeEnv.setEnv(masterEnv);
+ MaterializedViewHandler materializedViewHandler =
Env.getCurrentEnv().getMaterializedViewHandler();
+
+ ArrayList<AlterOp> alterOps = new ArrayList<>();
+ alterOps.add(op);
+ Database db =
masterEnv.getInternalCatalog().getDbOrDdlException(CatalogTestUtil.testDbId1);
+ OlapTable olapTable = (OlapTable)
db.getTableOrDdlException(CatalogTestUtil.testTableId1);
+ materializedViewHandler.process(alterOps, db, olapTable);
+ Map<Long, AlterJobV2> alterJobsV2 =
materializedViewHandler.getAlterJobsV2();
+ Assert.assertEquals(1, alterJobsV2.size());
+ RollupJobV2 rollupJob = (RollupJobV2)
alterJobsV2.values().stream().findAny().get();
+
+ materializedViewHandler.runAfterCatalogReady();
+ Assert.assertEquals(JobState.WAITING_TXN, rollupJob.getJobState());
+
+ materializedViewHandler.runAfterCatalogReady();
+ Assert.assertEquals(JobState.RUNNING, rollupJob.getJobState());
+
+ List<AgentTask> tasks = AgentTaskQueue.getTask(TTaskType.ALTER);
+ Assert.assertEquals(3, tasks.size());
+ long failedTabletId = tasks.get(0).getTabletId();
+ int failedTaskCount = 0;
+ for (AgentTask agentTask : tasks) {
+ if (agentTask.getTabletId() == failedTabletId) {
+ agentTask.failedWithMsg("backend " + agentTask.getBackendId()
+ " is not alive");
+ failedTaskCount++;
+ }
+ if (failedTaskCount == 2) {
+ break;
+ }
+ }
+ Assert.assertEquals(2, failedTaskCount);
+
+ materializedViewHandler.runAfterCatalogReady();
+ Assert.assertEquals(JobState.CANCELLED, rollupJob.getJobState());
+ }
+
@Test
public void testSchemaChangeWhileTabletNotStable() throws Exception {
if (fakeEnv != null) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
