Add recover IT for Procedure, and delete ProcedureStore

integration-test/src/test/java/org/apache/iotdb/confignode/it/procedure/IoTDBProcedureIT.java

@@ -0,0 +1,138 @@
+/*
+ * 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.confignode.it.procedure;
+
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.schema.SchemaConstant;
+import org.apache.iotdb.confignode.procedure.impl.CreateManyDatabasesProcedure;
+import org.apache.iotdb.confignode.rpc.thrift.TGetDatabaseReq;
+import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowDatabaseStatement;
+import org.apache.iotdb.db.utils.constant.SqlConstant;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+
+import org.awaitility.Awaitility;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.iotdb.confignode.procedure.impl.CreateManyDatabasesProcedure.MAX_STATE;
+import static org.apache.iotdb.consensus.ConsensusFactory.RATIS_CONSENSUS;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({LocalStandaloneIT.class, ClusterIT.class})
+public class IoTDBProcedureIT {
+  private static Logger LOGGER = LoggerFactory.getLogger(IoTDBProcedureIT.class);
+
+  @Before
+  public void setUp() {
+    EnvFactory.getEnv()
+        .getConfig()
+        .getCommonConfig()
+        .setConfigNodeConsensusProtocolClass(RATIS_CONSENSUS)
+        .setSchemaRegionConsensusProtocolClass(RATIS_CONSENSUS)
+        .setDataRegionConsensusProtocolClass(RATIS_CONSENSUS)
+        .setDataReplicationFactor(1);
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    EnvFactory.getEnv().cleanClusterEnvironment();
+  }
+
+  /**
+   * During CreateManyDatabasesProcedure executing, we expect that the procedure will be interrupted
+   * only once. So let us shutdown the leader at the middle of the procedure.
+   */
+  @Test
+  public void procedureRecoverAtAnotherConfigNodeTest() throws Exception {
+    recoverTest(3, false);
+  }
+
+  @Test
+  public void procedureRecoverAtTheSameConfigNodeTest() throws Exception {
+    recoverTest(1, true);
+  }
+
+  private void recoverTest(int configNodeNum, boolean needReopenLeader) throws Exception {
+    EnvFactory.getEnv().initClusterEnvironment(configNodeNum, 1);
+
+    // prepare expectedDatabases
+    Set<String> expectedDatabases = new HashSet<>();
+    for (int id = CreateManyDatabasesProcedure.getInitialStateStatic(); id < MAX_STATE; id++) {
+      expectedDatabases.add(CreateManyDatabasesProcedure.DATABASE_NAME_PREFIX + id);
+    }
+    Assert.assertEquals(MAX_STATE, expectedDatabases.size());
+
+    // prepare req
+    final TGetDatabaseReq req =
+        new TGetDatabaseReq(
+            Arrays.asList(
+                new ShowDatabaseStatement(new PartialPath(SqlConstant.getSingleRootArray()))
+                    .getPathPattern()
+                    .getNodes()),
+            SchemaConstant.ALL_MATCH_SCOPE.serialize());
+
+    SyncConfigNodeIServiceClient leaderClient =
+        (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection();
+    leaderClient.createManyDatabases();
+
+    // Check whether the procedure is ongoing, and is still far from finishing
+    TShowDatabaseResp resp = leaderClient.showDatabase(req);
+    Assert.assertTrue(0 < resp.getDatabaseInfoMap().size());
+    Assert.assertTrue(resp.getDatabaseInfoMap().size() < MAX_STATE);
+    // Then shutdown the leader, wait the new leader exist and the procedure continue
+    final int oldLeaderIndex = EnvFactory.getEnv().getLeaderConfigNodeIndex();
+    EnvFactory.getEnv().getConfigNodeWrapper(oldLeaderIndex).stop();
+    if (needReopenLeader) {
+      EnvFactory.getEnv().getConfigNodeWrapper(oldLeaderIndex).start();
+    }
+    SyncConfigNodeIServiceClient newLeaderClient =
+        (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection();
+    Callable<Boolean> finalCheck =
+        () -> {
+          TShowDatabaseResp resp1 = newLeaderClient.showDatabase(req);
+          if (MAX_STATE != resp1.getDatabaseInfoMap().size()) {
+            return false;
+          }
+          resp1
+              .getDatabaseInfoMap()
+              .keySet()
+              .forEach(databaseName -> expectedDatabases.remove(databaseName));
+          return expectedDatabases.isEmpty();
+        };
+    Awaitility.await().atMost(5, TimeUnit.SECONDS).until(finalCheck);
+  }
+}

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java

@@ -45,6 +45,7 @@
 import org.apache.iotdb.commons.utils.AuthUtils;
 import org.apache.iotdb.commons.utils.PathUtils;
 import org.apache.iotdb.commons.utils.StatusUtils;
+import org.apache.iotdb.commons.utils.TestOnly;
 import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
 import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
 import org.apache.iotdb.confignode.conf.SystemPropertiesUtils;
@@ -605,6 +606,12 @@ public synchronized TSStatus deleteDatabases(List<String> deletedPaths) {
     }
   }
 
