Propagate snapshot load failure during IoTConsensus AddPeer

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java

@@ -246,9 +246,14 @@ public boolean takeSnapshot(File snapshotDir) {
  }
 
  @Override
- public void loadSnapshot(final File latestSnapshotRootDir) {
+ 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 {
- executor.loadSnapshot(latestSnapshotRootDir);
  // We recompute the snapshot for pipe listener when loading snapshot
  // to recover the newest snapshot in cache
  PipeConfigNodeAgent.runtime()
@@ -261,6 +266,7 @@ public void loadSnapshot(final File latestSnapshotRootDir) {
  e);
  }
  }
+ return loadSucceeded;
  }
 
  @Override

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) {

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

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java

@@ -511,14 +511,29 @@ 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) {
+ // 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.
+ //
+ // 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;
+ }
+ 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
+ // relinking. Report the failure so the AddPeer coordinator does not activate this peer.
+ return false;
+ }
+ }
+ return true;
  }
 
  private File getSnapshotPath(String curStorageDir, String snapshotRelativePath) {

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java

@@ -343,7 +343,17 @@ 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);
+ // 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);
+ }
  KillPoint.setKillPoint(DataNodeKillPoints.DESTINATION_ADD_PEER_TRANSITION);
  return new TTriggerSnapshotLoadRes(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
  }

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,14 +261,21 @@ 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;
  }
 
  // require the application statemachine to load the latest snapshot
- applicationStateMachine.loadSnapshot(latestSnapshotDir);
+ if (!applicationStateMachine.loadSnapshot(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());
  }

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);
  }
 }

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;
+ }
 }