From 98246cec9063c4a2acf2134c848cf3fea847cc4b Mon Sep 17 00:00:00 2001 From: Yongzao <532741407@qq.com> Date: Fri, 12 Jun 2026 20:05:15 +0800 Subject: [PATCH 1/7] Propagate snapshot load failure during IoTConsensus AddPeer During region migration, when the target peer failed to load the transferred snapshot, the failure was silently swallowed: the target's IoTConsensus RPC handler returned SUCCESS regardless, so the coordinator activated the new peer and marked AddRegionPeerProcedure / RegionMigrateProcedure successful. The migration was reported complete while the destination replica actually had no data, leading to silent data loss once the source replica was dropped. The coordinator side already handles a non-SUCCESS triggerSnapshotLoad response correctly (it throws ConsensusGroupModifyPeerException, which fails the AddPeer task and rolls the procedure back without deleting the source replica). The only broken link was that snapshot-load failure was never reportable, because IStateMachine.loadSnapshot returned void and the implementations swallowed errors. Change IStateMachine.loadSnapshot to return boolean (true on success): - DataRegionStateMachine / SchemaRegionStateMachine / ConfigRegionState Machine return false when loading fails (and SchemaRegionStateMachine now guards its body so an exception is reported rather than thrown). - IoTConsensusServerImpl.loadSnapshot returns false if loading any receive folder fails (removing the long-standing TODO). - IoTConsensusRPCServiceProcessor.triggerSnapshotLoad returns a non- SUCCESS status when loadSnapshot fails, so the coordinator's existing error path fires and AddPeer fails instead of falsely succeeding. - SimpleConsensusServerImpl forwards the boolean; the Ratis ApplicationStateMachineProxy logs a failure (its behavior is otherwise unchanged). Test state machines updated accordingly. Add AddPeerSnapshotLoadFailureTest: a real two-node IoTConsensus group where the target's loadSnapshot is forced to fail; it verifies that addRemotePeer reaches the load step, throws ConsensusException, and does not leave the target peer active. The test fails against the old code and passes with the fix. --- .../ConfigRegionStateMachine.java | 4 +- .../apache/iotdb/consensus/IStateMachine.java | 5 +- .../consensus/iot/IoTConsensusServerImpl.java | 19 +- .../IoTConsensusRPCServiceProcessor.java | 10 +- .../ratis/ApplicationStateMachineProxy.java | 4 +- .../simple/SimpleConsensusServerImpl.java | 4 +- .../iotdb/consensus/EmptyStateMachine.java | 4 +- .../iot/AddPeerSnapshotLoadFailureTest.java | 244 ++++++++++++++++++ .../consensus/iot/util/TestStateMachine.java | 4 +- .../iotdb/consensus/ratis/TestUtils.java | 4 +- .../consensus/simple/SimpleConsensusTest.java | 4 +- .../dataregion/DataRegionStateMachine.java | 6 +- .../SchemaRegionStateMachine.java | 22 +- 13 files changed, 306 insertions(+), 28 deletions(-) create mode 100644 iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/AddPeerSnapshotLoadFailureTest.java diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java index b3029e6602808..1c2fb38470b61 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java @@ -246,7 +246,7 @@ public boolean takeSnapshot(File snapshotDir) { } @Override - public void loadSnapshot(final File latestSnapshotRootDir) { + public boolean loadSnapshot(final File latestSnapshotRootDir) { try { executor.loadSnapshot(latestSnapshotRootDir); // We recompute the snapshot for pipe listener when loading snapshot @@ -254,12 +254,14 @@ public void loadSnapshot(final File latestSnapshotRootDir) { PipeConfigNodeAgent.runtime() .listener() .tryListenToSnapshots(ConfigNodeSnapshotParser.getSnapshots()); + return true; } catch (final IOException e) { if (PipeConfigNodeAgent.runtime().listener().isOpened()) { LOGGER.warn( ConfigNodeMessages.CONFIG_REGION_LISTENING_QUEUE_LISTEN_TO_SNAPSHOT_FAILED_WHEN_STARTUP, e); } + return false; } } diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IStateMachine.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IStateMachine.java index 3354c83699b54..c7705d93896e8 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IStateMachine.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IStateMachine.java @@ -112,8 +112,11 @@ default boolean clearSnapshot() { * Load the latest snapshot from given dir. * * @param latestSnapshotRootDir dir where the latest snapshot sits + * @return {@code true} if the snapshot was loaded successfully, {@code false} otherwise. Callers + * (e.g. the IoTConsensus AddPeer flow) rely on this to avoid activating a new peer whose + * snapshot failed to load, which would otherwise silently lose data. */ - void loadSnapshot(File latestSnapshotRootDir); + boolean loadSnapshot(File latestSnapshotRootDir); /** * given a snapshot dir, ask statemachine to provide all snapshot files. By default, it will list diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java index f3c7d4e50ae26..27a2691d63136 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java @@ -511,14 +511,17 @@ private void clearOldSnapshot() { } } - public void loadSnapshot(String snapshotId) { - // TODO: (xingtanzjr) throw exception if the snapshot load failed - recvFolderManager - .getFolders() - .forEach( - dir -> { - stateMachine.loadSnapshot(getSnapshotPath(dir, snapshotId)); - }); + public boolean loadSnapshot(String snapshotId) { + // Load the snapshot from every receive folder. If any of them fails, report the failure so the + // AddPeer coordinator does not activate this peer with incomplete data (which would silently + // lose data on this replica). + boolean success = true; + for (String dir : recvFolderManager.getFolders()) { + if (!stateMachine.loadSnapshot(getSnapshotPath(dir, snapshotId))) { + success = false; + } + } + return success; } private File getSnapshotPath(String curStorageDir, String snapshotRelativePath) { diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java index c9a9901dbcf6f..91e17b370ca1e 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java @@ -343,7 +343,15 @@ public TTriggerSnapshotLoadRes triggerSnapshotLoad(TTriggerSnapshotLoadReq req) status.setMessage(message); return new TTriggerSnapshotLoadRes(status); } - impl.loadSnapshot(req.snapshotId); + if (!impl.loadSnapshot(req.snapshotId)) { + String message = + String.format( + "Failed to load snapshot %s for consensus group %s", req.snapshotId, groupId); + LOGGER.error(message); + TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); + status.setMessage(message); + return new TTriggerSnapshotLoadRes(status); + } KillPoint.setKillPoint(DataNodeKillPoints.DESTINATION_ADD_PEER_TRANSITION); return new TTriggerSnapshotLoadRes(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); } diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java index 1134d8fd6f206..dc009a71887e0 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java @@ -268,7 +268,9 @@ private void loadSnapshot(File latestSnapshotDir) { } // require the application statemachine to load the latest snapshot - applicationStateMachine.loadSnapshot(latestSnapshotDir); + if (!applicationStateMachine.loadSnapshot(latestSnapshotDir)) { + logger.error("{}: failed to load snapshot from {}", this, latestSnapshotDir); + } TermIndex snapshotTermIndex = Utils.getTermIndexFromDir(latestSnapshotDir); updateLastAppliedTermIndex(snapshotTermIndex.getTerm(), snapshotTermIndex.getIndex()); } diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensusServerImpl.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensusServerImpl.java index 4ed4a7b41ce41..b76bcdeaaebbc 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensusServerImpl.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensusServerImpl.java @@ -89,7 +89,7 @@ public synchronized boolean takeSnapshot(File snapshotDir) { } @Override - public synchronized void loadSnapshot(File latestSnapshotRootDir) { - stateMachine.loadSnapshot(latestSnapshotRootDir); + public synchronized boolean loadSnapshot(File latestSnapshotRootDir) { + return stateMachine.loadSnapshot(latestSnapshotRootDir); } } diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/EmptyStateMachine.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/EmptyStateMachine.java index aafa0be5bb5ff..997120b00f96d 100644 --- a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/EmptyStateMachine.java +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/EmptyStateMachine.java @@ -54,5 +54,7 @@ public boolean takeSnapshot(File snapshotDir) { } @Override - public void loadSnapshot(File latestSnapshotRootDir) {} + public boolean loadSnapshot(File latestSnapshotRootDir) { + return true; + } } diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/AddPeerSnapshotLoadFailureTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/AddPeerSnapshotLoadFailureTest.java new file mode 100644 index 0000000000000..aa69907e6ed4e --- /dev/null +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/AddPeerSnapshotLoadFailureTest.java @@ -0,0 +1,244 @@ +/* + * 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.consensus.iot; + +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.consensus.ConsensusGroupId; +import org.apache.iotdb.commons.consensus.DataRegionId; +import org.apache.iotdb.commons.exception.StartupException; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.consensus.common.Peer; +import org.apache.iotdb.consensus.config.ConsensusConfig; +import org.apache.iotdb.consensus.exception.ConsensusException; +import org.apache.iotdb.consensus.iot.util.TestEntry; +import org.apache.iotdb.consensus.iot.util.TestStateMachine; + +import org.apache.ratis.util.FileUtils; +import org.junit.After; +import org.junit.Assert; +import org.junit.Assume; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.net.ServerSocket; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +/** + * Regression test for the snapshot-load-failure bug (a failed snapshot load on the AddPeer target + * was silently swallowed, so the new peer was activated and the migration falsely reported + * successful, losing data on the new replica). + * + *

