Skip to content

Fix CQ recovery gap and stale callback contamination #17734

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jt2594838 merged 4 commits into from
Jun 2, 2026
Merged

Fix CQ recovery gap and stale callback contamination #17734

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
09c0a97
fix
Caideyipi May 21, 2026
6f7a337
sp
Caideyipi May 21, 2026
f09c3b0
Fix CQ local schedule cancellation
Caideyipi Jun 1, 2026
8219e4b
Merge branch 'master' of https://github.com/apache/iotdb into cq-fix
Caideyipi Jun 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 40 additions & 2 deletions iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/cq/CQManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,10 @@

import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

Expand All @@ -57,11 +60,14 @@

private final ReadWriteLock lock;

private final ConcurrentMap<String, String> locallyScheduledCQs;

@jt2594838 jt2594838 Jun 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Explain the key and the value.

@Caideyipi Caideyipi Jun 1, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in f09c3b0: added an inline explanation. The key is CQ id, and the value is the local task plus the metadata token it owns.


private ScheduledExecutorService executor;

public CQManager(ConfigManager configManager) {
this.configManager = configManager;
this.lock = new ReentrantReadWriteLock();
this.locallyScheduledCQs = new ConcurrentHashMap<>();
this.executor =
IoTDBThreadPoolFactory.newScheduledThreadPool(
CONF.getCqSubmitThread(), ThreadName.CQ_SCHEDULER.getName());
Expand All @@ -79,7 +85,11 @@

public TSStatus dropCQ(TDropCQReq req) {
try {
return configManager.getConsensusManager().write(new DropCQPlan(req.cqId));
TSStatus status = configManager.getConsensusManager().write(new DropCQPlan(req.cqId));
if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
locallyScheduledCQs.remove(req.cqId);
}
return status;
Comment on lines -82 to +94

@jt2594838 jt2594838 Jun 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Dropping a CQ does not abort it？

@Caideyipi Caideyipi Jun 1, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in f09c3b0: dropCQ now coordinates with scheduler start/stop and cancels the local CQScheduleTask after the metadata drop succeeds, so queued future executions are aborted.

} catch (ConsensusException e) {
LOGGER.warn(ManagerMessages.UNEXPECTED_ERROR_HAPPENED_WHILE_DROPPING_CQ, req.cqId, e);
// consensus layer related errors
Expand Down Expand Up @@ -113,7 +123,7 @@
return res;
}

public void startCQScheduler() {

Check failure on line 126 in iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/cq/CQManager.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this method to reduce its Cognitive Complexity from 22 to the 15 allowed. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ5InLAGIQnTQqJg_n2S&open=AZ5InLAGIQnTQqJg_n2S&pullRequest=17734
lock.writeLock().lock();
try {
// 1. shutdown previous cq schedule thread pool
Expand All @@ -132,6 +142,7 @@
executor =
IoTDBThreadPoolFactory.newScheduledThreadPool(
CONF.getCqSubmitThread(), ThreadName.CQ_SCHEDULER.getName());
locallyScheduledCQs.clear();

// 3. get all CQs
List<CQInfo.CQEntry> allCQs = null;
Expand All @@ -155,8 +166,16 @@
if (allCQs != null) {
for (CQInfo.CQEntry entry : allCQs) {
if (entry.getState() == CQState.ACTIVE) {
if (!markCQLocallyScheduled(entry.getCqId(), entry.getMd5())) {
continue;
}
CQScheduleTask cqScheduleTask = new CQScheduleTask(entry, executor, configManager);
cqScheduleTask.submitSelf();
try {
cqScheduleTask.submitSelf();
} catch (RuntimeException e) {
unmarkCQLocallyScheduled(entry.getCqId(), entry.getMd5());
throw e;
}

@jt2594838 jt2594838 Jun 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add the task as part of the value in locallyScheduledCQs?
So that you may abort it when dropping.

@Caideyipi Caideyipi Jun 1, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in f09c3b0: locallyScheduledCQs now stores a LocallyScheduledCQ value containing the token and CQScheduleTask. CQScheduleTask owns its latest ScheduledFuture and exposes cancel(), so drop and token replacement can cancel it.

}
}
}
Expand All @@ -176,11 +195,30 @@
try {
previous = executor;
executor = null;
locallyScheduledCQs.clear();
} finally {
lock.writeLock().unlock();
}
if (previous != null) {
previous.shutdown();
}
}

public boolean markCQLocallyScheduled(String cqId, String md5) {

Check warning on line 207 in iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/cq/CQManager.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Abbreviation in name 'markCQLocallyScheduled' must contain no more than '2' consecutive capital letters. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ5InLAGIQnTQqJg_n2T&open=AZ5InLAGIQnTQqJg_n2T&pullRequest=17734
AtomicBoolean shouldSchedule = new AtomicBoolean(false);
locallyScheduledCQs.compute(
cqId,
(ignored, previousMd5) -> {
if (!md5.equals(previousMd5)) {

@luoluoyuyu luoluoyuyu May 27, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When previousMd5 is null, !md5.equals(previousMd5) schedules the CQ as expected. When previousMd5 equals md5, scheduling is skipped, which avoids duplicates.

If the same cqId is updated with a new md5, this returns true and schedules again. Please confirm in this PR that any previously running CQScheduleTask for the old md5 is stopped or superseded; otherwise two tasks could run briefly.

@Caideyipi Caideyipi Jun 1, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in f09c3b0: the local schedule registry now stores the metadata token together with the CQScheduleTask. If the same CQ id is registered with a different token, the old task is cancelled before the new one is scheduled. Drop/start/stop also cancel local tasks, and cancelled callbacks no longer resubmit.

shouldSchedule.set(true);
return md5;
}
return previousMd5;
});
return shouldSchedule.get();
}

public void unmarkCQLocallyScheduled(String cqId, String md5) {

Check warning on line 221 in iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/cq/CQManager.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Abbreviation in name 'unmarkCQLocallyScheduled' must contain no more than '2' consecutive capital letters. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ5InLAGIQnTQqJg_n2U&open=AZ5InLAGIQnTQqJg_n2U&pullRequest=17734
locallyScheduledCQs.remove(cqId, md5);

@luoluoyuyu luoluoyuyu May 27, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Using remove(cqId, md5) ensures we only unmark the entry we own after a failed submitSelf. Good defensive pattern.

}
}
61 changes: 57 additions & 4 deletions ...ignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/cq/CreateCQProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,11 +22,15 @@
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
import org.apache.iotdb.confignode.consensus.request.read.cq.ShowCQPlan;
import org.apache.iotdb.confignode.consensus.request.write.cq.ActiveCQPlan;
import org.apache.iotdb.confignode.consensus.request.write.cq.AddCQPlan;
import org.apache.iotdb.confignode.consensus.request.write.cq.DropCQPlan;
import org.apache.iotdb.confignode.consensus.response.cq.ShowCQResp;
import org.apache.iotdb.confignode.i18n.ProcedureMessages;
import org.apache.iotdb.confignode.manager.cq.CQManager;
import org.apache.iotdb.confignode.manager.cq.CQScheduleTask;
import org.apache.iotdb.confignode.persistence.cq.CQInfo;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.impl.node.AbstractNodeProcedure;
Expand All @@ -45,6 +49,8 @@
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;

import static org.apache.iotdb.confignode.procedure.state.cq.CreateCQState.INACTIVE;
Expand Down Expand Up @@ -75,7 +81,7 @@
public CreateCQProcedure(TCreateCQReq req, ScheduledExecutorService executor) {
super();
this.req = req;
this.md5 = DigestUtils.md2Hex(req.cqId);
this.md5 = generateCQToken(req.cqId);

@jt2594838 jt2594838 Jun 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Improper to call it md5 now.

@Caideyipi Caideyipi Jun 1, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in f09c3b0: renamed the CQ identifier from md5 to cqToken across the CQ plans, metadata, procedure and scheduler code.

this.executor = executor;
this.firstExecutionTime =
CQScheduleTask.getFirstExecutionTime(req.boundaryTime, req.everyInterval);
Expand All @@ -91,12 +97,15 @@
addCQ(env);
return Flow.HAS_MORE_STATE;
case INACTIVE:
CQScheduleTask cqScheduleTask =
new CQScheduleTask(req, firstExecutionTime, md5, executor, env.getConfigManager());
cqScheduleTask.submitSelf();
submitScheduleTask(
env,
new CQScheduleTask(req, firstExecutionTime, md5, executor, env.getConfigManager()));
setNextState(SCHEDULED);
break;
case SCHEDULED:
if (isStateDeserialized()) {
recoverScheduledTask(env);
}
activeCQ(env);
return Flow.NO_MORE_STATE;
default:
Expand Down Expand Up @@ -168,6 +177,42 @@
}
}

