This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new d4fd2de8c90 IT framework supports restarting cluster (#12022)
d4fd2de8c90 is described below
commit d4fd2de8c906f4a0b156f955618905dea265bf1f
Author: YuFengLiu <[email protected]>
AuthorDate: Tue Feb 6 18:13:01 2024 +0800
IT framework supports restarting cluster (#12022)
---
.../it/env/cluster/config/MppCommonConfig.java | 6 +
.../env/cluster/config/MppSharedCommonConfig.java | 7 +
.../iotdb/it/env/cluster/env/AbstractEnv.java | 199 +++++++------
.../it/env/remote/config/RemoteCommonConfig.java | 5 +
.../iotdb/it/env/remote/env/RemoteServerEnv.java | 20 ++
.../java/org/apache/iotdb/itbase/env/BaseEnv.java | 24 +-
.../org/apache/iotdb/itbase/env/CommonConfig.java | 2 +
.../it/cluster/IoTDBClusterRestartIT.java | 43 +--
.../org/apache/iotdb/db/it/IoTDBRecoverIT.java | 307 +++++++--------------
.../apache/iotdb/db/it/IoTDBRecoverUnclosedIT.java | 167 +++++------
.../org/apache/iotdb/db/it/utils/TestUtils.java | 21 +-
.../apache/iotdb/pipe/it/IoTDBPipeClusterIT.java | 31 +--
.../apache/iotdb/pipe/it/IoTDBPipeLifeCycleIT.java | 25 +-
.../apache/iotdb/db/utils/EnvironmentUtils.java | 21 +-
14 files changed, 383 insertions(+), 495 deletions(-)
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java
index e1b16ee6e7d..496e9e5e23f 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java
+++
b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java
@@ -420,6 +420,12 @@ public class MppCommonConfig extends MppBaseConfig
implements CommonConfig {
return this;
}
+ @Override
+ public CommonConfig setWalMode(String walMode) {
+ setProperty("wal_mode", walMode);
+ return this;
+ }
+
// For part of the log directory
public String getClusterConfigStr() {
return
fromConsensusFullNameToAbbr(properties.getProperty(CONFIG_NODE_CONSENSUS_PROTOCOL_CLASS))
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
index 3b478765c2c..cfbe1814afa 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
+++
b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
@@ -424,4 +424,11 @@ public class MppSharedCommonConfig implements CommonConfig
{
cnConfig.setDriverTaskExecutionTimeSliceInMs(driverTaskExecutionTimeSliceInMs);
return this;
}
+
+ @Override
+ public CommonConfig setWalMode(String walMode) {
+ dnConfig.setWalMode(walMode);
+ cnConfig.setWalMode(walMode);
+ return this;
+ }
}
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
index b96c8f4d4f9..b7e4c4dcb50 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
+++
b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
@@ -77,8 +77,7 @@ public abstract class AbstractEnv implements BaseEnv {
protected String testMethodName = null;
protected int index = 0;
protected long startTime;
- protected int testWorkingRetryCount = 30;
-
+ protected int retryCount = 30;
private IClientManager<TEndPoint, SyncConfigNodeIServiceClient>
clientManager;
/**
@@ -133,11 +132,11 @@ public abstract class AbstractEnv implements BaseEnv {
}
protected void initEnvironment(int configNodesNum, int dataNodesNum) {
- initEnvironment(configNodesNum, dataNodesNum, 30);
+ initEnvironment(configNodesNum, dataNodesNum, retryCount);
}
- protected void initEnvironment(int configNodesNum, int dataNodesNum, int
testWorkingRetryCount) {
- this.testWorkingRetryCount = testWorkingRetryCount;
+ protected void initEnvironment(int configNodesNum, int dataNodesNum, int
retryCount) {
+ this.retryCount = retryCount;
this.configNodeWrapperList = new ArrayList<>();
this.dataNodeWrapperList = new ArrayList<>();
@@ -245,7 +244,7 @@ public abstract class AbstractEnv implements BaseEnv {
throw new AssertionError();
}
- testWorkingNoUnknown();
+ checkClusterStatusWithoutUnknown();
}
public String getTestClassName() {
@@ -265,64 +264,34 @@ public abstract class AbstractEnv implements BaseEnv {
return result;
}
- public void testWorkingNoUnknown() {
- testWorking(nodeStatusMap ->
nodeStatusMap.values().stream().noneMatch("Unknown"::equals));
+ public boolean checkClusterStatusWithoutUnknown() {
+ return checkClusterStatus(
+ nodeStatusMap ->
nodeStatusMap.values().stream().noneMatch("Unknown"::equals))
+ && testJDBCConnection();
}
- public void testWorkingOneUnknownOtherRunning() {
- testWorking(
- nodeStatus -> {
- Map<String, Integer> count = countNodeStatus(nodeStatus);
- return count.getOrDefault("Unknown", 0) == 1
- && count.getOrDefault("Running", 0) == nodeStatus.size() - 1;
- });
+ public boolean checkClusterStatusOneUnknownOtherRunning() {
+ return checkClusterStatus(
+ nodeStatus -> {
+ Map<String, Integer> count = countNodeStatus(nodeStatus);
+ return count.getOrDefault("Unknown", 0) == 1
+ && count.getOrDefault("Running", 0) == nodeStatus.size() - 1;
+ })
+ && testJDBCConnection();
}
-
- public void testWorking(Predicate<Map<Integer, String>> statusCheck) {
- logger.info("Testing DataNode connection...");
- List<String> endpoints =
- dataNodeWrapperList.stream()
- .map(DataNodeWrapper::getIpAndPortString)
- .collect(Collectors.toList());
- RequestDelegate<Void> testDelegate =
- new ParallelRequestDelegate<>(endpoints, NODE_START_TIMEOUT);
- for (DataNodeWrapper dataNode : dataNodeWrapperList) {
- final String dataNodeEndpoint = dataNode.getIpAndPortString();
- testDelegate.addRequest(
- () -> {
- Exception lastException = null;
- for (int i = 0; i < testWorkingRetryCount; i++) {
- try (Connection ignored = getConnection(dataNodeEndpoint,
PROBE_TIMEOUT_MS)) {
- logger.info("Successfully connecting to DataNode: {}.",
dataNodeEndpoint);
- return null;
- } catch (Exception e) {
- lastException = e;
- TimeUnit.SECONDS.sleep(1L);
- }
- }
- throw lastException;
- });
- }
- try {
- long startTime = System.currentTimeMillis();
- testDelegate.requestAll();
- if (!configNodeWrapperList.isEmpty()) {
- checkNodeHeartbeat(statusCheck);
- }
- logger.info("Start cluster costs: {}s", (System.currentTimeMillis() -
startTime) / 1000.0);
- } catch (Exception e) {
- logger.error("exception in testWorking of ClusterID, message: {}",
e.getMessage(), e);
- throw new AssertionError(
- String.format("After %d times retry, the cluster can't work!",
testWorkingRetryCount));
- }
- }
-
- private void checkNodeHeartbeat(Predicate<Map<Integer, String>> statusCheck)
throws Exception {
+ /**
+ * Returns whether the all nodes' status all match the provided predicate.
check nodes with RPC
+ *
+ * @param statusCheck the predicate to test the status of nodes
+ * @return {@code true} if all nodes' status of the cluster match the
provided predicate,
+ * otherwise {@code false}
+ */
+ public boolean checkClusterStatus(Predicate<Map<Integer, String>>
statusCheck) {
logger.info("Testing cluster environment...");
TShowClusterResp showClusterResp;
Exception lastException = null;
boolean flag;
- for (int i = 0; i < 30; i++) {
+ for (int i = 0; i < retryCount; i++) {
try (SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient) getLeaderConfigNodeConnection()) {
flag = true;
@@ -347,18 +316,26 @@ public abstract class AbstractEnv implements BaseEnv {
if (flag) {
logger.info("The cluster is now ready for testing!");
- return;
+ return true;
}
} catch (Exception e) {
lastException = e;
}
- TimeUnit.SECONDS.sleep(1L);
+ try {
+ TimeUnit.SECONDS.sleep(1L);
+ } catch (InterruptedException e) {
+ lastException = e;
+ Thread.currentThread().interrupt();
+ }
}
-
if (lastException != null) {
- throw lastException;
+ logger.error(
+ "exception in testWorking of ClusterID, message: {}",
+ lastException.getMessage(),
+ lastException);
}
- throw new Exception("Check not pass");
+ logger.info("checkNodeHeartbeat failed after {} retries", retryCount);
+ return false;
}
@Override
@@ -392,18 +369,6 @@ public abstract class AbstractEnv implements BaseEnv {
getReadConnections(null, username, password));
}
- private Connection getConnection(String endpoint, int queryTimeout) throws
SQLException {
- IoTDBConnection connection =
- (IoTDBConnection)
- DriverManager.getConnection(
- Config.IOTDB_URL_PREFIX + endpoint + getParam(null,
queryTimeout),
- System.getProperty("User", "root"),
- System.getProperty("Password", "root"));
- connection.setQueryTimeout(queryTimeout);
-
- return connection;
- }
-
@Override
public Connection getConnection(Constant.Version version, String username,
String password)
throws SQLException {
@@ -519,6 +484,54 @@ public abstract class AbstractEnv implements BaseEnv {
return readConnRequestDelegate.requestAll();
}
+ // use this to avoid some runtimeExceptions when try to get jdbc connections.
+ // because it is hard to add retry and handle exception when getting jdbc
connections in
+ // getWriteConnectionWithSpecifiedDataNode and getReadConnections.
+ // so use this function to add retry when cluster is ready.
+ protected boolean testJDBCConnection() {
+ logger.info("Testing JDBC connection...");
+ List<String> endpoints =
+ dataNodeWrapperList.stream()
+ .map(DataNodeWrapper::getIpAndPortString)
+ .collect(Collectors.toList());
+ RequestDelegate<Void> testDelegate =
+ new ParallelRequestDelegate<>(endpoints, NODE_START_TIMEOUT);
+ for (DataNodeWrapper dataNode : dataNodeWrapperList) {
+ final String dataNodeEndpoint = dataNode.getIpAndPortString();
+ testDelegate.addRequest(
+ () -> {
+ Exception lastException = null;
+ for (int i = 0; i < retryCount; i++) {
+ try (IoTDBConnection ignored =
+ (IoTDBConnection)
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX
+ + dataNodeEndpoint
+ + getParam(null, NODE_NETWORK_TIMEOUT_MS),
+ System.getProperty("User", "root"),
+ System.getProperty("Password", "root"))) {
+ logger.info("Successfully connecting to DataNode: {}.",
dataNodeEndpoint);
+ return null;
+ } catch (Exception e) {
+ lastException = e;
+ TimeUnit.SECONDS.sleep(1L);
+ }
+ }
+ if (lastException != null) {
+ throw lastException;
+ }
+ return null;
+ });
+ }
+ try {
+ testDelegate.requestAll();
+ } catch (Exception e) {
+ logger.error("Failed to connect to DataNode", e);
+ return false;
+ }
+ return true;
+ }
+
private String getParam(Constant.Version version, int timeout) {
StringBuilder sb = new StringBuilder("?");
sb.append(Config.NETWORK_TIMEOUT).append("=").append(timeout);
@@ -570,7 +583,7 @@ public abstract class AbstractEnv implements BaseEnv {
throws IOException, InterruptedException {
Exception lastException = null;
ConfigNodeWrapper lastErrorNode = null;
- for (int i = 0; i < 30; i++) {
+ for (int i = 0; i < retryCount; i++) {
for (ConfigNodeWrapper configNodeWrapper : configNodeWrapperList) {
try {
lastErrorNode = configNodeWrapper;
@@ -632,7 +645,7 @@ public abstract class AbstractEnv implements BaseEnv {
public int getLeaderConfigNodeIndex() throws IOException,
InterruptedException {
Exception lastException = null;
ConfigNodeWrapper lastErrorNode = null;
- for (int retry = 0; retry < 30; retry++) {
+ for (int retry = 0; retry < retryCount; retry++) {
for (int configNodeId = 0; configNodeId < configNodeWrapperList.size();
configNodeId++) {
ConfigNodeWrapper configNodeWrapper =
configNodeWrapperList.get(configNodeId);
lastErrorNode = configNodeWrapper;
@@ -675,11 +688,25 @@ public abstract class AbstractEnv implements BaseEnv {
configNodeWrapperList.get(index).start();
}
+ @Override
+ public void startAllConfigNodes() {
+ for (ConfigNodeWrapper configNodeWrapper : configNodeWrapperList) {
+ configNodeWrapper.start();
+ }
+ }
+
@Override
public void shutdownConfigNode(int index) {
configNodeWrapperList.get(index).stop();
}
+ @Override
+ public void shutdownAllConfigNodes() {
+ for (ConfigNodeWrapper configNodeWrapper : configNodeWrapperList) {
+ configNodeWrapper.stop();
+ }
+ }
+
@Override
public ConfigNodeWrapper getConfigNodeWrapper(int index) {
return configNodeWrapperList.get(index);
@@ -765,7 +792,7 @@ public abstract class AbstractEnv implements BaseEnv {
if (isNeedVerify) {
// Test whether register success
- testWorkingNoUnknown();
+ checkClusterStatusWithoutUnknown();
}
}
@@ -790,7 +817,7 @@ public abstract class AbstractEnv implements BaseEnv {
if (isNeedVerify) {
// Test whether register success
- testWorkingNoUnknown();
+ checkClusterStatusWithoutUnknown();
}
}
@@ -799,16 +826,30 @@ public abstract class AbstractEnv implements BaseEnv {
dataNodeWrapperList.get(index).start();
}
+ @Override
+ public void startAllDataNodes() {
+ for (DataNodeWrapper dataNodeWrapper : dataNodeWrapperList) {
+ dataNodeWrapper.start();
+ }
+ }
+
@Override
public void shutdownDataNode(int index) {
dataNodeWrapperList.get(index).stop();
}
+ @Override
+ public void shutdownAllDataNodes() {
+ for (DataNodeWrapper dataNodeWrapper : dataNodeWrapperList) {
+ dataNodeWrapper.stop();
+ }
+ }
+
@Override
public void ensureNodeStatus(List<BaseNodeWrapper> nodes, List<NodeStatus>
targetStatus)
throws IllegalStateException {
Throwable lastException = null;
- for (int i = 0; i < 30; i++) {
+ for (int i = 0; i < retryCount; i++) {
try (SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient)
EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
List<String> errorMessages = new ArrayList<>(nodes.size());
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java
index f7ee0789b77..0495edb933e 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java
+++
b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java
@@ -298,4 +298,9 @@ public class RemoteCommonConfig implements CommonConfig {
public CommonConfig setDriverTaskExecutionTimeSliceInMs(long
driverTaskExecutionTimeSliceInMs) {
return this;
}
+
+ @Override
+ public CommonConfig setWalMode(String walMode) {
+ return this;
+ }
}
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
index 15f07a14370..25a9ac319ca 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
+++
b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
@@ -244,11 +244,21 @@ public class RemoteServerEnv implements BaseEnv {
throw new UnsupportedOperationException();
}
+ @Override
+ public void startAllConfigNodes() {
+ throw new UnsupportedOperationException();
+ }
+
@Override
public void shutdownConfigNode(int index) {
throw new UnsupportedOperationException();
}
+ @Override
+ public void shutdownAllConfigNodes() {
+ throw new UnsupportedOperationException();
+ }
+
@Override
public void ensureNodeStatus(List<BaseNodeWrapper> nodes, List<NodeStatus>
targetStatus) {
throw new UnsupportedOperationException();
@@ -299,11 +309,21 @@ public class RemoteServerEnv implements BaseEnv {
throw new UnsupportedOperationException();
}
+ @Override
+ public void startAllDataNodes() {
+ throw new UnsupportedOperationException();
+ }
+
@Override
public void shutdownDataNode(int index) {
throw new UnsupportedOperationException();
}
+ @Override
+ public void shutdownAllDataNodes() {
+ throw new UnsupportedOperationException();
+ }
+
@Override
public int getMqttPort() {
throw new UnsupportedOperationException();
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java
index ebdeb6a1875..120b0c1a810 100644
--- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java
+++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java
@@ -52,10 +52,6 @@ public interface BaseEnv {
*/
void initClusterEnvironment(int configNodesNum, int dataNodesNum);
- default void addClusterDataNodes(int dataNodesNum) throws IOException,
InterruptedException {
- throw new UnsupportedOperationException();
- }
-
/**
* Init a cluster with the specified number of ConfigNodes and DataNodes.
*
@@ -131,10 +127,6 @@ public interface BaseEnv {
IConfigNodeRPCService.Iface getLeaderConfigNodeConnection()
throws ClientManagerException, IOException, InterruptedException;
- default IConfigNodeRPCService.Iface getConfigNodeConnection(int index)
throws Exception {
- throw new UnsupportedOperationException();
- }
-
ISessionPool getSessionPool(int maxSize);
ISession getSessionConnection() throws IoTDBConnectionException;
@@ -150,12 +142,22 @@ public interface BaseEnv {
*/
int getLeaderConfigNodeIndex() throws IOException, InterruptedException;
+ default IConfigNodeRPCService.Iface getConfigNodeConnection(int index)
throws Exception {
+ throw new UnsupportedOperationException();
+ }
+
/** Start an existed ConfigNode. */
void startConfigNode(int index);
+ /** Start all existed ConfigNodes. */
+ void startAllConfigNodes();
+
/** Shutdown an existed ConfigNode. */
void shutdownConfigNode(int index);
+ /** Shutdown all existed ConfigNodes. */
+ void shutdownAllConfigNodes();
+
/**
* Ensure all the nodes being in the corresponding status.
*
@@ -211,9 +213,15 @@ public interface BaseEnv {
/** Start an existed DataNode. */
void startDataNode(int index);
+ /** Start all existed DataNodes. */
+ void startAllDataNodes();
+
/** Shutdown an existed DataNode. */
void shutdownDataNode(int index);
+ /** Shutdown all existed DataNodes. */
+ void shutdownAllDataNodes();
+
int getMqttPort();
String getIP();
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java
index 3bc7407470e..fa0e6992704 100644
---
a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java
+++
b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java
@@ -133,4 +133,6 @@ public interface CommonConfig {
CommonConfig setPipeAirGapReceiverEnabled(boolean
isPipeAirGapReceiverEnabled);
CommonConfig setDriverTaskExecutionTimeSliceInMs(long
driverTaskExecutionTimeSliceInMs);
+
+ CommonConfig setWalMode(String walMode);
}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterRestartIT.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterRestartIT.java
index 8e5e17b17bf..f05274f1289 100644
---
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterRestartIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterRestartIT.java
@@ -93,36 +93,18 @@ public class IoTDBClusterRestartIT {
}
@Test
- public void clusterRestartTest() throws InterruptedException {
+ public void clusterRestartTest() {
// Shutdown all cluster nodes
- for (int i = 0; i < testConfigNodeNum; i++) {
- EnvFactory.getEnv().shutdownConfigNode(i);
- }
- for (int i = 0; i < testDataNodeNum; i++) {
- EnvFactory.getEnv().shutdownDataNode(i);
- }
-
- // Sleep 1s before restart
- TimeUnit.SECONDS.sleep(1);
+ logger.info("Shutting down all ConfigNodes and DataNodes...");
+ EnvFactory.getEnv().shutdownAllConfigNodes();
+ EnvFactory.getEnv().shutdownAllDataNodes();
// Restart all cluster nodes
logger.info("Restarting all ConfigNodes...");
- for (int i = 0; i < testConfigNodeNum; i++) {
- EnvFactory.getEnv().startConfigNode(i);
- }
- try (SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient)
EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- // Do noting, just try to connect to the ConfigNode-leader
- // in order to ensure the ConfigNode-leader is ready
- } catch (Exception e) {
- logger.error("Failed to get ConfigNode-leader connection", e);
- }
+ EnvFactory.getEnv().startAllConfigNodes();
logger.info("Restarting all DataNodes...");
- for (int i = 0; i < testDataNodeNum; i++) {
- EnvFactory.getEnv().startDataNode(i);
- }
-
- ((AbstractEnv) EnvFactory.getEnv()).testWorkingNoUnknown();
+ EnvFactory.getEnv().startAllDataNodes();
+ Assert.assertTrue(((AbstractEnv)
EnvFactory.getEnv()).checkClusterStatusWithoutUnknown());
}
@Test
@@ -168,9 +150,7 @@ public class IoTDBClusterRestartIT {
dataPartitionTableResp.getDataPartitionTable());
// Shutdown all DataNodes
- for (int i = 0; i < testDataNodeNum; i++) {
- EnvFactory.getEnv().shutdownDataNode(i);
- }
+ EnvFactory.getEnv().shutdownAllDataNodes();
TimeUnit.SECONDS.sleep(1);
List<DataNodeWrapper> dataNodeWrapperList =
EnvFactory.getEnv().getDataNodeWrapperList();
@@ -246,11 +226,10 @@ public class IoTDBClusterRestartIT {
for (int i = 1; i < testConfigNodeNum; i++) {
EnvFactory.getEnv().startConfigNode(i);
}
- for (int i = 0; i < testDataNodeNum; i++) {
- EnvFactory.getEnv().startDataNode(i);
- }
+ EnvFactory.getEnv().startAllDataNodes();
logger.info("Restarted");
- ((AbstractEnv) EnvFactory.getEnv()).testWorkingOneUnknownOtherRunning();
+ Assert.assertTrue(
+ ((AbstractEnv)
EnvFactory.getEnv()).checkClusterStatusOneUnknownOtherRunning());
logger.info("Working without Seed-ConfigNode");
}
}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRecoverIT.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRecoverIT.java
index c93ebde8f59..2b6c558618d 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRecoverIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRecoverIT.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.it;
import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.env.AbstractEnv;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.itbase.category.ClusterIT;
import org.apache.iotdb.itbase.category.LocalStandaloneIT;
@@ -27,14 +28,14 @@ import org.apache.iotdb.itbase.category.LocalStandaloneIT;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
-import org.junit.Ignore;
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.sql.Connection;
import java.sql.ResultSet;
-import java.sql.SQLException;
import java.sql.Statement;
import java.util.Locale;
@@ -45,14 +46,15 @@ import static
org.apache.iotdb.db.utils.constant.TestConstant.minValue;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.fail;
-@Ignore
@RunWith(IoTDBTestRunner.class)
@Category({LocalStandaloneIT.class, ClusterIT.class})
public class IoTDBRecoverIT {
+ private static final Logger logger =
LoggerFactory.getLogger(IoTDBRecoverIT.class);
+
private static final String TIMESTAMP_STR = "Time";
private static final String TEMPERATURE_STR =
"root.ln.wf01.wt01.temperature";
- private static String[] creationSqls =
+ private static final String[] creationSqls =
new String[] {
"CREATE DATABASE root.vehicle.d0",
"CREATE DATABASE root.vehicle.d1",
@@ -62,7 +64,7 @@ public class IoTDBRecoverIT {
"CREATE TIMESERIES root.vehicle.d0.s3 WITH DATATYPE=TEXT,
ENCODING=PLAIN",
"CREATE TIMESERIES root.vehicle.d0.s4 WITH DATATYPE=BOOLEAN,
ENCODING=PLAIN"
};
- private static String[] dataSet2 =
+ private static final String[] dataSet2 =
new String[] {
"CREATE DATABASE root.ln.wf01.wt01",
"CREATE TIMESERIES root.ln.wf01.wt01.status WITH DATATYPE=BOOLEAN,
ENCODING=PLAIN",
@@ -83,8 +85,6 @@ public class IoTDBRecoverIT {
private final String d0s1 = "root.vehicle.d0.s1";
private final String d0s2 = "root.vehicle.d0.s2";
private final String d0s3 = "root.vehicle.d0.s3";
- private String insertTemplate =
- "INSERT INTO root.vehicle.d0(timestamp,s0,s1,s2,s3,s4)" + "
VALUES(%d,%d,%d,%f,%s,%s)";
@Before
public void setUp() throws Exception {
@@ -98,53 +98,51 @@ public class IoTDBRecoverIT {
}
@Test
- public void mergeTest() {
+ public void RecoverTest1() {
+ // stop cluster
+ EnvFactory.getEnv().shutdownAllDataNodes();
+ logger.info("All DataNodes are shut down");
+ EnvFactory.getEnv().shutdownAllConfigNodes();
+ logger.info("All ConfigNodes are shut down");
+ EnvFactory.getEnv().startAllConfigNodes();
+ logger.info("All ConfigNodes are started");
+ EnvFactory.getEnv().startAllDataNodes();
+ logger.info("All DataNodes are started");
+ // check cluster whether restart
+ Assert.assertTrue(((AbstractEnv)
EnvFactory.getEnv()).checkClusterStatusWithoutUnknown());
String[] retArray = new String[] {"0,2", "0,4", "0,3"};
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
String selectSql = "select count(temperature) from root.ln.wf01.wt01
where time > 3";
- int cnt;
try (ResultSet resultSet = statement.executeQuery(selectSql)) {
assertNotNull(resultSet);
- cnt = 0;
- while (resultSet.next()) {
- String ans =
- resultSet.getString(TIMESTAMP_STR)
- + ","
- + resultSet.getString(count(TEMPERATURE_STR));
- Assert.assertEquals(retArray[cnt], ans);
- cnt++;
- }
- Assert.assertEquals(1, cnt);
+ resultSet.next();
+ String ans =
+ resultSet.getString(TIMESTAMP_STR) + "," +
resultSet.getString(count(TEMPERATURE_STR));
+ Assert.assertEquals(retArray[0], ans);
}
selectSql = "select min_time(temperature) from root.ln.wf01.wt01 where
time > 3";
try (ResultSet resultSet = statement.executeQuery(selectSql)) {
assertNotNull(resultSet);
- while (resultSet.next()) {
- String ans =
- resultSet.getString(TIMESTAMP_STR)
- + ","
- + resultSet.getString(minTime(TEMPERATURE_STR));
- Assert.assertEquals(retArray[cnt], ans);
- cnt++;
- }
- Assert.assertEquals(2, cnt);
+ resultSet.next();
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(minTime(TEMPERATURE_STR));
+ Assert.assertEquals(retArray[1], ans);
}
selectSql = "select min_time(temperature) from root.ln.wf01.wt01 where
temperature > 3";
try (ResultSet resultSet = statement.executeQuery(selectSql)) {
assertNotNull(resultSet);
- while (resultSet.next()) {
- String ans =
- resultSet.getString(TIMESTAMP_STR)
- + ","
- + resultSet.getString(minTime(TEMPERATURE_STR));
- Assert.assertEquals(retArray[cnt], ans);
- cnt++;
- }
- Assert.assertEquals(3, cnt);
+ resultSet.next();
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(minTime(TEMPERATURE_STR));
+ Assert.assertEquals(retArray[2], ans);
}
} catch (Exception e) {
@@ -152,85 +150,7 @@ public class IoTDBRecoverIT {
fail(e.getMessage());
}
- // we want to recover
- // TODO: replace stopDaemon() and activeDaemon() with new methods in Env.
- // EnvironmentUtils.stopDaemon();
- // wait for close
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- Thread.currentThread().interrupt();
- }
- // EnvironmentUtils.activeDaemon();
-
- // count test
- retArray = new String[] {"0,2001,2001,2001,2001", "0,7500,7500,7500,7500"};
- try (Connection connection = EnvFactory.getEnv().getConnection();
- Statement statement = connection.createStatement()) {
-
- String selectSql =
- "select count(s0),count(s1),count(s2),count(s3) "
- + "from root.vehicle.d0 where time >= 6000 and time <= 9000";
- int cnt;
- try (ResultSet resultSet = statement.executeQuery(selectSql)) {
- assertNotNull(resultSet);
- cnt = 0;
- while (resultSet.next()) {
- String ans =
- resultSet.getString(TIMESTAMP_STR)
- + ","
- + resultSet.getString(count(d0s0))
- + ","
- + resultSet.getString(count(d0s1))
- + ","
- + resultSet.getString(count(d0s2))
- + ","
- + resultSet.getString(count(d0s3));
- Assert.assertEquals(retArray[cnt], ans);
- cnt++;
- }
- Assert.assertEquals(1, cnt);
- }
-
- selectSql = "select count(s0),count(s1),count(s2),count(s3) " + "from
root.vehicle.d0";
- try (ResultSet resultSet = statement.executeQuery(selectSql)) {
- assertNotNull(resultSet);
- while (resultSet.next()) {
- String ans =
- resultSet.getString(TIMESTAMP_STR)
- + ","
- + resultSet.getString(count(d0s0))
- + ","
- + resultSet.getString(count(d0s1))
- + ","
- + resultSet.getString(count(d0s2))
- + ","
- + resultSet.getString(count(d0s3));
- Assert.assertEquals(retArray[cnt], ans);
- cnt++;
- }
- Assert.assertEquals(2, cnt);
- }
- } catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
-
- // we want to recover
- // EnvironmentUtils.stopDaemon();
- // wait for close
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- Thread.currentThread().interrupt();
- }
-
- // EnvironmentUtils.activeDaemon();
-
- // maxminValueTest
-
+ // max min ValueTest
retArray = new String[] {"0,8499,500.0", "0,2499,500.0"};
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
@@ -238,37 +158,29 @@ public class IoTDBRecoverIT {
String selectSql =
"select max_value(s0),min_value(s2) "
+ "from root.vehicle.d0 where time >= 100 and time < 9000";
- int cnt;
try (ResultSet resultSet = statement.executeQuery(selectSql)) {
assertNotNull(resultSet);
- cnt = 0;
- while (resultSet.next()) {
- String ans =
- resultSet.getString(TIMESTAMP_STR)
- + ","
- + resultSet.getString(maxValue(d0s0))
- + ","
- + resultSet.getString(minValue(d0s2));
- Assert.assertEquals(retArray[cnt], ans);
- cnt++;
- }
- Assert.assertEquals(1, cnt);
+ resultSet.next();
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(maxValue(d0s0))
+ + ","
+ + resultSet.getString(minValue(d0s2));
+ Assert.assertEquals(retArray[0], ans);
}
selectSql = "select max_value(s0),min_value(s2) from root.vehicle.d0
where time < 2500";
try (ResultSet resultSet = statement.executeQuery(selectSql)) {
assertNotNull(resultSet);
- while (resultSet.next()) {
- String ans =
- resultSet.getString(TIMESTAMP_STR)
- + ","
- + resultSet.getString(maxValue(d0s0))
- + ","
- + resultSet.getString(minValue(d0s2));
- Assert.assertEquals(retArray[cnt], ans);
- cnt++;
- }
- Assert.assertEquals(2, cnt);
+ resultSet.next();
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(maxValue(d0s0))
+ + ","
+ + resultSet.getString(minValue(d0s2));
+ Assert.assertEquals(retArray[1], ans);
}
} catch (Exception e) {
e.printStackTrace();
@@ -277,29 +189,18 @@ public class IoTDBRecoverIT {
}
@Test
- public void vmTest() throws SQLException {
- try (Connection connection = EnvFactory.getEnv().getConnection();
- Statement statement = connection.createStatement()) {
- // prepare more data to flush
- for (int i = 2000; i < 2500; i++) {
- statement.execute(
- String.format(
- Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i +
"'", "false"));
- }
- statement.execute("flush");
- }
-
- // we want to recover
- // EnvironmentUtils.stopDaemon();
- // wait for close
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- Thread.currentThread().interrupt();
- }
- // EnvironmentUtils.activeDaemon();
-
+ public void recoverTest2() {
+ // stop cluster
+ EnvFactory.getEnv().shutdownAllDataNodes();
+ logger.info("All DataNodes are shut down");
+ EnvFactory.getEnv().shutdownAllConfigNodes();
+ logger.info("All ConfigNodes are shut down");
+ EnvFactory.getEnv().startAllConfigNodes();
+ logger.info("All ConfigNodes are started");
+ EnvFactory.getEnv().startAllDataNodes();
+ logger.info("All DataNodes are started");
+ // wait for cluster to start and check
+ Assert.assertTrue(((AbstractEnv)
EnvFactory.getEnv()).checkClusterStatusWithoutUnknown());
// count test
String[] retArray = new String[] {"0,2001,2001,2001,2001",
"0,7500,7500,7500,7500"};
try (Connection connection = EnvFactory.getEnv().getConnection();
@@ -308,45 +209,37 @@ public class IoTDBRecoverIT {
String selectSql =
"select count(s0),count(s1),count(s2),count(s3) "
+ "from root.vehicle.d0 where time >= 6000 and time <= 9000";
- int cnt;
try (ResultSet resultSet = statement.executeQuery(selectSql)) {
assertNotNull(resultSet);
- cnt = 0;
- while (resultSet.next()) {
- String ans =
- resultSet.getString(TIMESTAMP_STR)
- + ","
- + resultSet.getString(count(d0s0))
- + ","
- + resultSet.getString(count(d0s1))
- + ","
- + resultSet.getString(count(d0s2))
- + ","
- + resultSet.getString(count(d0s3));
- Assert.assertEquals(retArray[cnt], ans);
- cnt++;
- }
- Assert.assertEquals(1, cnt);
+ resultSet.next();
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(count(d0s0))
+ + ","
+ + resultSet.getString(count(d0s1))
+ + ","
+ + resultSet.getString(count(d0s2))
+ + ","
+ + resultSet.getString(count(d0s3));
+ Assert.assertEquals(retArray[0], ans);
}
selectSql = "select count(s0),count(s1),count(s2),count(s3) from
root.vehicle.d0";
try (ResultSet resultSet = statement.executeQuery(selectSql)) {
assertNotNull(resultSet);
- while (resultSet.next()) {
- String ans =
- resultSet.getString(TIMESTAMP_STR)
- + ","
- + resultSet.getString(count(d0s0))
- + ","
- + resultSet.getString(count(d0s1))
- + ","
- + resultSet.getString(count(d0s2))
- + ","
- + resultSet.getString(count(d0s3));
- Assert.assertEquals(retArray[cnt], ans);
- cnt++;
- }
- Assert.assertEquals(2, cnt);
+ resultSet.next();
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(count(d0s0))
+ + ","
+ + resultSet.getString(count(d0s1))
+ + ","
+ + resultSet.getString(count(d0s2))
+ + ","
+ + resultSet.getString(count(d0s3));
+ Assert.assertEquals(retArray[1], ans);
}
} catch (Exception e) {
e.printStackTrace();
@@ -367,45 +260,53 @@ public class IoTDBRecoverIT {
}
// prepare BufferWrite file
+ String insertTemplate =
+ "INSERT INTO root.vehicle.d0(timestamp,s0,s1,s2,s3,s4)" + "
VALUES(%d,%d,%d,%f,%s,%s)";
for (int i = 5000; i < 7000; i++) {
- statement.execute(
+ statement.addBatch(
String.format(
Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i +
"'", "true"));
}
+ statement.executeBatch();
statement.execute("flush");
for (int i = 7500; i < 8500; i++) {
- statement.execute(
+ statement.addBatch(
String.format(
Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i +
"'", "false"));
}
+ statement.executeBatch();
statement.execute("flush");
// prepare Unseq-File
for (int i = 500; i < 1500; i++) {
- statement.execute(
+ statement.addBatch(
String.format(
Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i +
"'", "true"));
}
+ statement.executeBatch();
statement.execute("flush");
for (int i = 3000; i < 6500; i++) {
- statement.execute(
+ statement.addBatch(
String.format(
Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i +
"'", "false"));
}
- statement.execute("merge");
+ statement.executeBatch();
+ statement.execute("flush");
// prepare BufferWrite cache
for (int i = 9000; i < 10000; i++) {
- statement.execute(
+ statement.addBatch(
String.format(
Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i +
"'", "true"));
}
+ statement.executeBatch();
// prepare Overflow cache
for (int i = 2000; i < 2500; i++) {
- statement.execute(
+ statement.addBatch(
String.format(
Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i +
"'", "false"));
}
-
+ statement.executeBatch();
+ statement.execute("flush");
} catch (Exception e) {
e.printStackTrace();
}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRecoverUnclosedIT.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRecoverUnclosedIT.java
index 48c7426614f..8478bf290da 100644
---
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRecoverUnclosedIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRecoverUnclosedIT.java
@@ -19,20 +19,20 @@
package org.apache.iotdb.db.it;
-import org.apache.iotdb.db.conf.IoTDBConfig;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode;
import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.env.AbstractEnv;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.itbase.category.ClusterIT;
import org.apache.iotdb.itbase.category.LocalStandaloneIT;
import org.junit.After;
+import org.junit.Assert;
import org.junit.Before;
-import org.junit.Ignore;
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.io.IOException;
import java.sql.Connection;
@@ -49,15 +49,13 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.fail;
-@Ignore
@RunWith(IoTDBTestRunner.class)
@Category({LocalStandaloneIT.class, ClusterIT.class})
public class IoTDBRecoverUnclosedIT {
- private static final IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
-
+ private static final Logger logger =
LoggerFactory.getLogger(IoTDBRecoverUnclosedIT.class);
private static final String TIMESTAMP_STR = "Time";
private static final String TEMPERATURE_STR =
"root.ln.wf01.wt01.temperature";
- private static String[] creationSqls =
+ private static final String[] creationSqls =
new String[] {
"CREATE DATABASE root.vehicle.d0",
"CREATE DATABASE root.vehicle.d1",
@@ -67,7 +65,7 @@ public class IoTDBRecoverUnclosedIT {
"CREATE TIMESERIES root.vehicle.d0.s3 WITH DATATYPE=TEXT,
ENCODING=PLAIN",
"CREATE TIMESERIES root.vehicle.d0.s4 WITH DATATYPE=BOOLEAN,
ENCODING=PLAIN"
};
- private static String[] dataSet2 =
+ private static final String[] dataSet2 =
new String[] {
"CREATE DATABASE root.ln.wf01.wt01",
"CREATE TIMESERIES root.ln.wf01.wt01.status WITH DATATYPE=BOOLEAN,
ENCODING=PLAIN",
@@ -84,26 +82,16 @@ public class IoTDBRecoverUnclosedIT {
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware)
"
+ "values(5, 5.5, false, 55)"
};
- private final String d0s0 = "root.vehicle.d0.s0";
- private final String d0s1 = "root.vehicle.d0.s1";
- private final String d0s2 = "root.vehicle.d0.s2";
- private final String d0s3 = "root.vehicle.d0.s3";
- private String insertTemplate =
- "INSERT INTO root.vehicle.d0(timestamp,s0,s1,s2,s3,s4)" + "
VALUES(%d,%d,%d,%f,%s,%s)";
-
- private WALMode prevWALMode;
@Before
public void setUp() throws Exception {
+ EnvFactory.getEnv().getConfig().getCommonConfig().setWalMode("SYNC");
EnvFactory.getEnv().initClusterEnvironment();
- prevWALMode = config.getWalMode();
- config.setWalMode(WALMode.SYNC);
prepareData();
}
@After
public void tearDown() throws Exception {
- config.setWalMode(prevWALMode);
EnvFactory.getEnv().cleanClusterEnvironment();
}
@@ -114,47 +102,34 @@ public class IoTDBRecoverUnclosedIT {
Statement statement = connection.createStatement()) {
String selectSql = "select count(temperature) from root.ln.wf01.wt01
where time > 3";
- int cnt;
try (ResultSet resultSet = statement.executeQuery(selectSql)) {
assertNotNull(resultSet);
- cnt = 0;
- while (resultSet.next()) {
- String ans =
- resultSet.getString(TIMESTAMP_STR)
- + ","
- + resultSet.getString(count(TEMPERATURE_STR));
- assertEquals(retArray[cnt], ans);
- cnt++;
- }
- assertEquals(1, cnt);
+ resultSet.next();
+ String ans =
+ resultSet.getString(TIMESTAMP_STR) + "," +
resultSet.getString(count(TEMPERATURE_STR));
+ assertEquals(retArray[0], ans);
}
selectSql = "select min_time(temperature) from root.ln.wf01.wt01 where
time > 3";
try (ResultSet resultSet = statement.executeQuery(selectSql)) {
assertNotNull(resultSet);
- while (resultSet.next()) {
- String ans =
- resultSet.getString(TIMESTAMP_STR)
- + ","
- + resultSet.getString(minTime(TEMPERATURE_STR));
- assertEquals(retArray[cnt], ans);
- cnt++;
- }
- assertEquals(2, cnt);
+ resultSet.next();
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(minTime(TEMPERATURE_STR));
+ assertEquals(retArray[1], ans);
}
selectSql = "select min_time(temperature) from root.ln.wf01.wt01 where
temperature > 3";
try (ResultSet resultSet = statement.executeQuery(selectSql)) {
assertNotNull(resultSet);
- while (resultSet.next()) {
- String ans =
- resultSet.getString(TIMESTAMP_STR)
- + ","
- + resultSet.getString(minTime(TEMPERATURE_STR));
- assertEquals(retArray[cnt], ans);
- cnt++;
- }
- assertEquals(3, cnt);
+ resultSet.next();
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(minTime(TEMPERATURE_STR));
+ assertEquals(retArray[2], ans);
}
} catch (Exception e) {
@@ -162,63 +137,60 @@ public class IoTDBRecoverUnclosedIT {
fail(e.getMessage());
}
- // TODO: replace restartDaemon() with new methods in Env.
- try {
- // EnvironmentUtils.restartDaemon();
- } catch (Exception e) {
- fail();
- }
insertMoreData();
- try {
- // EnvironmentUtils.restartDaemon();
- } catch (Exception e) {
- fail();
- }
- // test count, max, min value
- retArray = new String[] {"0,8499,500.0", "0,2499,500.0"};
+ // stop cluster
+ EnvFactory.getEnv().shutdownAllDataNodes();
+ logger.info("All DataNodes are shut down");
+ EnvFactory.getEnv().shutdownAllConfigNodes();
+ logger.info("All ConfigNodes are shut down");
+ EnvFactory.getEnv().startAllConfigNodes();
+ logger.info("All ConfigNodes are started");
+ EnvFactory.getEnv().startAllDataNodes();
+ logger.info("All DataNodes are started");
+ // wait for cluster to start and check
+ Assert.assertTrue(((AbstractEnv)
EnvFactory.getEnv()).checkClusterStatusWithoutUnknown());
+
+ // test count,
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
String selectSql = "select count(*) from root.vehicle.d0";
ResultSet tempResultSet = statement.executeQuery(selectSql);
assertNotNull(tempResultSet);
tempResultSet.next();
+ String d0s0 = "root.vehicle.d0.s0";
+ String d0s1 = "root.vehicle.d0.s1";
+ String d0s2 = "root.vehicle.d0.s2";
assertEquals(7500, tempResultSet.getInt("count(" + d0s0 + ")"));
+ // test max, min value
+ retArray = new String[] {"0,8499,500.0", "0,2499,500.0"};
selectSql =
"select max_value(s0),min_value(s2) "
+ "from root.vehicle.d0 where time >= 100 and time < 9000";
- int cnt;
try (ResultSet resultSet = statement.executeQuery(selectSql)) {
assertNotNull(resultSet);
- cnt = 0;
- while (resultSet.next()) {
- String ans =
- resultSet.getString(TIMESTAMP_STR)
- + ","
- + resultSet.getString(maxValue(d0s0))
- + ","
- + resultSet.getString(minValue(d0s2));
- assertEquals(retArray[cnt], ans);
- cnt++;
- }
- assertEquals(1, cnt);
+ resultSet.next();
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(maxValue(d0s0))
+ + ","
+ + resultSet.getString(minValue(d0s2));
+ assertEquals(retArray[0], ans);
}
- selectSql = "select max_value(s0),min_value(s2) from root.vehicle.d0
where time < 2500";
+ selectSql = "select max_value(s1),min_value(s2) from root.vehicle.d0
where time < 2500";
try (ResultSet resultSet = statement.executeQuery(selectSql)) {
assertNotNull(resultSet);
- while (resultSet.next()) {
- String ans =
- resultSet.getString(TIMESTAMP_STR)
- + ","
- + resultSet.getString(maxValue(d0s0))
- + ","
- + resultSet.getString(minValue(d0s2));
- assertEquals(retArray[cnt], ans);
- cnt++;
- }
- assertEquals(2, cnt);
+ resultSet.next();
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(maxValue(d0s1))
+ + ","
+ + resultSet.getString(minValue(d0s2));
+ assertEquals(retArray[1], ans);
}
} catch (Exception e) {
e.printStackTrace();
@@ -247,41 +219,46 @@ public class IoTDBRecoverUnclosedIT {
Statement statement = connection.createStatement()) {
// prepare BufferWrite file
+ String insertTemplate =
+ "INSERT INTO root.vehicle.d0(timestamp,s0,s1,s2,s3,s4)" + "
VALUES(%d,%d,%d,%f,%s,%s)";
for (int i = 5000; i < 7000; i++) {
- statement.execute(
+ statement.addBatch(
String.format(
Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i +
"'", "true"));
}
+ statement.executeBatch();
for (int i = 7500; i < 8500; i++) {
- statement.execute(
+ statement.addBatch(
String.format(
Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i +
"'", "false"));
}
+ statement.executeBatch();
// prepare Unseq-File
for (int i = 500; i < 1500; i++) {
- statement.execute(
+ statement.addBatch(
String.format(
Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i +
"'", "true"));
}
for (int i = 3000; i < 6500; i++) {
- statement.execute(
+ statement.addBatch(
String.format(
Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i +
"'", "false"));
}
-
+ statement.executeBatch();
// prepare BufferWrite cache
for (int i = 9000; i < 10000; i++) {
- statement.execute(
+ statement.addBatch(
String.format(
Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i +
"'", "true"));
}
+ statement.executeBatch();
// prepare Overflow cache
for (int i = 2000; i < 2500; i++) {
- statement.execute(
+ statement.addBatch(
String.format(
Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i +
"'", "false"));
}
-
+ statement.executeBatch();
} catch (Exception e) {
e.printStackTrace();
}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java
index 5e5c8bdd1cb..5062f6e9ee9 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java
@@ -631,21 +631,12 @@ public class TestUtils {
}
}
- public static void restartCluster(BaseEnv env) throws Exception {
- for (int i = 0; i < env.getConfigNodeWrapperList().size(); ++i) {
- env.shutdownConfigNode(i);
- }
- for (int i = 0; i < env.getDataNodeWrapperList().size(); ++i) {
- env.shutdownDataNode(i);
- }
- TimeUnit.SECONDS.sleep(1);
- for (int i = 0; i < env.getConfigNodeWrapperList().size(); ++i) {
- env.startConfigNode(i);
- }
- for (int i = 0; i < env.getDataNodeWrapperList().size(); ++i) {
- env.startDataNode(i);
- }
- ((AbstractEnv) env).testWorkingNoUnknown();
+ public static boolean restartCluster(BaseEnv env) {
+ env.shutdownAllDataNodes();
+ env.shutdownAllConfigNodes();
+ env.startAllConfigNodes();
+ env.startAllDataNodes();
+ return ((AbstractEnv) env).checkClusterStatusWithoutUnknown();
}
public static void assertDataOnEnv(
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/IoTDBPipeClusterIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/IoTDBPipeClusterIT.java
index 91a4c13d1f9..0639497a6b1 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/IoTDBPipeClusterIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/IoTDBPipeClusterIT.java
@@ -237,7 +237,7 @@ public class IoTDBPipeClusterIT extends AbstractPipeDualIT {
}
try {
senderEnv.startDataNode(i);
- ((AbstractEnv) senderEnv).testWorkingNoUnknown();
+ Assert.assertTrue(((AbstractEnv)
senderEnv).checkClusterStatusWithoutUnknown());
} catch (Exception e) {
e.printStackTrace();
return;
@@ -261,14 +261,8 @@ public class IoTDBPipeClusterIT extends AbstractPipeDualIT
{
"count(root.db.d1.s1),",
Collections.singleton("2,"));
}
-
- try {
- TestUtils.restartCluster(senderEnv);
- TestUtils.restartCluster(receiverEnv);
- } catch (Exception e) {
- e.printStackTrace();
- return;
- }
+ Assert.assertTrue(TestUtils.restartCluster(senderEnv));
+ Assert.assertTrue(TestUtils.restartCluster(receiverEnv));
try (SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient)
senderEnv.getLeaderConfigNodeConnection()) {
@@ -363,13 +357,8 @@ public class IoTDBPipeClusterIT extends AbstractPipeDualIT
{
Collections.singleton("2,"));
}
- try {
- TestUtils.restartCluster(senderEnv);
- TestUtils.restartCluster(receiverEnv);
- } catch (Exception e) {
- e.printStackTrace();
- return;
- }
+ Assert.assertTrue(TestUtils.restartCluster(senderEnv));
+ Assert.assertTrue(TestUtils.restartCluster(receiverEnv));
try (SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient)
senderEnv.getLeaderConfigNodeConnection()) {
@@ -637,7 +626,7 @@ public class IoTDBPipeClusterIT extends AbstractPipeDualIT {
senderEnv.startDataNode(senderEnv.getDataNodeWrapperList().size() - 1);
senderEnv.shutdownDataNode(senderEnv.getDataNodeWrapperList().size() -
1);
senderEnv.getDataNodeWrapperList().remove(senderEnv.getDataNodeWrapperList().size()
- 1);
- ((AbstractEnv) senderEnv).testWorkingNoUnknown();
+ Assert.assertTrue(((AbstractEnv)
senderEnv).checkClusterStatusWithoutUnknown());
} catch (Exception e) {
e.printStackTrace();
return;
@@ -695,13 +684,7 @@ public class IoTDBPipeClusterIT extends AbstractPipeDualIT
{
return;
}
- try {
- TestUtils.restartCluster(senderEnv);
- } catch (Exception e) {
- e.printStackTrace();
- return;
- }
-
+ Assert.assertTrue(TestUtils.restartCluster(senderEnv));
TestUtils.assertDataOnEnv(
receiverEnv,
"select count(*) from root.**",
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/IoTDBPipeLifeCycleIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/IoTDBPipeLifeCycleIT.java
index aadf05a9b6b..473b9ea8086 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/IoTDBPipeLifeCycleIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/IoTDBPipeLifeCycleIT.java
@@ -427,13 +427,8 @@ public class IoTDBPipeLifeCycleIT extends
AbstractPipeDualIT {
receiverEnv, "select * from root.**", "Time,root.db.d1.s1,",
expectedResSet);
}
- try {
- TestUtils.restartCluster(senderEnv);
- TestUtils.restartCluster(receiverEnv);
- } catch (Exception e) {
- e.printStackTrace();
- return;
- }
+ Assert.assertTrue(TestUtils.restartCluster(senderEnv));
+ Assert.assertTrue(TestUtils.restartCluster(receiverEnv));
try (SyncConfigNodeIServiceClient ignored =
(SyncConfigNodeIServiceClient)
senderEnv.getLeaderConfigNodeConnection()) {
@@ -496,12 +491,7 @@ public class IoTDBPipeLifeCycleIT extends
AbstractPipeDualIT {
});
t.start();
- try {
- TestUtils.restartCluster(receiverEnv);
- } catch (Exception e) {
- e.printStackTrace();
- return;
- }
+ Assert.assertTrue(TestUtils.restartCluster(receiverEnv));
t.join();
TestUtils.assertDataOnEnv(
@@ -674,13 +664,8 @@ public class IoTDBPipeLifeCycleIT extends
AbstractPipeDualIT {
TestUtils.assertDataOnEnv(
receiverEnv, "select * from root.**", "Time,root.db.d1.s1,",
expectedResSet);
- try {
- TestUtils.restartCluster(senderEnv);
- TestUtils.restartCluster(receiverEnv);
- } catch (Exception e) {
- e.printStackTrace();
- return;
- }
+ Assert.assertTrue(TestUtils.restartCluster(senderEnv));
+ Assert.assertTrue(TestUtils.restartCluster(receiverEnv));
for (int i = 400; i < 500; ++i) {
if (!TestUtils.tryExecuteNonQueryWithRetry(
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java
index e63162ed263..c433a28e773 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java
@@ -171,7 +171,7 @@ public class EnvironmentUtils {
private static boolean examinePorts() {
TTransport transport = TSocketWrapper.wrap(tConfiguration, "127.0.0.1",
6667, 100);
- if (!transport.isOpen()) {
+ if (transport != null && !transport.isOpen()) {
try {
transport.open();
logger.error("stop daemon failed. 6667 can be connected now.");
@@ -183,7 +183,7 @@ public class EnvironmentUtils {
}
// try sync service
transport = TSocketWrapper.wrap(tConfiguration, "127.0.0.1", 5555, 100);
- if (!transport.isOpen()) {
+ if (transport != null && !transport.isOpen()) {
try {
transport.open();
logger.error("stop Sync daemon failed. 5555 can be connected now.");
@@ -284,23 +284,6 @@ public class EnvironmentUtils {
TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID);
}
- public static void stopDaemon() {}
-
- public static void shutdownDaemon() throws Exception {}
-
- public static void activeDaemon() {}
-
- public static void reactiveDaemon() {}
-
- public static void restartDaemon() throws Exception {
- shutdownDaemon();
- stopDaemon();
- TsFileResourceManager.getInstance().clear();
- WALManager.getInstance().clear();
- WALRecoverManager.getInstance().clear();
- reactiveDaemon();
- }
-
private static void createAllDir() {
// create sequential files
for (String path : tierManager.getAllLocalSequenceFileFolders()) {
