This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 2a96cca633e Fix Region recovery validation before DataNode startup
success (#18311) (#18321)
2a96cca633e is described below
commit 2a96cca633e87e1dce83bfc6ce7589bd811e6655
Author: Caideyipi <[email protected]>
AuthorDate: Tue Jul 28 18:36:04 2026 +0800
Fix Region recovery validation before DataNode startup success (#18311)
(#18321)
* Fix NPE when creating SchemaRegion state machine
* Propagate Region recovery failures before startup success
---
.../apache/iotdb/consensus/pipe/PipeConsensus.java | 99 +++++++++++++++-------
.../iotdb/consensus/pipe/PipeConsensusTest.java | 55 ++++++++++++
.../db/consensus/SchemaRegionConsensusImpl.java | 23 ++++-
.../java/org/apache/iotdb/db/service/DataNode.java | 11 ++-
.../iotdb/db/storageengine/StorageEngine.java | 4 +-
.../db/storageengine/dataregion/DataRegion.java | 3 -
6 files changed, 152 insertions(+), 43 deletions(-)
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java
index 86e8fec7bd8..fc70f5c2d83 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java
@@ -70,8 +70,12 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;
@@ -116,7 +120,7 @@ public class PipeConsensus implements IConsensus {
@Override
public synchronized void start() throws IOException {
- initAndRecover();
+ Future<Void> recoverFuture = initAndRecover();
rpcService.initSyncedServiceImpl(new
PipeConsensusRPCServiceProcessor(this, config.getPipe()));
try {
@@ -125,50 +129,83 @@ public class PipeConsensus implements IConsensus {
throw new IOException(e);
}
+ waitForRecovery(recoverFuture);
+
consensusPipeGuardian.start(
CONSENSUS_PIPE_GUARDIAN_TASK_ID,
this::checkAllConsensusPipe,
config.getPipe().getConsensusPipeGuardJobIntervalInSeconds());
}
- private void initAndRecover() throws IOException {
+ static void waitForRecovery(Future<Void> recoverFuture) throws IOException {
+ try {
+ recoverFuture.get();
+ } catch (CancellationException e) {
+ throw new IOException("IoTV2 Recover Task is cancelled", e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof CompletionException && cause.getCause() != null) {
+ cause = cause.getCause();
+ }
+ if (cause instanceof IOException) {
+ throw (IOException) cause;
+ }
+ if (cause instanceof RuntimeException) {
+ throw (RuntimeException) cause;
+ }
+ if (cause instanceof Error) {
+ throw (Error) cause;
+ }
+ throw new IOException("Exception while waiting for recover future
completion", cause);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("IoTV2 Recover Task is interrupted", e);
+ }
+ }
+
+ private Future<Void> initAndRecover() throws IOException {
if (!storageDir.exists()) {
// init
if (!storageDir.mkdirs()) {
LOGGER.warn("Unable to create consensus dir at {}", storageDir);
throw new IOException(String.format("Unable to create consensus dir at
%s", storageDir));
}
+ return CompletableFuture.completedFuture(null);
} else {
// asynchronously recover, retry logic is implemented at
PipeConsensusImpl
- CompletableFuture<Void> future =
- CompletableFuture.runAsync(
- () -> {
- try (DirectoryStream<Path> stream =
- Files.newDirectoryStream(storageDir.toPath())) {
- for (Path path : stream) {
- ConsensusGroupId consensusGroupId =
- parsePeerFileName(path.getFileName().toString());
- PipeConsensusServerImpl consensus =
- new PipeConsensusServerImpl(
- new Peer(consensusGroupId, thisNodeId,
thisNode),
- registry.apply(consensusGroupId),
- path.toString(),
- new ArrayList<>(),
- config,
- consensusPipeManager,
- syncClientManager);
- stateMachineMap.put(consensusGroupId, consensus);
- checkPeerListAndStartIfEligible(consensusGroupId,
consensus);
- }
- } catch (Exception e) {
- LOGGER.error("Failed to recover consensus from {}",
storageDir, e);
- }
- })
- .exceptionally(
- e -> {
- LOGGER.error("Failed to recover consensus from {}",
storageDir, e);
- return null;
- });
+ return CompletableFuture.runAsync(
+ () -> {
+ try (DirectoryStream<Path> stream =
Files.newDirectoryStream(storageDir.toPath())) {
+ for (Path path : stream) {
+ ConsensusGroupId consensusGroupId =
+ parsePeerFileName(path.getFileName().toString());
+ IStateMachine stateMachine = registry.apply(consensusGroupId);
+ try {
+ PipeConsensusServerImpl consensus =
+ new PipeConsensusServerImpl(
+ new Peer(consensusGroupId, thisNodeId, thisNode),
+ stateMachine,
+ path.toString(),
+ new ArrayList<>(),
+ config,
+ consensusPipeManager,
+ syncClientManager);
+ stateMachineMap.put(consensusGroupId, consensus);
+ checkPeerListAndStartIfEligible(consensusGroupId, consensus);
+ } catch (Exception e) {
+ LOGGER.error(
+ "Failed to recover consensus from {} for {}, ignore it
and continue recover other group, async backend checker thread will
automatically deregister related pipe side effects for this failed consensus
group.",
+ storageDir,
+ consensusGroupId,
+ e);
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error(
+ "Failed to recover consensus from {} because read dir
failed", storageDir, e);
+ throw new CompletionException(e);
+ }
+ });
}
}
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/pipe/PipeConsensusTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/pipe/PipeConsensusTest.java
new file mode 100644
index 00000000000..442e7c97021
--- /dev/null
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/pipe/PipeConsensusTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.pipe;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+
+public class PipeConsensusTest {
+
+ @Test
+ public void testWaitForRecoveryPropagatesRuntimeException() {
+ IllegalArgumentException cause = new IllegalArgumentException("missing
DataRegion");
+ CompletableFuture<Void> recoverFuture = new CompletableFuture<>();
+ recoverFuture.completeExceptionally(cause);
+
+ IllegalArgumentException exception =
+ Assert.assertThrows(
+ IllegalArgumentException.class, () ->
PipeConsensus.waitForRecovery(recoverFuture));
+
+ Assert.assertSame(cause, exception);
+ }
+
+ @Test
+ public void
testWaitForRecoveryPropagatesIOExceptionWrappedByCompletionException() {
+ IOException cause = new IOException("failed to read consensus directory");
+ CompletableFuture<Void> recoverFuture = new CompletableFuture<>();
+ recoverFuture.completeExceptionally(new CompletionException(cause));
+
+ IOException exception =
+ Assert.assertThrows(IOException.class, () ->
PipeConsensus.waitForRecovery(recoverFuture));
+
+ Assert.assertSame(cause, exception);
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/SchemaRegionConsensusImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/SchemaRegionConsensusImpl.java
index 0cc42363072..12ddfa20a3b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/SchemaRegionConsensusImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/SchemaRegionConsensusImpl.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.consensus;
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.SchemaRegionId;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.IConsensus;
@@ -30,9 +31,12 @@ import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import
org.apache.iotdb.db.consensus.statemachine.schemaregion.SchemaRegionStateMachine;
import org.apache.iotdb.db.schemaengine.SchemaEngine;
+import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion;
import org.apache.ratis.util.SizeInBytes;
import org.apache.ratis.util.TimeDuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.concurrent.TimeUnit;
@@ -42,6 +46,8 @@ import java.util.concurrent.TimeUnit;
*/
public class SchemaRegionConsensusImpl {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(SchemaRegionConsensusImpl.class);
+
private SchemaRegionConsensusImpl() {
// do nothing
}
@@ -170,9 +176,7 @@ public class SchemaRegionConsensusImpl {
.build())
.setStorageDir(CONF.getSchemaRegionConsensusDir())
.build(),
- gid ->
- new SchemaRegionStateMachine(
-
SchemaEngine.getInstance().getSchemaRegion((SchemaRegionId) gid)))
+
SchemaRegionConsensusImplHolder::createSchemaRegionStateMachine)
.orElseThrow(
() ->
new IllegalArgumentException(
@@ -180,5 +184,18 @@ public class SchemaRegionConsensusImpl {
ConsensusFactory.CONSTRUCT_FAILED_MSG,
CONF.getSchemaRegionConsensusProtocolClass())));
}
+
+ private static SchemaRegionStateMachine
createSchemaRegionStateMachine(ConsensusGroupId gid) {
+ ISchemaRegion schemaRegion =
SchemaEngine.getInstance().getSchemaRegion((SchemaRegionId) gid);
+ if (schemaRegion == null) {
+ String errorMsg =
+ String.format(
+ "Failed to create state machine for consensus group %s,
because schema region does not exist",
+ gid);
+ LOGGER.error(errorMsg);
+ throw new IllegalArgumentException(errorMsg);
+ }
+ return new SchemaRegionStateMachine(schemaRegion);
+ }
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
index 44240e767e9..8967504e25a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -169,6 +169,7 @@ public class DataNode extends ServerCommandLine implements
DataNodeMBean {
private boolean schemaRegionConsensusStarted = false;
private boolean dataRegionConsensusStarted = false;
+ private long schemaEngineRecoveryTimeInMs;
private static Thread watcherThread;
public DataNode() {
@@ -710,12 +711,12 @@ public class DataNode extends ServerCommandLine
implements DataNodeMBean {
logger.error("Meet error while starting up.", e);
throw new StartupException("Error in activating IoTDB DataNode.");
}
- logger.info("IoTDB DataNode has started.");
try {
long startTime = System.currentTimeMillis();
SchemaRegionConsensusImpl.getInstance().start();
long schemaRegionEndTime = System.currentTimeMillis();
+ logger.info("Recover schema successfully, which takes {} ms.",
schemaEngineRecoveryTimeInMs);
logger.info(
"SchemaRegion consensus start successfully, which takes {} ms.",
(schemaRegionEndTime - startTime));
@@ -731,6 +732,7 @@ public class DataNode extends ServerCommandLine implements
DataNodeMBean {
} catch (IOException e) {
throw new StartupException(e);
}
+ logger.info("IoTDB DataNode has started.");
}
void processPid() {
@@ -789,7 +791,9 @@ public class DataNode extends ServerCommandLine implements
DataNodeMBean {
}
}
long endTime = System.currentTimeMillis();
- logger.info("Wait for all databases ready, which takes {} ms.", (endTime -
startTime));
+ logger.info(
+ "Wait for local DataRegion recovery tasks to finish, which takes {}
ms.",
+ (endTime - startTime));
// Must init after SchemaEngine and StorageEngine prepared well
DataNodeRegionManager.getInstance().init();
@@ -1153,8 +1157,7 @@ public class DataNode extends ServerCommandLine
implements DataNodeMBean {
private void initSchemaEngine() {
long startTime = System.currentTimeMillis();
SchemaEngine.getInstance().init();
- long endTime = System.currentTimeMillis();
- logger.info("Recover schema successfully, which takes {} ms.", (endTime -
startTime));
+ schemaEngineRecoveryTimeInMs = System.currentTimeMillis() - startTime;
}
private void classLoader() {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
index ee214fd1358..10a3aec8395 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
@@ -218,7 +218,7 @@ public class StorageEngine implements IService {
checkResults(futures, "StorageEngine failed to recover.");
isReadyForReadAndWrite.set(true);
LOGGER.info(
- "Storage Engine recover cost: {}s.",
+ "Storage Engine local recovery tasks finished in {}s.",
(System.currentTimeMillis() - startRecoverTime) / 1000);
},
ThreadName.STORAGE_ENGINE_RECOVER_TRIGGER.getName());
@@ -247,7 +247,7 @@ public class StorageEngine implements IService {
}
dataRegionMap.put(dataRegionId, dataRegion);
LOGGER.info(
- "Data regions have been recovered {}/{}",
+ "Local DataRegion loading progress: {}/{}.",
readyDataRegionNum.incrementAndGet(),
recoverDataRegionNum);
return null;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index 47032a273a8..2900d26ef74 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -675,9 +675,6 @@ public class DataRegion implements IDataRegionForQuery {
}
logger.info(
"The data region {}[{}] is created successfully", databaseName,
dataRegionIdString);
- } else {
- logger.info(
- "The data region {}[{}] is recovered successfully", databaseName,
dataRegionIdString);
}
}