void recoverScheduledTask(ConfigNodeProcedureEnv env) throws ConsensusException {
Optional<CQInfo.CQEntry> cqEntry = getCurrentCQEntry(env);
if (!cqEntry.isPresent()) {
LOGGER.info(
"Skip recovering the schedule task of CQ {} because its metadata is unavailable.",
req.cqId);
return;
}
submitScheduleTask(env, new CQScheduleTask(cqEntry.get(), executor, env.getConfigManager()));
}

Optional<CQInfo.CQEntry> getCurrentCQEntry(ConfigNodeProcedureEnv env) throws ConsensusException {

Check warning on line 191 in iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/cq/CreateCQProcedure.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Abbreviation in name 'getCurrentCQEntry' must contain no more than '2' consecutive capital letters. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ5InK_lIQnTQqJg_n2Q&open=AZ5InK_lIQnTQqJg_n2Q&pullRequest=17734
ShowCQResp response =
(ShowCQResp) env.getConfigManager().getConsensusManager().read(new ShowCQPlan());
return response.getCqList().stream()
.filter(entry -> req.cqId.equals(entry.getCqId()) && md5.equals(entry.getMd5()))
.findFirst();
}
Comment on lines +191 to +197

@jt2594838 jt2594838 Jun 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it possible to moderate ShowCQPlan to only return the desired CQs instead of filtering at the upper level?

