From 270ab783464d8accb4d9ab816ea5eea076f63019 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 21 Oct 2026 15:44:28 +0800 Subject: [PATCH] [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 (cherry picked from commit 20cab121be9de8a901ec4d34d31123641c8f16da) --- .../it/autocreate/IoTDBPipeDataSinkIT.java | 91 +++++++++++++++++++ .../common/request/IConsensusRequest.java | 21 +++++ .../iotdb/consensus/ratis/RatisConsensus.java | 6 ++ .../consensus/ratis/RatisConsensusTest.java | 40 ++++++++ .../node/pipe/PipeOperateSchemaQueueNode.java | 7 ++ .../pipe/PipeOperateSchemaQueueNodeTest.java | 43 +++++++++ 6 files changed, 208 insertions(+) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/pipe/PipeOperateSchemaQueueNodeTest.java diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeDataSinkIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeDataSinkIT.java index 86e534810745b..ed692c47e22c2 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeDataSinkIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeDataSinkIT.java @@ -39,15 +39,18 @@ 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({MultiClusterIT2AutoCreateSchema.class}) @@ -566,4 +569,92 @@ public void testSpecialPartialInsert() throws Exception { "1635232151960,null,null,2.0,2.1,null,", "1635232143960,6.0,4.0,null,null,null,"))); } + + @Test + public void testTransferMods() { + TestUtils.executeNonQueries( + senderEnv, + Arrays.asList( + "create database root.sg_nonAligned", + "create TIMESERIES root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s0 with datatype=boolean, encoding=RLE,compressor=snappy", + "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s1 with datatype=int32, encoding=PLAIN,compressor=LZ4", + "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s2 with datatype=int64,encoding=gorilla,compressor=uncompressed", + "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s3 with datatype=float,encoding=chimp,compressor=gzip", + "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s4 with datatype=double,encoding=ts_2diff,compressor=zstd", + "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s5 with datatype=text,encoding=dictionary,compressor=lzma2", + "insert into root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`(time,s1, s2,s3,s4,s0,s5) values(1,1,10,5.39,5.51234,true,''),(11,null,20,5.39,15.51234,false,'第2条 device_nonAligned'),(21,3,null,5.39,25.51234,true,'第3条device_nonAligned'),(31,4,40,null,35.51234,true,'第4条device_nonAligned'),(41,5,50,5.39,null,false,'第5条device_nonAligned'),(51,6,60,5.39,55.51234,null,'第6条device_nonAligned'),(61,7,70,5.39,65.51234,false,null),(71,8,80,5.39,75.51234,false,'第8条device_nonAligned'),(81,9,90,5.39,85.51234,false,'第9条device_nonAligned'),(91,10,100,5.39,95.51234,false,'第10条device_nonAligned'),(101,11,110,5.39,105.51234,false,'第11条device_nonAligned'),(111,12,120,5.39,115.51234,false,'第12条device_nonAligned'),(121,13,130,5.39,125.51234,false,'第13条device_nonAligned'),(131,14,140,5.39,135.51234,false,'第14条device_nonAligned'),(141,15,150,5.39,145.51234,false,'第15条'),(151,16,160,5.39,155.51234,false,'第16条'),(161,17,170,5.39,165.51234,false,'第17条'),(171,18,180,5.39,175.51234,false,'第18条'),(181,19,190,5.39,185.51234,false,'第19条'),(191,20,200,5.39,195.51234,false,'第20条'),(201,21,210,5.39,null,false,'第21条')", + "flush", + "delete timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.s0", + String.format( + "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和压缩方式.*", + "count(timeseries),", + Collections.singleton("5,")); + + TestUtils.executeNonQueries( + senderEnv, + Arrays.asList("drop pipe if exists test_history", "drop pipe if exists test_realtime")); + + TestUtils.executeNonQuery(receiverEnv, "drop database root.**"); + + TestUtils.executeNonQueries( + senderEnv, + Arrays.asList( + "delete timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.s1", + String.format( + "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和压缩方式.*", + "count(timeseries),", + Collections.singleton("4,"), + 15); + TestUtils.assertDataAlwaysOnEnv( + receiverEnv, + "count timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.*", + "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/common/request/IConsensusRequest.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/common/request/IConsensusRequest.java index cae54c2bcbcbc..933a9cb0b0ce3 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/common/request/IConsensusRequest.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/common/request/IConsensusRequest.java @@ -41,6 +41,27 @@ default long getMemorySize() { 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. + * + *

A deferred request keeps just its source objects while it waits in the replication queues, + * so it must be able to rebuild the very same bytes on demand and must stay small. Consensus + * implementations may postpone {@link #serializeToByteBuffer()} for such requests and keep the + * source objects alive until then. + */ + default boolean isSerializationDeferred() { + return false; + } + default void markAsGeneratedByRemoteConsensusLeader() { // do nothing by default } 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 20c3092243249..9f628ec193b1c 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 @@ -366,6 +366,12 @@ && waitUntilLeaderReady(raftGroupId)) { } } + // 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 842ea78af7234..875b14abece15 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 @@ -23,12 +23,14 @@ import org.apache.iotdb.consensus.IStateMachine; import org.apache.iotdb.consensus.common.ConsensusGroup; import org.apache.iotdb.consensus.common.Peer; +import org.apache.iotdb.consensus.common.request.IConsensusRequest; import org.apache.iotdb.consensus.config.RatisConfig; import org.apache.iotdb.consensus.exception.ConsensusException; 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 void basicConsensus3Copy() throws Exception { 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 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 5674f264e229e..33c1f3095bb3d 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 @@ -53,6 +53,13 @@ public boolean isOpen() { 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 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 0000000000000..491c46fabddf6 --- /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()); + } +}