This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 20cab121be9 [Fix] Prevent stale schema queue closes after leader
transfer (#18948)
20cab121be9 is described below
commit 20cab121be9de8a901ec4d34d31123641c8f16da
Author: Caideyipi <[email protected]>
AuthorDate: Wed Oct 21 15:44:28 2026 +0800
[Fix] Prevent stale schema queue closes after leader transfer (#18948)
* [Fix] Prevent stale schema queue closes from following leader transfers
* [Pipe] Wait for historical transfer before asserting mods
---
.../treemodel/auto/basic/IoTDBPipeDataSinkIT.java | 39 ++++++++++++++++++++
.../iotdb/consensus/ratis/RatisConsensus.java | 6 +++
.../iotdb/consensus/ratis/RatisConsensusTest.java | 40 ++++++++++++++++++++
.../plan/node/pipe/PipeOperateSchemaQueueNode.java | 7 ++++
.../node/pipe/PipeOperateSchemaQueueNodeTest.java | 43 ++++++++++++++++++++++
.../iotdb/commons/request/IConsensusRequest.java | 8 ++++
6 files changed, 143 insertions(+)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java
index 03835e7743b..30cee019f66 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java
@@ -41,15 +41,18 @@ import org.junit.runner.RunWith;
import java.nio.charset.Charset;
import java.sql.Connection;
+import java.sql.ResultSet;
import java.sql.Statement;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import static org.apache.iotdb.db.it.IoTDBRestServiceIT.getHttpPost;
+import static org.awaitility.Awaitility.await;
@RunWith(IoTDBTestRunner.class)
@Category({MultiClusterIT2DualTreeAutoBasic.class})
@@ -582,6 +585,7 @@ public class IoTDBPipeDataSinkIT extends
AbstractPipeDualTreeModelAutoIT {
"create pipe test with source
('source.realtime.mode'='stream','inclusion'='data','path'='root.sg_nonAligned.非对齐序列带有encoding和压缩方式.**','source.realtime.enable'='true','mods.enable'='true')
with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='%s')",
receiverEnv.getDataNodeWrapperList().get(0).getIpAndPortString())));
+ awaitHistoricalPipeCompletion("test");
TestUtils.assertDataEventuallyOnEnv(
receiverEnv,
"count timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.*",
@@ -602,6 +606,7 @@ public class IoTDBPipeDataSinkIT extends
AbstractPipeDualTreeModelAutoIT {
"create pipe test with source
('source.realtime.mode'='stream','inclusion'='data','path'='root.sg_nonAligned.非对齐序列带有encoding和压缩方式.**','source.realtime.enable'='true','mods.enable'='true')
with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='%s')",
receiverEnv.getDataNodeWrapperList().get(0).getIpAndPortString())));
+ awaitHistoricalPipeCompletion("test");
TestUtils.assertDataEventuallyOnEnv(
receiverEnv,
"count timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.*",
@@ -614,4 +619,38 @@ public class IoTDBPipeDataSinkIT extends
AbstractPipeDualTreeModelAutoIT {
"count(timeseries),",
Collections.singleton("4,"));
}
+
+ private void awaitHistoricalPipeCompletion(final String pipeName) {
+ await()
+ .pollInSameThread()
+ .pollDelay(1L, TimeUnit.SECONDS)
+ .pollInterval(1L, TimeUnit.SECONDS)
+ .atMost(600L, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ boolean sourcePipeFound = false;
+ boolean sourcePipeCompleted = true;
+ boolean realtimePipeFound = false;
+
+ for (final DataNodeWrapper dataNode :
senderEnv.getDataNodeWrapperList()) {
+ try (final Connection connection =
senderEnv.getConnection(dataNode);
+ final Statement statement = connection.createStatement();
+ final ResultSet resultSet = statement.executeQuery("show
pipes")) {
+ while (resultSet.next()) {
+ final String id = resultSet.getString(1);
+ if (id.equals(pipeName) || id.equals(pipeName +
"_history")) {
+ sourcePipeFound = true;
+ sourcePipeCompleted &=
"0".equals(resultSet.getString(8));
+ } else if (id.equals(pipeName + "_realtime")) {
+ realtimePipeFound = true;
+ }
+ }
+ }
+ }
+
+ // The historical sub-pipe may be auto-dropped after all events
are committed.
+ Assert.assertTrue(sourcePipeFound || realtimePipeFound);
+ Assert.assertTrue(!sourcePipeFound || sourcePipeCompleted);
+ });
+ }
}
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
index b7369546e7c..c2028cfd0da 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
@@ -373,6 +373,12 @@ class RatisConsensus implements IConsensus {
}
}
+ // A leader-local decision may already be stale after a leadership change.
Let the current
+ // leader make its own decision instead of replaying the old leader's
request there.
+ if (!request.isForwardable()) {
+ return StatusUtils.getStatus(TSStatusCode.REDIRECTION_RECOMMEND);
+ }
+
// 2. try raft client
TSStatus writeResult;
try (AutoCloseable ignored =
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
index 842ea78af72..3af937ee4e4 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.consensus.ratis;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
+import org.apache.iotdb.commons.request.IConsensusRequest;
import org.apache.iotdb.consensus.IStateMachine;
import org.apache.iotdb.consensus.common.ConsensusGroup;
import org.apache.iotdb.consensus.common.Peer;
@@ -29,6 +30,7 @@ import
org.apache.iotdb.consensus.exception.ConsensusGroupAlreadyExistException;
import org.apache.iotdb.consensus.exception.ConsensusGroupNotExistException;
import
org.apache.iotdb.consensus.exception.PeerAlreadyInConsensusGroupException;
import org.apache.iotdb.consensus.exception.PeerNotInConsensusGroupException;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.ratis.util.TimeDuration;
import org.junit.After;
@@ -113,6 +115,44 @@ public class RatisConsensusTest {
doConsensus(0, 10, 10);
}
+ @Test
+ public void testNonForwardableWriteAfterLeaderTransfer() throws Exception {
+ for (RatisConsensus server : servers) {
+ server.createLocalPeer(gid, peers);
+ }
+ miniCluster.waitUntilActiveLeaderElectedAndReady();
+
+ final int oldLeaderId = servers.get(0).getLeader(gid).getNodeId();
+ final int newLeaderId = (oldLeaderId + 1) % servers.size();
+ final RatisConsensus oldLeader = servers.get(oldLeaderId);
+ final IConsensusRequest leaderLocalRequest =
+ new
TestUtils.TestRequest(TestUtils.TestRequest.incrRequest().serializeToByteBuffer())
{
+ @Override
+ public boolean isForwardable() {
+ return false;
+ }
+ };
+
+ oldLeader.transferLeader(gid, peers.get(newLeaderId));
+ miniCluster.waitUntilActiveLeaderElectedAndReady();
+ Assert.assertEquals(newLeaderId, oldLeader.getLeader(gid).getNodeId());
+
+ // The new leader has already made its own decision before the old request
arrives.
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ servers.get(newLeaderId).write(gid, leaderLocalRequest).getCode());
+ Assert.assertEquals(
+ TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode(),
+ oldLeader.write(gid, leaderLocalRequest).getCode());
+ Assert.assertEquals(1, miniCluster.mustRead(newLeaderId));
+
+ // Ordinary writes issued to a follower must still reach the current
leader.
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ oldLeader.write(gid, TestUtils.TestRequest.incrRequest()).getCode());
+ Assert.assertEquals(2, miniCluster.mustRead(newLeaderId));
+ }
+
@Test
public void addMemberToGroup() throws Exception {
List<Peer> original = peers.subList(0, 1);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/pipe/PipeOperateSchemaQueueNode.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/pipe/PipeOperateSchemaQueueNode.java
index 14512835068..3a24635ee5d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/pipe/PipeOperateSchemaQueueNode.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/pipe/PipeOperateSchemaQueueNode.java
@@ -54,6 +54,13 @@ public class PipeOperateSchemaQueueNode extends PlanNode {
return isOpen;
}
+ @Override
+ public boolean isForwardable() {
+ // A close is decided from the local pipe metadata. After a leader change
the new leader may
+ // already have newer metadata and an active source, so an old close must
not be forwarded.
+ return isOpen;
+ }
+
@Override
public List<PlanNode> getChildren() {
return Collections.emptyList();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/pipe/PipeOperateSchemaQueueNodeTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/pipe/PipeOperateSchemaQueueNodeTest.java
new file mode 100644
index 00000000000..491c46fabdd
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/pipe/PipeOperateSchemaQueueNodeTest.java
@@ -0,0 +1,43 @@
+/*
+ * 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.iotdb.db.queryengine.plan.planner.plan.node.pipe;
+
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import
org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class PipeOperateSchemaQueueNodeTest {
+
+ @Test
+ public void testCloseCannotBeForwardedToNewLeader() {
+ final PipeOperateSchemaQueueNode close =
+ new PipeOperateSchemaQueueNode(new PlanNodeId("close"), false);
+ Assert.assertFalse(close.isForwardable());
+ Assert.assertFalse(close.clone().isForwardable());
+
Assert.assertFalse(PlanNodeType.deserialize(close.serializeToByteBuffer()).isForwardable());
+
+ final PipeOperateSchemaQueueNode open =
+ new PipeOperateSchemaQueueNode(new PlanNodeId("open"), true);
+ Assert.assertTrue(open.isForwardable());
+
Assert.assertTrue(PlanNodeType.deserialize(open.serializeToByteBuffer()).isForwardable());
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/request/IConsensusRequest.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/request/IConsensusRequest.java
index 204b5dd75a9..490c88f2b3f 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/request/IConsensusRequest.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/request/IConsensusRequest.java
@@ -41,6 +41,14 @@ public interface IConsensusRequest {
return 0;
}
+ /**
+ * Whether a consensus implementation may forward this write to another
leader. Requests based on
+ * the local leader's runtime state must not be forwarded after that leader
steps down.
+ */
+ default boolean isForwardable() {
+ return true;
+ }
+
/**
* Returns whether this request materializes its bytes only when it is about
to be sent, instead
* of when it is handed over to the replication layer.