@Caideyipi Caideyipi Jun 1, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in f09c3b0: ShowCQPlan now accepts an optional CQ id, and CQInfo filters by CQ id before returning the response. The recovery path now reads only the target CQ and then checks the token.


private static String generateCQToken(String cqId) {

Check warning on line 199 in iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/cq/CreateCQProcedure.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Abbreviation in name 'generateCQToken' must contain no more than '2' consecutive capital letters. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ5InK_lIQnTQqJg_n2R&open=AZ5InK_lIQnTQqJg_n2R&pullRequest=17734
return DigestUtils.md2Hex(cqId + "-" + UUID.randomUUID());
}

@jt2594838 jt2594838 Jun 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Now that you use UUID, it is no longer necessary to use md5.

@Caideyipi Caideyipi Jun 1, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in f09c3b0: generateCQToken now uses UUID.randomUUID().toString() directly and the DigestUtils/md2Hex dependency was removed from CreateCQProcedure.


private void submitScheduleTask(ConfigNodeProcedureEnv env, CQScheduleTask cqScheduleTask) {
CQManager cqManager = env.getConfigManager().getCQManager();
if (!cqManager.markCQLocallyScheduled(req.cqId, md5)) {
return;
}
try {
cqScheduleTask.submitSelf();
} catch (RuntimeException e) {
cqManager.unmarkCQLocallyScheduled(req.cqId, md5);
throw e;
}
}

@Override
protected void rollbackState(ConfigNodeProcedureEnv env, CreateCQState state)
throws IOException, InterruptedException, ProcedureException {
Expand Down Expand Up @@ -272,4 +317,12 @@
md5,
firstExecutionTime);
}

public String getCqId() {
return req == null ? null : req.getCqId();
}

public String getMd5() {
return md5;
}
}
35 changes: 35 additions & 0 deletions iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/CQInfoTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,11 @@
package org.apache.iotdb.confignode.persistence;

import org.apache.iotdb.confignode.consensus.request.write.cq.AddCQPlan;
import org.apache.iotdb.confignode.consensus.request.write.cq.DropCQPlan;
import org.apache.iotdb.confignode.consensus.request.write.cq.UpdateCQLastExecTimePlan;
import org.apache.iotdb.confignode.persistence.cq.CQInfo;
import org.apache.iotdb.confignode.rpc.thrift.TCreateCQReq;
import org.apache.iotdb.rpc.TSStatusCode;

import org.apache.thrift.TException;
import org.apache.tsfile.external.commons.io.FileUtils;
Expand Down Expand Up @@ -99,4 +102,36 @@ public void testSnapshot() throws TException, IOException {

Assert.assertEquals(cqInfo, actualCQInfo);
}

@Test
public void testOldCallbackCannotTouchRecreatedCQ() throws Exception {
long executionTime = System.currentTimeMillis();
TCreateCQReq req =
new TCreateCQReq(
"testCq3",
1000,
0,
1000,
0,
(byte) 0,
"select s1 into root.backup.d3.s1 from root.sg.d3",
"create cq testCq3 BEGIN select s1 into root.backup.d3.s1 from root.sg.d3 END",
"Asia",
"root");

cqInfo.addCQ(new AddCQPlan(req, "oldMd5", executionTime));
cqInfo.dropCQ(new DropCQPlan("testCq3"));
cqInfo.addCQ(new AddCQPlan(req, "newMd5", executionTime));

Assert.assertEquals(
TSStatusCode.NO_SUCH_CQ.getStatusCode(),
cqInfo.updateCQLastExecutionTime(
new UpdateCQLastExecTimePlan("testCq3", executionTime + 1000, "oldMd5"))
.code);
Assert.assertEquals(
TSStatusCode.SUCCESS_STATUS.getStatusCode(),
cqInfo.updateCQLastExecutionTime(
new UpdateCQLastExecTimePlan("testCq3", executionTime + 1000, "newMd5"))
.code);
}
}
26 changes: 26 additions & 0 deletions ...gnode/src/test/java/org/apache/iotdb/confignode/procedure/impl/CreateCQProcedureTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,10 +36,36 @@
import java.util.concurrent.ScheduledExecutorService;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.fail;

public class CreateCQProcedureTest {

@Test
public void tokenShouldBeUniqueForSameCQId() {
ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
try {
TCreateCQReq req =
new TCreateCQReq(
"testCq1",
1000,
0,
1000,
0,
(byte) 0,
"select s1 into root.backup.d1(s1) from root.sg.d1",
"create cq testCq1 BEGIN select s1 into root.backup.d1(s1) from root.sg.d1 END",
"Asia",
"root");
CreateCQProcedure createCQProcedure1 = new CreateCQProcedure(req, executor);
CreateCQProcedure createCQProcedure2 = new CreateCQProcedure(req, executor);

assertNotEquals(createCQProcedure1.getMd5(), createCQProcedure2.getMd5());
} finally {
executor.shutdown();
}
}

@Test
public void serializeDeserializeTest() {

Expand Down
Loading
Loading