It builds a real two-node IoTConsensus group, forces the target peer's {@link + * org.apache.iotdb.consensus.IStateMachine#loadSnapshot} to fail, and verifies that {@code + * addRemotePeer}: + * + *

+ */ +public class AddPeerSnapshotLoadFailureTest { + + private final Logger logger = LoggerFactory.getLogger(AddPeerSnapshotLoadFailureTest.class); + + private final ConsensusGroupId gid = new DataRegionId(1); + + private final int basePort = 9200; + + private final List peers = + Arrays.asList( + new Peer(gid, 1, new TEndPoint("127.0.0.1", basePort - 1)), + new Peer(gid, 2, new TEndPoint("127.0.0.1", basePort))); + + private final List peersStorage = + Arrays.asList( + new File("target" + File.separator + "snapshot-load-fail-1"), + new File("target" + File.separator + "snapshot-load-fail-2")); + + private final List> peersRecvSnapshotDirs = + Arrays.asList( + Arrays.asList( + "target" + File.separator + "snapshot-load-fail-1-recv-1", + "target" + File.separator + "snapshot-load-fail-1-recv-2"), + Arrays.asList( + "target" + File.separator + "snapshot-load-fail-2-recv-1", + "target" + File.separator + "snapshot-load-fail-2-recv-2")); + + private final List servers = new ArrayList<>(); + private final List stateMachines = new ArrayList<>(); + + /** A {@link TestStateMachine} whose snapshot load can be made to fail on demand. */ + private static class ControllableStateMachine extends TestStateMachine { + private volatile boolean failLoadSnapshot = false; + private volatile boolean loadSnapshotInvoked = false; + + void setFailLoadSnapshot(boolean failLoadSnapshot) { + this.failLoadSnapshot = failLoadSnapshot; + } + + boolean isLoadSnapshotInvoked() { + return loadSnapshotInvoked; + } + + @Override + public boolean loadSnapshot(File latestSnapshotRootDir) { + loadSnapshotInvoked = true; + if (failLoadSnapshot) { + return false; + } + return super.loadSnapshot(latestSnapshotRootDir); + } + + @Override + public boolean takeSnapshot(File snapshotDir) { + return true; + } + + // TestStateMachine does not implement clearSnapshot (the IStateMachine default throws). The + // AddPeer flow calls it in a finally block to clean up the local snapshot, so we provide a + // no-op here; otherwise that cleanup would mask the ConsensusException we are asserting on. + @Override + public boolean clearSnapshot() { + return true; + } + } + + @Before + public void setUp() throws Exception { + for (File file : peersStorage) { + file.mkdirs(); + stateMachines.add(new ControllableStateMachine()); + } + peersRecvSnapshotDirs.forEach(innerList -> innerList.forEach(dir -> new File(dir).mkdirs())); + initServer(); + } + + @After + public void tearDown() throws Exception { + servers.parallelStream().forEach(IoTConsensus::stop); + servers.clear(); + for (File file : peersStorage) { + FileUtils.deleteFully(file); + } + peersRecvSnapshotDirs.forEach( + innerList -> + innerList.forEach( + dir -> { + try { + FileUtils.deleteFully(new File(dir)); + } catch (IOException e) { + throw new RuntimeException(e); + } + })); + } + + private void initServer() throws IOException { + Assume.assumeTrue(checkPortAvailable()); + try { + for (int i = 0; i < peers.size(); i++) { + int finalI = i; + servers.add( + (IoTConsensus) + ConsensusFactory.getConsensusImpl( + ConsensusFactory.IOT_CONSENSUS, + ConsensusConfig.newBuilder() + .setThisNodeId(peers.get(i).getNodeId()) + .setThisNode(peers.get(i).getEndpoint()) + .setStorageDir(peersStorage.get(i).getAbsolutePath()) + .setRecvSnapshotDirs(peersRecvSnapshotDirs.get(i)) + .setConsensusGroupType(TConsensusGroupType.DataRegion) + .build(), + groupId -> stateMachines.get(finalI)) + .orElseThrow( + () -> + new IllegalArgumentException( + String.format( + ConsensusFactory.CONSTRUCT_FAILED_MSG, + ConsensusFactory.IOT_CONSENSUS)))); + } + for (int i = 0; i < peers.size(); i++) { + servers.get(i).start(); + } + } catch (IOException e) { + if (e.getCause() instanceof StartupException) { + // just succeed when can not bind socket + logger.info("Can not start IoTConsensus because", e); + Assume.assumeTrue(false); + } else { + logger.error("Failed because", e); + Assert.fail("Failed because " + e.getMessage()); + } + } + } + + @Test + public void addRemotePeerMustFailWhenTargetSnapshotLoadFails() throws Exception { + // node 0 is the sole initial member; node 1 will be added as a new peer. Mirroring the real + // region-migration flow, the destination peer (node 1) is pre-created locally with the full + // target peer list (IoTConsensus, unlike Ratis, requires a non-empty peer list here). + servers.get(0).createLocalPeer(gid, peers.subList(0, 1)); + servers.get(1).createLocalPeer(gid, peers); + + // Put some data into the group so the snapshot transfer is meaningful. + for (int i = 0; i < 10; i++) { + servers.get(0).write(gid, new TestEntry(i, peers.get(0))); + } + + // Force the target peer (node 1) to fail loading the transferred snapshot. + stateMachines.get(1).setFailLoadSnapshot(true); + + // Before the fix, addRemotePeer swallowed the load failure and returned normally, leaving the + // target peer active with incomplete data. It must now surface the failure. + Assert.assertThrows( + ConsensusException.class, () -> servers.get(0).addRemotePeer(gid, peers.get(1))); + + // The failure must be the snapshot load itself, i.e. the AddPeer flow actually reached the + // load step on the target rather than aborting earlier. + Assert.assertTrue( + "Target peer's loadSnapshot was never invoked; the failure came from an earlier step", + stateMachines.get(1).isLoadSnapshotInvoked()); + + // The target peer must not be left active with an incompletely-loaded snapshot. + Assert.assertFalse( + "Target peer was activated despite a failed snapshot load", + servers.get(1).getImpl(gid).isActive()); + } + + private boolean checkPortAvailable() { + for (Peer peer : this.peers) { + try (ServerSocket ignored = new ServerSocket(peer.getEndpoint().port)) { + logger.info("check port {} success for node {}", peer.getEndpoint().port, peer.getNodeId()); + } catch (IOException e) { + logger.error("check port {} failed for node {}", peer.getEndpoint().port, peer.getNodeId()); + return false; + } + } + return true; + } +} diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/util/TestStateMachine.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/util/TestStateMachine.java index a879a03478457..9454dcb0c4392 100644 --- a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/util/TestStateMachine.java +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/util/TestStateMachine.java @@ -142,5 +142,7 @@ public boolean takeSnapshot(File snapshotDir) { } @Override - public void loadSnapshot(File latestSnapshotRootDir) {} + public boolean loadSnapshot(File latestSnapshotRootDir) { + return true; + } } diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java index 5cac2173396a3..3217d1cc58e0c 100644 --- a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java @@ -177,13 +177,15 @@ public boolean takeSnapshot(File snapshotDir) { } @Override - public void loadSnapshot(File latestSnapshotRootDir) { + public boolean loadSnapshot(File latestSnapshotRootDir) { File snapshot = new File(latestSnapshotRootDir.getAbsolutePath() + File.separator + "snapshot"); try (Scanner scanner = new Scanner(snapshot)) { integer.set(Integer.parseInt(scanner.next())); + return true; } catch (FileNotFoundException e) { logger.error("cannot find snapshot file {}", snapshot); + return false; } } diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/simple/SimpleConsensusTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/simple/SimpleConsensusTest.java index d50ab992f6f9c..ff062fb63a85b 100644 --- a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/simple/SimpleConsensusTest.java +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/simple/SimpleConsensusTest.java @@ -123,7 +123,9 @@ public boolean takeSnapshot(File snapshotDir) { } @Override - public void loadSnapshot(File latestSnapshotRootDir) {} + public boolean loadSnapshot(File latestSnapshotRootDir) { + return true; + } } @Before diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java index 2de1ec9fdc0ce..62bb498afa19a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java @@ -124,7 +124,7 @@ public boolean clearSnapshot() { } @Override - public void loadSnapshot(File latestSnapshotRootDir) { + public boolean loadSnapshot(File latestSnapshotRootDir) { String databaseName = region.getDatabaseName(); String dataRegionIdString = region.getDataRegionIdString(); DataRegionId regionId = new DataRegionId(Integer.parseInt(dataRegionIdString)); @@ -141,14 +141,16 @@ public void loadSnapshot(File latestSnapshotRootDir) { .loadSnapshotForStateMachine()); if (newRegion == null) { logger.error(DataNodeMiscMessages.FAIL_LOAD_SNAPSHOT, latestSnapshotRootDir); - return; + return false; } this.region = newRegion; ChunkCache.getInstance().clear(); TimeSeriesMetadataCache.getInstance().clear(); BloomFilterCache.getInstance().clear(); + return true; } catch (Exception e) { logger.error(DataNodeMiscMessages.EXCEPTION_REPLACING_DATA_REGION, e); + return false; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java index edab5862d11e8..3f647ced7a5e7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java @@ -136,14 +136,20 @@ public boolean takeSnapshot(final File snapshotDir) { } @Override - public void loadSnapshot(final File latestSnapshotRootDir) { - schemaRegion.loadSnapshot(latestSnapshotRootDir); - PipeDataNodeAgent.runtime() - .schemaListener(schemaRegion.getSchemaRegionId()) - .loadSnapshot(latestSnapshotRootDir); - // We recompute the snapshot for pipe listener when loading snapshot - // to recover the newest snapshot in cache - listen2Snapshot4PipeListener(false); + public boolean loadSnapshot(final File latestSnapshotRootDir) { + try { + schemaRegion.loadSnapshot(latestSnapshotRootDir); + PipeDataNodeAgent.runtime() + .schemaListener(schemaRegion.getSchemaRegionId()) + .loadSnapshot(latestSnapshotRootDir); + // We recompute the snapshot for pipe listener when loading snapshot + // to recover the newest snapshot in cache + listen2Snapshot4PipeListener(false); + return true; + } catch (Exception e) { + logger.error("Failed to load snapshot from {}", latestSnapshotRootDir, e); + return false; + } } public void listen2Snapshot4PipeListener(final boolean isTmp) { From 1c96b5bd964ff27c86bd19fbb93cd01cae8fb4a4 Mon Sep 17 00:00:00 2001 From: Yongzao <532741407@qq.com> Date: Mon, 15 Jun 2026 11:13:39 +0800 Subject: [PATCH 2/7] Address review: don't fail AddPeer on receive folders missing the snapshot Two fixes following review feedback on snapshot-load-failure propagation: 1. IoTConsensusServerImpl.loadSnapshot: a DataRegion configures one receive folder per local data dir and the FolderManager spreads snapshot fragments across them, so a given snapshot only materializes under the folders that actually received fragments. Reporting failure when *any* folder lacks the snapshot turned a healthy multi-data-dir transfer into a spurious failure. Now only load from folders that contain the snapshot and skip the rest, while still failing when a folder that has it fails to load or when no folder received the snapshot at all. 2. ApplicationStateMachineProxy.loadSnapshot (Ratis): previously logged and then advanced lastAppliedTermIndex even when the application state machine rejected the snapshot, letting Ratis proceed on incomplete data. Now fail (re)initialization instead of advancing the applied index. Adds a multi-recvSnapshotDir regression test asserting a healthy transfer that spreads across a subset of receive folders still succeeds (verified to fail against the previous load-from-every-folder behavior). --- .../consensus/iot/IoTConsensusServerImpl.java | 21 ++++-- .../ratis/ApplicationStateMachineProxy.java | 11 ++- .../iot/AddPeerSnapshotLoadFailureTest.java | 71 ++++++++++++++++++- 3 files changed, 92 insertions(+), 11 deletions(-) diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java index 27a2691d63136..036d0dcb772be 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java @@ -512,16 +512,27 @@ private void clearOldSnapshot() { } public boolean loadSnapshot(String snapshotId) { - // Load the snapshot from every receive folder. If any of them fails, report the failure so the - // AddPeer coordinator does not activate this peer with incomplete data (which would silently - // lose data on this replica). + // Snapshot fragments are spread across the receive folders by the FolderManager (a DataRegion, + // for example, uses one receive folder per local data dir), so a given snapshot only exists + // under the folders that actually received fragments. Load from those folders and skip the + // others; otherwise the state machine would fail on a folder that never received this snapshot + // and turn a healthy multi-data-dir transfer into a spurious failure. + boolean snapshotFound = false; boolean success = true; for (String dir : recvFolderManager.getFolders()) { - if (!stateMachine.loadSnapshot(getSnapshotPath(dir, snapshotId))) { + File snapshotDir = getSnapshotPath(dir, snapshotId); + if (!snapshotDir.exists()) { + continue; + } + snapshotFound = true; + if (!stateMachine.loadSnapshot(snapshotDir)) { success = false; } } - return success; + // If no receive folder contained the snapshot, nothing was loaded. Report the failure so the + // AddPeer coordinator does not activate this peer with incomplete data (which would silently + // lose data on this replica). + return snapshotFound && success; } private File getSnapshotPath(String curStorageDir, String snapshotRelativePath) { diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java index dc009a71887e0..3e2451e65dd42 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java @@ -98,7 +98,7 @@ public void initialize(RaftServer raftServer, RaftGroupId raftGroupId, RaftStora } @Override - public void reinitialize() { + public void reinitialize() throws IOException { setLastAppliedTermIndex(null); loadSnapshot(snapshotStorage.findLatestSnapshotDir()); if (getLifeCycleState() == LifeCycle.State.PAUSED) { @@ -261,7 +261,7 @@ private void deleteIncompleteSnapshot(File snapshotDir) throws IOException { } } - private void loadSnapshot(File latestSnapshotDir) { + private void loadSnapshot(File latestSnapshotDir) throws IOException { snapshotStorage.updateSnapshotCache(); if (latestSnapshotDir == null) { return; @@ -269,7 +269,12 @@ private void loadSnapshot(File latestSnapshotDir) { // require the application statemachine to load the latest snapshot if (!applicationStateMachine.loadSnapshot(latestSnapshotDir)) { - logger.error("{}: failed to load snapshot from {}", this, latestSnapshotDir); + // The application state machine rejected this snapshot. Do not advance lastAppliedTermIndex: + // claiming the snapshot as applied would let Ratis proceed as if it were installed and run on + // incomplete data (silent data loss). Fail (re)initialization instead so the snapshot install + // is treated as failed and can be retried. + throw new IOException( + String.format("%s: failed to load snapshot from %s", this, latestSnapshotDir)); } TermIndex snapshotTermIndex = Utils.getTermIndexFromDir(latestSnapshotDir); updateLastAppliedTermIndex(snapshotTermIndex.getTerm(), snapshotTermIndex.getIndex()); diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/AddPeerSnapshotLoadFailureTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/AddPeerSnapshotLoadFailureTest.java index aa69907e6ed4e..b901c8b13b19b 100644 --- a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/AddPeerSnapshotLoadFailureTest.java +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/AddPeerSnapshotLoadFailureTest.java @@ -43,6 +43,8 @@ import java.io.File; import java.io.IOException; import java.net.ServerSocket; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -93,7 +95,13 @@ public class AddPeerSnapshotLoadFailureTest { private final List servers = new ArrayList<>(); private final List stateMachines = new ArrayList<>(); - /** A {@link TestStateMachine} whose snapshot load can be made to fail on demand. */ + /** + * A {@link TestStateMachine} that takes a real (single-file) snapshot and whose snapshot load can + * be made to fail on demand. When not forced to fail, {@link #loadSnapshot} mirrors the real + * {@code SnapshotLoader} contract by reporting failure when the snapshot root does not exist — + * which is exactly what happens for the receive folders that never got a fragment of a snapshot + * in a multi-data-dir deployment. + */ private static class ControllableStateMachine extends TestStateMachine { private volatile boolean failLoadSnapshot = false; private volatile boolean loadSnapshotInvoked = false; @@ -112,12 +120,23 @@ public boolean loadSnapshot(File latestSnapshotRootDir) { if (failLoadSnapshot) { return false; } - return super.loadSnapshot(latestSnapshotRootDir); + // Mirror SnapshotLoader: a receive folder that never received a fragment of this snapshot has + // no snapshot root, so loading from it must report failure. + return latestSnapshotRootDir.exists(); } @Override public boolean takeSnapshot(File snapshotDir) { - return true; + // Write a real (single) snapshot file so the transfer actually moves data and the receiver + // materializes the snapshot under a subset of its receive folders. + try { + Files.write( + new File(snapshotDir, "snapshot.data").toPath(), + "snapshot".getBytes(StandardCharsets.UTF_8)); + return true; + } catch (IOException e) { + return false; + } } // TestStateMachine does not implement clearSnapshot (the IStateMachine default throws). The @@ -230,6 +249,52 @@ public void addRemotePeerMustFailWhenTargetSnapshotLoadFails() throws Exception servers.get(1).getImpl(gid).isActive()); } + /** + * A target peer configures one receive folder per local data dir, and snapshot fragments are + * spread across those folders by the FolderManager, so a small snapshot only materializes under a + * subset of them. A healthy transfer must therefore still succeed even though some receive + * folders never received any fragment of the snapshot (regression for the multi-data-dir + * false-failure reported on the load-failure-propagation PR). + */ + @Test + public void addRemotePeerSucceedsWhenSnapshotSpansSubsetOfRecvDirs() throws Exception { + servers.get(0).createLocalPeer(gid, peers.subList(0, 1)); + servers.get(1).createLocalPeer(gid, peers); + + for (int i = 0; i < 10; i++) { + servers.get(0).write(gid, new TestEntry(i, peers.get(0))); + } + + // Do NOT force a load failure: this is a healthy transfer. The (small) snapshot lands in only a + // subset of the target's receive folders, so the others legitimately have no snapshot root. + // Before the fix, loadSnapshot() loaded from every receive folder and treated the missing ones + // as failures, turning this healthy multi-data-dir transfer into a spurious failure. + servers.get(0).addRemotePeer(gid, peers.get(1)); + + Assert.assertTrue( + "Target peer's loadSnapshot was never invoked", + stateMachines.get(1).isLoadSnapshotInvoked()); + Assert.assertTrue( + "Target peer was not activated after a successful snapshot load", + servers.get(1).getImpl(gid).isActive()); + + // Sanity-check that the test actually exercised a partial spread: at least one of the target's + // receive folders must hold no snapshot at all, otherwise the skipped-folder path is untested. + long emptyRecvFolders = + peersRecvSnapshotDirs.get(1).stream() + .map(dir -> new File(dir, IoTConsensusServerImpl.SNAPSHOT_DIR_NAME)) + .filter(recvFolder -> isEmptyOrMissing(recvFolder)) + .count(); + Assert.assertTrue( + "Expected at least one receive folder without the snapshot, but every folder had it", + emptyRecvFolders > 0); + } + + private static boolean isEmptyOrMissing(File dir) { + String[] children = dir.list(); + return children == null || children.length == 0; + } + private boolean checkPortAvailable() { for (Peer peer : this.peers) { try (ServerSocket ignored = new ServerSocket(peer.getEndpoint().port)) { From 7fe8e9c8e064dc108e3dc7f4b22c52e37f790fde Mon Sep 17 00:00:00 2001 From: Yongzao <532741407@qq.com> Date: Mon, 15 Jun 2026 11:51:28 +0800 Subject: [PATCH 3/7] Propagate ConfigRegion snapshot load failure to the boolean contract ConfigPlanExecutor.loadSnapshot was void: it returned early on a missing snapshot dir and tracked per-processor failures only to decide whether to log success, so ConfigRegionStateMachine.loadSnapshot always returned true even when the load failed. Make ConfigPlanExecutor.loadSnapshot return a boolean (false on missing dir or any processor failure) and have ConfigRegionStateMachine.loadSnapshot propagate it, so AddPeer and other callers observe a failed ConfigRegion snapshot load. --- .../consensus/statemachine/ConfigRegionStateMachine.java | 4 ++-- .../persistence/executor/ConfigPlanExecutor.java | 7 +++++-- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java index 1c2fb38470b61..26a65633b559a 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java @@ -248,13 +248,13 @@ public boolean takeSnapshot(File snapshotDir) { @Override public boolean loadSnapshot(final File latestSnapshotRootDir) { try { - executor.loadSnapshot(latestSnapshotRootDir); + final boolean loadSucceeded = executor.loadSnapshot(latestSnapshotRootDir); // We recompute the snapshot for pipe listener when loading snapshot // to recover the newest snapshot in cache PipeConfigNodeAgent.runtime() .listener() .tryListenToSnapshots(ConfigNodeSnapshotParser.getSnapshots()); - return true; + return loadSucceeded; } catch (final IOException e) { if (PipeConfigNodeAgent.runtime().listener().isOpened()) { LOGGER.warn( diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java index b6d1bba67bd6b..ed23fda8a6f8f 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java @@ -772,12 +772,12 @@ public boolean takeSnapshot(File snapshotDir) { return result.get(); } - public void loadSnapshot(final File latestSnapshotRootDir) { + public boolean loadSnapshot(final File latestSnapshotRootDir) { if (!latestSnapshotRootDir.exists()) { LOGGER.error( ConfigNodeMessages.SNAPSHOT_DIRECTORY_IS_NOT_EXIST_CAN_NOT_LOAD_SNAPSHOT_WITH, latestSnapshotRootDir.getAbsolutePath()); - return; + return false; } final AtomicBoolean result = new AtomicBoolean(true); @@ -805,6 +805,9 @@ public void loadSnapshot(final File latestSnapshotRootDir) { ConfigNodeMessages.CONFIGNODESNAPSHOT_LOAD_SNAPSHOT_SUCCESS_LATESTSNAPSHOTROOTDIR, latestSnapshotRootDir); } + // Propagate any snapshot-load failure so callers (e.g. the AddPeer flow) do not treat a + // partially or wholly failed load as success. + return result.get(); } private DataSet getSchemaNodeManagementPartition(ConfigPhysicalPlan req) { From 7e8514b3d446964951c80323aa12e3dd5e8fbfb8 Mon Sep 17 00:00:00 2001 From: Yongzao <532741407@qq.com> Date: Mon, 15 Jun 2026 12:30:57 +0800 Subject: [PATCH 4/7] Use a meaningful status code and fail fast on snapshot load Two follow-ups on the IoTConsensus snapshot-load path: - IoTConsensusRPCServiceProcessor.triggerSnapshotLoad now reports MIGRATE_REGION_ERROR (the snapshot load runs as part of the AddPeer flow, and this is the code the region-migration handlers already use) instead of the generic INTERNAL_SERVER_ERROR. - IoTConsensusServerImpl.loadSnapshot now returns on the first failing receive folder instead of continuing. Once one folder fails the replica's snapshot is already broken, and loading the remaining folders is not only pointless but harmful, since each DataRegion load wipes the data dirs before relinking. The other consensus protocols were audited and need no change: Simple and Ratis load a snapshot with a single state-machine call (already fail-fast), and IoTConsensusV2 replicates via pipe rather than a snapshot-load loop. --- .../iotdb/consensus/iot/IoTConsensusServerImpl.java | 8 +++++--- .../iot/service/IoTConsensusRPCServiceProcessor.java | 4 +++- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java index 036d0dcb772be..75f72a59058c9 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java @@ -518,7 +518,6 @@ public boolean loadSnapshot(String snapshotId) { // others; otherwise the state machine would fail on a folder that never received this snapshot // and turn a healthy multi-data-dir transfer into a spurious failure. boolean snapshotFound = false; - boolean success = true; for (String dir : recvFolderManager.getFolders()) { File snapshotDir = getSnapshotPath(dir, snapshotId); if (!snapshotDir.exists()) { @@ -526,13 +525,16 @@ public boolean loadSnapshot(String snapshotId) { } snapshotFound = true; if (!stateMachine.loadSnapshot(snapshotDir)) { - success = false; + // Stop at the first failure. The snapshot is already broken on this replica, and loading + // the remaining folders is both pointless and harmful: a load wipes the data dirs before + // relinking. Report the failure so the AddPeer coordinator does not activate this peer. + return false; } } // If no receive folder contained the snapshot, nothing was loaded. Report the failure so the // AddPeer coordinator does not activate this peer with incomplete data (which would silently // lose data on this replica). - return snapshotFound && success; + return snapshotFound; } private File getSnapshotPath(String curStorageDir, String snapshotRelativePath) { diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java index 91e17b370ca1e..82394b6a244a1 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java @@ -348,7 +348,9 @@ public TTriggerSnapshotLoadRes triggerSnapshotLoad(TTriggerSnapshotLoadReq req) String.format( "Failed to load snapshot %s for consensus group %s", req.snapshotId, groupId); LOGGER.error(message); - TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); + // Surface a region-migration-specific code (the snapshot load runs as part of the AddPeer + // flow) rather than a generic internal error, so the coordinator's failure is meaningful. + TSStatus status = new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode()); status.setMessage(message); return new TTriggerSnapshotLoadRes(status); } From f86d5aaa197224e7d7a60d53a814609b89565ee0 Mon Sep 17 00:00:00 2001 From: Yongzao <532741407@qq.com> Date: Mon, 15 Jun 2026 13:58:07 +0800 Subject: [PATCH 5/7] Don't fail ConfigRegion snapshot load on best-effort pipe-listener error ConfigRegionStateMachine.loadSnapshot wrapped both the actual snapshot data load and the best-effort pipe-listener recomputation in one try/catch, and my previous change made the catch return false. That meant a recoverable pipe-listener IOException (which the original code deliberately logged and swallowed) would now be reported as a snapshot-load failure, which in turn aborts ConfigNode (re)initialization via the Ratis loadSnapshot contract. Scope the boolean to the data load (executor.loadSnapshot) only, and keep the pipe-listener recomputation as non-fatal post-processing: a failure there is still logged but no longer flips the load result to false. --- .../statemachine/ConfigRegionStateMachine.java | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java index 26a65633b559a..ad0a82bcf56b3 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java @@ -247,22 +247,26 @@ public boolean takeSnapshot(File snapshotDir) { @Override public boolean loadSnapshot(final File latestSnapshotRootDir) { + // The boolean result must reflect whether the ConfigRegion state-machine data was loaded, so + // callers (e.g. the AddPeer flow) can detect a real failure. The pipe-listener recomputation + // below is best-effort post-processing: a failure there is logged but must NOT be reported as a + // snapshot-load failure, otherwise it would (e.g.) abort ConfigNode (re)initialization on what + // is actually a healthy data load. + final boolean loadSucceeded = executor.loadSnapshot(latestSnapshotRootDir); try { - final boolean loadSucceeded = executor.loadSnapshot(latestSnapshotRootDir); // We recompute the snapshot for pipe listener when loading snapshot // to recover the newest snapshot in cache PipeConfigNodeAgent.runtime() .listener() .tryListenToSnapshots(ConfigNodeSnapshotParser.getSnapshots()); - return loadSucceeded; } catch (final IOException e) { if (PipeConfigNodeAgent.runtime().listener().isOpened()) { LOGGER.warn( ConfigNodeMessages.CONFIG_REGION_LISTENING_QUEUE_LISTEN_TO_SNAPSHOT_FAILED_WHEN_STARTUP, e); } - return false; } + return loadSucceeded; } @Override From edf5971fca6a74d5fb22f4f2872ae74892858a7a Mon Sep 17 00:00:00 2001 From: Yongzao <532741407@qq.com> Date: Mon, 15 Jun 2026 15:02:52 +0800 Subject: [PATCH 6/7] Treat an empty snapshot as a successful load in IoTConsensus AddPeer An empty region produces a snapshot with zero fragments, so the source transmits 0 files and the target materializes no snapshot under any receive folder. My earlier change reported "no folder contained the snapshot" as a failure, which broke AddPeer for empty regions: migrating an empty data region (e.g. while removing a DataNode) failed at DO_ADD_REGION_PEER and the removal timed out. This reproduced in the Cluster IT - 1C3D job (IoTDBRemoveDataNodeNormalIT.success1C4DIoTTestUseTableSQL). loadSnapshot now loads only from the receive folders that actually contain the snapshot, fails on the first folder whose load fails (preserving the original V2-987 failure propagation), and treats an absent snapshot (empty region) as a no-op success. Adds addRemotePeerSucceedsWhenSnapshotIsEmpty, verified to fail against the previous behavior. --- .../consensus/iot/IoTConsensusServerImpl.java | 11 +++-- .../iot/AddPeerSnapshotLoadFailureTest.java | 45 +++++++++++++++++++ 2 files changed, 50 insertions(+), 6 deletions(-) diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java index 75f72a59058c9..017b646e60693 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java @@ -517,13 +517,15 @@ public boolean loadSnapshot(String snapshotId) { // under the folders that actually received fragments. Load from those folders and skip the // others; otherwise the state machine would fail on a folder that never received this snapshot // and turn a healthy multi-data-dir transfer into a spurious failure. - boolean snapshotFound = false; + // + // Note: an empty region produces a snapshot with zero fragments, so none of the receive folders + // contains it. That is a legitimate (no-op) load, not a failure, so an absent snapshot must not + // be reported as failure here. for (String dir : recvFolderManager.getFolders()) { File snapshotDir = getSnapshotPath(dir, snapshotId); if (!snapshotDir.exists()) { continue; } - snapshotFound = true; if (!stateMachine.loadSnapshot(snapshotDir)) { // Stop at the first failure. The snapshot is already broken on this replica, and loading // the remaining folders is both pointless and harmful: a load wipes the data dirs before @@ -531,10 +533,7 @@ public boolean loadSnapshot(String snapshotId) { return false; } } - // If no receive folder contained the snapshot, nothing was loaded. Report the failure so the - // AddPeer coordinator does not activate this peer with incomplete data (which would silently - // lose data on this replica). - return snapshotFound; + return true; } private File getSnapshotPath(String curStorageDir, String snapshotRelativePath) { diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/AddPeerSnapshotLoadFailureTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/AddPeerSnapshotLoadFailureTest.java index b901c8b13b19b..64c81a84b5896 100644 --- a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/AddPeerSnapshotLoadFailureTest.java +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/AddPeerSnapshotLoadFailureTest.java @@ -105,11 +105,16 @@ public class AddPeerSnapshotLoadFailureTest { private static class ControllableStateMachine extends TestStateMachine { private volatile boolean failLoadSnapshot = false; private volatile boolean loadSnapshotInvoked = false; + private volatile boolean emptySnapshot = false; void setFailLoadSnapshot(boolean failLoadSnapshot) { this.failLoadSnapshot = failLoadSnapshot; } + void setEmptySnapshot(boolean emptySnapshot) { + this.emptySnapshot = emptySnapshot; + } + boolean isLoadSnapshotInvoked() { return loadSnapshotInvoked; } @@ -127,6 +132,11 @@ public boolean loadSnapshot(File latestSnapshotRootDir) { @Override public boolean takeSnapshot(File snapshotDir) { + if (emptySnapshot) { + // Mirror an empty region: the snapshot has no fragments to transmit. The receiver then + // materializes no snapshot under any receive folder, which must still load successfully. + return true; + } // Write a real (single) snapshot file so the transfer actually moves data and the receiver // materializes the snapshot under a subset of its receive folders. try { @@ -295,6 +305,41 @@ private static boolean isEmptyOrMissing(File dir) { return children == null || children.length == 0; } + /** + * An empty region produces a snapshot with zero fragments, so nothing is transmitted and the + * target materializes no snapshot under any receive folder. This must still be treated as a + * successful (no-op) load — otherwise migrating an empty region (e.g. while removing a DataNode) + * fails. Regression for the multi-data-dir AddPeer false-failure on empty regions. + */ + @Test + public void addRemotePeerSucceedsWhenSnapshotIsEmpty() throws Exception { + servers.get(0).createLocalPeer(gid, peers.subList(0, 1)); + servers.get(1).createLocalPeer(gid, peers); + + // No data is written, and the source takes an empty snapshot: zero fragments are transmitted. + stateMachines.get(0).setEmptySnapshot(true); + + // Before the fix, loadSnapshot() reported failure when no receive folder contained the + // snapshot, so adding a peer for an empty region failed and DataNode removal timed out. + servers.get(0).addRemotePeer(gid, peers.get(1)); + + Assert.assertTrue( + "Target peer was not activated after an empty snapshot load", + servers.get(1).getImpl(gid).isActive()); + + // Confirm the test really exercised the empty-snapshot path: no receive folder holds a + // snapshot. + long nonEmptyRecvFolders = + peersRecvSnapshotDirs.get(1).stream() + .map(dir -> new File(dir, IoTConsensusServerImpl.SNAPSHOT_DIR_NAME)) + .filter(recvFolder -> !isEmptyOrMissing(recvFolder)) + .count(); + Assert.assertEquals( + "Expected no receive folder to contain a snapshot for an empty region", + 0, + nonEmptyRecvFolders); + } + private boolean checkPortAvailable() { for (Peer peer : this.peers) { try (ServerSocket ignored = new ServerSocket(peer.getEndpoint().port)) { From ddeb9f8fd16eb9e4c73706aa91c098cda4d22e1e Mon Sep 17 00:00:00 2001 From: Yongzao <532741407@qq.com> Date: Mon, 15 Jun 2026 19:49:59 +0800 Subject: [PATCH 7/7] Propagate SchemaRegion snapshot load failure to the boolean contract SchemaRegionStateMachine.loadSnapshot returned true after calling schemaRegion.loadSnapshot(...), but the underlying ISchemaRegion implementations swallowed load failures: both SchemaRegionPBTreeImpl and SchemaRegionMemoryImpl catch the load exception, log FAILED_TO_LOAD_SNAPSHOT, fall back to init(), and return void. So a failed schema-snapshot load was reported as success, weakening the IStateMachine.loadSnapshot boolean contract (and letting the Ratis snapshot-install path advance the applied index after a rejected snapshot). Make ISchemaRegion.loadSnapshot return boolean (false when the load failed and the region fell back to an empty state), have both implementations report it, and have SchemaRegionStateMachine propagate the data-load result while keeping the pipe-listener recomputation as non-fatal post-processing. Adds testLoadSnapshotReportsFailureWhenSnapshotIsMissing and asserts the success path returns true in the existing snapshot tests; verified the new test fails across all schema-region modes against the previous behavior. --- .../SchemaRegionStateMachine.java | 7 +++-- .../schemaregion/ISchemaRegion.java | 10 ++++++- .../impl/SchemaRegionMemoryImpl.java | 6 +++- .../impl/SchemaRegionPBTreeImpl.java | 6 +++- .../SchemaRegionManagementTest.java | 28 ++++++++++++++++--- 5 files changed, 48 insertions(+), 9 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java index 3f647ced7a5e7..2f5ae61e85e0a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java @@ -138,14 +138,17 @@ public boolean takeSnapshot(final File snapshotDir) { @Override public boolean loadSnapshot(final File latestSnapshotRootDir) { try { - schemaRegion.loadSnapshot(latestSnapshotRootDir); + // The boolean result must reflect whether the schema-region data was actually loaded, so + // callers (e.g. the AddPeer flow and the Ratis snapshot-install path) can detect a real + // failure instead of treating a fallback-to-empty load as success. + final boolean loadSucceeded = schemaRegion.loadSnapshot(latestSnapshotRootDir); PipeDataNodeAgent.runtime() .schemaListener(schemaRegion.getSchemaRegionId()) .loadSnapshot(latestSnapshotRootDir); // We recompute the snapshot for pipe listener when loading snapshot // to recover the newest snapshot in cache listen2Snapshot4PipeListener(false); - return true; + return loadSucceeded; } catch (Exception e) { logger.error("Failed to load snapshot from {}", latestSnapshotRootDir, e); return false; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java index 8d87435e108a8..540d5d4274c46 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java @@ -115,7 +115,15 @@ public interface ISchemaRegion { boolean createSnapshot(final File snapshotDir); - void loadSnapshot(final File latestSnapshotRootDir); + /** + * Load the latest schema snapshot from the given dir. + * + * @return {@code true} if the snapshot was loaded successfully, {@code false} if loading failed + * (in which case the region falls back to an empty/re-initialized state). Callers rely on + * this to honor the {@link org.apache.iotdb.consensus.IStateMachine#loadSnapshot} + * success/failure contract (e.g. the AddPeer flow and Ratis snapshot install). + */ + boolean loadSnapshot(final File latestSnapshotRootDir); // endregion diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java index 0394a81680f07..1f2408cca1237 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java @@ -556,7 +556,7 @@ public synchronized boolean createSnapshot(final File snapshotDir) { // currently, this method is only used for cluster-ratis mode @Override - public void loadSnapshot(final File latestSnapshotRootDir) { + public boolean loadSnapshot(final File latestSnapshotRootDir) { clear(); logger.info(DataNodeSchemaMessages.START_LOADING_SNAPSHOT, schemaRegionId); @@ -651,6 +651,7 @@ public void loadSnapshot(final File latestSnapshotRootDir) { schemaRegionId, System.currentTimeMillis() - startTime); logger.info(DataNodeSchemaMessages.SUCCESSFULLY_LOAD_SNAPSHOT, schemaRegionId); + return true; } catch (final Exception e) { logger.error( DataNodeSchemaMessages.FAILED_TO_LOAD_SNAPSHOT, schemaRegionId, e.getMessage(), e); @@ -662,6 +663,9 @@ public void loadSnapshot(final File latestSnapshotRootDir) { logger.error( DataNodeSchemaMessages.ERROR_DURING_INIT_SCHEMA_REGION, schemaRegionId, exception); } + // The snapshot was not loaded (the region fell back to an empty re-initialized state). Report + // the failure so callers honoring the loadSnapshot success/failure contract can react. + return false; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java index df5ad41cff6e6..8d87ef3f0618f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java @@ -534,7 +534,7 @@ public boolean createSnapshot(File snapshotDir) { // currently, this method is only used for cluster-ratis mode @Override - public void loadSnapshot(File latestSnapshotRootDir) { + public boolean loadSnapshot(File latestSnapshotRootDir) { clear(); logger.info(DataNodeSchemaMessages.START_LOADING_SNAPSHOT, schemaRegionId); @@ -579,6 +579,7 @@ public void loadSnapshot(File latestSnapshotRootDir) { schemaRegionId, System.currentTimeMillis() - startTime); logger.info(DataNodeSchemaMessages.SUCCESSFULLY_LOAD_SNAPSHOT, schemaRegionId); + return true; } catch (IOException | MetadataException e) { logger.error( DataNodeSchemaMessages.FAILED_TO_LOAD_SNAPSHOT, schemaRegionId, e.getMessage(), e); @@ -592,6 +593,9 @@ public void loadSnapshot(File latestSnapshotRootDir) { schemaRegionId, metadataException); } + // The snapshot was not loaded (the region fell back to an empty re-initialized state). Report + // the failure so callers honoring the loadSnapshot success/failure contract can react. + return false; } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionManagementTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionManagementTest.java index 38f56b66dd9f0..0c0b3a5b953e6 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionManagementTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionManagementTest.java @@ -96,7 +96,7 @@ public void testRatisModeSnapshot() throws Exception { snapshotDir.mkdir(); schemaRegion.createSnapshot(snapshotDir); - schemaRegion.loadSnapshot(snapshotDir); + Assert.assertTrue(schemaRegion.loadSnapshot(snapshotDir)); List result = SchemaRegionTestUtil.showTimeseries( @@ -112,7 +112,7 @@ public void testRatisModeSnapshot() throws Exception { simulateRestart(); ISchemaRegion newSchemaRegion = getSchemaRegion("root.sg", 0); - newSchemaRegion.loadSnapshot(snapshotDir); + Assert.assertTrue(newSchemaRegion.loadSnapshot(snapshotDir)); result = SchemaRegionTestUtil.showTimeseries( newSchemaRegion, new PartialPath("root.sg.**"), false, "tag-key", "tag-value"); @@ -171,7 +171,7 @@ public void testEmptySnapshot() throws Exception { snapshotDir.mkdir(); schemaRegion.createSnapshot(snapshotDir); - schemaRegion.loadSnapshot(snapshotDir); + Assert.assertTrue(schemaRegion.loadSnapshot(snapshotDir)); List result = SchemaRegionTestUtil.showTimeseries( @@ -182,7 +182,7 @@ public void testEmptySnapshot() throws Exception { simulateRestart(); ISchemaRegion newSchemaRegion = getSchemaRegion("root.sg", 0); - newSchemaRegion.loadSnapshot(snapshotDir); + Assert.assertTrue(newSchemaRegion.loadSnapshot(snapshotDir)); result = SchemaRegionTestUtil.showTimeseries( newSchemaRegion, new PartialPath("root.sg.**"), false, "tag-key", "tag-value"); @@ -193,6 +193,26 @@ public void testEmptySnapshot() throws Exception { } } + @Test + public void testLoadSnapshotReportsFailureWhenSnapshotIsMissing() throws Exception { + String schemaRegionConsensusProtocolClass = config.getSchemaRegionConsensusProtocolClass(); + config.setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS); + try { + ISchemaRegion schemaRegion = getSchemaRegion("root.sg", 0); + + // Loading from a directory that does not contain a snapshot must report failure rather than + // silently falling back to an empty region and reporting success. Callers (the AddPeer flow + // and the Ratis snapshot-install path) rely on this success/failure contract. + File missingSnapshotDir = + new File(config.getSchemaDir() + File.separator + "non-existent-snapshot"); + Assert.assertFalse(missingSnapshotDir.exists()); + + Assert.assertFalse(schemaRegion.loadSnapshot(missingSnapshotDir)); + } finally { + config.setSchemaRegionConsensusProtocolClass(schemaRegionConsensusProtocolClass); + } + } + @Test @Ignore public void testSnapshotPerformance() throws Exception {