+  @TestOnly
+  @Override
+  public TSStatus createManyDatabases() {
+    return getProcedureManager().createManyDatabases();
+  }
+
   private List<TSeriesPartitionSlot> calculateRelatedSlot(PartialPath path, PartialPath database) {
     // The path contains `**`
     if (path.getFullPath().contains(IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD)) {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java

@@ -26,6 +26,7 @@
 import org.apache.iotdb.common.rpc.thrift.TSetSpaceQuotaReq;
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.path.PathPatternTree;
+import org.apache.iotdb.commons.utils.TestOnly;
 import org.apache.iotdb.confignode.consensus.request.auth.AuthorPlan;
 import org.apache.iotdb.confignode.consensus.request.read.database.CountDatabasePlan;
 import org.apache.iotdb.confignode.consensus.request.read.database.GetDatabasePlan;
@@ -314,6 +315,14 @@ public interface IManager {
    */
   TSStatus deleteDatabases(List<String> deletedPaths);
 
+  /**
+   * Create many databases.
+   *
+   * @return status
+   */
+  @TestOnly
+  TSStatus createManyDatabases();
+
   /**
    * Get SchemaPartition.
    *

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java

@@ -37,6 +37,7 @@
 import org.apache.iotdb.commons.service.metric.MetricService;
 import org.apache.iotdb.commons.trigger.TriggerInformation;
 import org.apache.iotdb.commons.utils.StatusUtils;
+import org.apache.iotdb.commons.utils.TestOnly;
 import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
 import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
 import org.apache.iotdb.confignode.consensus.request.auth.AuthorPlan;
@@ -50,6 +51,7 @@
 import org.apache.iotdb.confignode.procedure.ProcedureExecutor;
 import org.apache.iotdb.confignode.procedure.ProcedureMetrics;
 import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import org.apache.iotdb.confignode.procedure.impl.CreateManyDatabasesProcedure;
 import org.apache.iotdb.confignode.procedure.impl.cq.CreateCQProcedure;
 import org.apache.iotdb.confignode.procedure.impl.node.AddConfigNodeProcedure;
 import org.apache.iotdb.confignode.procedure.impl.node.RemoveConfigNodeProcedure;
@@ -81,7 +83,6 @@
 import org.apache.iotdb.confignode.procedure.store.ConfigProcedureStore;
 import org.apache.iotdb.confignode.procedure.store.IProcedureStore;
 import org.apache.iotdb.confignode.procedure.store.ProcedureFactory;
-import org.apache.iotdb.confignode.procedure.store.ProcedureStore;
 import org.apache.iotdb.confignode.procedure.store.ProcedureType;
 import org.apache.iotdb.confignode.rpc.thrift.TAlterLogicalViewReq;
 import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq;
@@ -172,6 +173,12 @@ public void shiftExecutor(boolean running) {
     }
   }
 
+  @TestOnly
+  public TSStatus createManyDatabases() {
+    this.executor.submitProcedure(new CreateManyDatabasesProcedure());
+    return StatusUtils.OK;
+  }
+
   public TSStatus deleteDatabases(ArrayList<TDatabaseSchema> deleteSgSchemaList) {
     List<Long> procedureIds = new ArrayList<>();
     for (TDatabaseSchema storageGroupSchema : deleteSgSchemaList) {
@@ -1023,10 +1030,6 @@ public IProcedureStore getStore() {
     return store;
   }
 
-  public void setStore(ProcedureStore store) {
-    this.store = store;
-  }
-
   public ConfigNodeProcedureEnv getEnv() {
     return env;
   }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/ProcedureInfo.java

@@ -25,7 +25,6 @@
 import org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProcedurePlan;
 import org.apache.iotdb.confignode.procedure.Procedure;
 import org.apache.iotdb.confignode.procedure.store.ProcedureFactory;
-import org.apache.iotdb.confignode.procedure.store.ProcedureStore;
 import org.apache.iotdb.confignode.procedure.store.ProcedureWAL;
 import org.apache.iotdb.rpc.TSStatusCode;
 
@@ -44,14 +43,16 @@ public class ProcedureInfo {
 
   private static final Logger LOG = LoggerFactory.getLogger(ProcedureInfo.class);
 
+  public static final String PROCEDURE_WAL_SUFFIX = ".proc.wal";
+
   private final ProcedureFactory procedureFactory = ProcedureFactory.getInstance();
   private final String procedureWalDir =
       CommonDescriptor.getInstance().getConfig().getProcedureWalFolder();
   private final ConcurrentHashMap<Long, ProcedureWAL> procWALMap = new ConcurrentHashMap<>();
 
   public void load(List<Procedure> procedureList) {
     try (Stream<Path> s = Files.list(Paths.get(procedureWalDir))) {
-      s.filter(path -> path.getFileName().toString().endsWith(ProcedureStore.PROCEDURE_WAL_SUFFIX))
+      s.filter(path -> path.getFileName().toString().endsWith(PROCEDURE_WAL_SUFFIX))
           .sorted(
               (p1, p2) ->
                   Long.compareUnsigned(
@@ -74,7 +75,7 @@ public void load(List<Procedure> procedureList) {
   public TSStatus updateProcedure(UpdateProcedurePlan updateProcedurePlan) {
     Procedure procedure = updateProcedurePlan.getProcedure();
     long procId = procedure.getProcId();
-    Path path = Paths.get(procedureWalDir, procId + ProcedureStore.PROCEDURE_WAL_SUFFIX);
+    Path path = Paths.get(procedureWalDir, procId + PROCEDURE_WAL_SUFFIX);
     ProcedureWAL procedureWAL =
         procWALMap.computeIfAbsent(procId, id -> new ProcedureWAL(path, procedureFactory));
     try {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java

@@ -52,6 +52,7 @@ public abstract class Procedure<Env> implements Comparable<Procedure<Env>> {
   private static final Logger LOG = LoggerFactory.getLogger(Procedure.class);
   public static final long NO_PROC_ID = -1;
   public static final long NO_TIMEOUT = -1;
+  private boolean isDeserialized = false;
 
   private long parentProcId = NO_PROC_ID;
   private long rootProcId = NO_PROC_ID;
@@ -188,6 +189,8 @@ public void serialize(DataOutputStream stream) throws IOException {
   public void deserialize(ByteBuffer byteBuffer) {
     // procid
     this.setProcId(byteBuffer.getLong());
+    // isDeserialized
+    this.setDeserialized(true);
     // state
     this.setState(ProcedureState.values()[byteBuffer.getInt()]);
     //  submit time
@@ -539,6 +542,10 @@ public long getProcId() {
     return procId;
   }
 
+  public boolean isDeserialized() {
+    return isDeserialized;
+  }
+
   public boolean hasParent() {
     return parentProcId != NO_PROC_ID;
   }
@@ -564,6 +571,10 @@ protected void setProcId(long procId) {
     this.procId = procId;
   }
 
+  private void setDeserialized(boolean isDeserialized) {
+    this.isDeserialized = isDeserialized;
+  }
+
   public void setProcRunnable() {
     this.submittedTime = System.currentTimeMillis();
     setState(ProcedureState.RUNNABLE);

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java

@@ -20,6 +20,7 @@
 package org.apache.iotdb.confignode.procedure;
 
 import org.apache.iotdb.commons.concurrent.ThreadName;
+import org.apache.iotdb.commons.utils.TestOnly;
 import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
 import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedException;
 import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException;
@@ -88,6 +89,7 @@ public ProcedureExecutor(
     this.lastProcId.incrementAndGet();
   }
 
+  @TestOnly
   public ProcedureExecutor(final Env environment, final IProcedureStore store) {
     this(environment, store, new SimpleProcedureScheduler());
   }