Skip to content

Commit 89b77f7

Browse files
jt2594838jt2594838
committed
fix test
1 parent 060a5b6 commit 89b77f7

4 files changed

Lines changed: 84 additions & 22 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionOperationReliabilityITFramework.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@
4545

4646
import org.apache.thrift.TException;
4747
import org.apache.tsfile.read.common.Field;
48+
import org.apache.tsfile.utils.Pair;
4849
import org.awaitility.Awaitility;
4950
import org.awaitility.core.ConditionTimeoutException;
5051
import org.junit.After;
@@ -448,6 +449,29 @@ public static Map<Integer, Set<Integer>> getDataRegionMap(Statement statement) t
448449
return regionMap;
449450
}
450451

452+
public static Map<Integer, Pair<Integer, Set<Integer>>> getDataRegionMapWithLeader(
453+
Statement statement) throws Exception {
454+
ResultSet showRegionsResult = statement.executeQuery(SHOW_REGIONS);
455+
Map<Integer, Pair<Integer, Set<Integer>>> regionMap = new HashMap<>();
456+
while (showRegionsResult.next()) {
457+
if (String.valueOf(TConsensusGroupType.DataRegion)
458+
.equals(showRegionsResult.getString(ColumnHeaderConstant.TYPE))
459+
&& !showRegionsResult
460+
.getString(ColumnHeaderConstant.DATABASE)
461+
.equals(SystemConstant.SYSTEM_DATABASE)) {
462+
int regionId = showRegionsResult.getInt(ColumnHeaderConstant.REGION_ID);
463+
int dataNodeId = showRegionsResult.getInt(ColumnHeaderConstant.DATA_NODE_ID);
464+
Pair<Integer, Set<Integer>> leaderNodesPair =
465+
regionMap.computeIfAbsent(regionId, id -> new Pair<>(-1, new HashSet<>()));
466+
leaderNodesPair.getRight().add(dataNodeId);
467+
if (showRegionsResult.getString(ColumnHeaderConstant.ROLE).equals("Leader")) {
468+
leaderNodesPair.setLeft(dataNodeId);
469+
}
470+
}
471+
}
472+
return regionMap;
473+
}
474+
451475
public static Map<Integer, Set<Integer>> getAllRegionMap(Statement statement) throws Exception {
452476
ResultSet showRegionsResult = statement.executeQuery(SHOW_REGIONS);
453477
Map<Integer, Set<Integer>> regionMap = new HashMap<>();

‎integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionReconstructForIoTV1IT.java‎

Lines changed: 56 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -21,17 +21,20 @@
2121

2222
import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
2323
import org.apache.iotdb.commons.cluster.NodeStatus;
24+
import org.apache.iotdb.commons.conf.CommonDescriptor;
2425
import org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework;
2526
import org.apache.iotdb.consensus.ConsensusFactory;
2627
import org.apache.iotdb.isession.SessionDataSet;
2728
import org.apache.iotdb.it.env.EnvFactory;
2829
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
2930
import org.apache.iotdb.it.framework.IoTDBTestRunner;
3031
import org.apache.iotdb.itbase.category.ClusterIT;
32+
import org.apache.iotdb.itbase.env.BaseEnv;
3133
import org.apache.iotdb.rpc.StatementExecutionException;
3234
import org.apache.iotdb.session.Session;
3335

3436
import org.apache.tsfile.read.common.RowRecord;
37+
import org.apache.tsfile.utils.Pair;
3538
import org.awaitility.Awaitility;
3639
import org.junit.Assert;
3740
import org.junit.Test;
@@ -43,12 +46,12 @@
4346
import java.io.File;
4447
import java.sql.Connection;
4548
import java.sql.Statement;
46-
import java.util.Iterator;
4749
import java.util.Map;
4850
import java.util.Set;
4951
import java.util.concurrent.TimeUnit;
5052

5153
import static org.apache.iotdb.util.MagicUtils.makeItCloseQuietly;
54+
import static org.junit.Assert.fail;
5255

5356
@Category({ClusterIT.class})
5457
@RunWith(IoTDBTestRunner.class)
@@ -62,13 +65,17 @@ private boolean deleteTsFiles(File file) {
6265
if (files != null) {
6366
for (File f : files) {
6467
if (!deleteTsFiles(f)) {
68+
LOGGER.info("{} cannot be removed", f);
6569
return false;
70+
} else {
71+
LOGGER.info("{} is removed or ignored", f);
6672
}
6773
}
6874
}
6975
} else if (file.getName().endsWith(".tsfile")) {
7076
return file.delete();
7177
}
78+
file.delete();
7279
return true;
7380
}
7481

@@ -94,16 +101,19 @@ public void normal1C3DTest() throws Exception {
94101
statement.execute(FLUSH_COMMAND);
95102

96103
// collect necessary information
97-
Map<Integer, Set<Integer>> dataRegionMap = getDataRegionMap(statement);
104+
Map<Integer, Pair<Integer, Set<Integer>>> dataRegionMap =
105+
getDataRegionMapWithLeader(statement);
98106
Set<Integer> allDataNodeId = getAllDataNodes(statement);
99107

100108
// select datanode
101109
final int selectedRegion = 3;
102110
Assert.assertTrue(dataRegionMap.containsKey(selectedRegion));
103-
Assert.assertEquals(2, dataRegionMap.get(selectedRegion).size());
104-
Iterator<Integer> iterator = dataRegionMap.get(selectedRegion).iterator();
105-
final int dataNodeToBeClosed = iterator.next();
106-
final int dataNodeToBeReconstructed = iterator.next();
111+
Pair<Integer, Set<Integer>> leaderAndNodeIds = dataRegionMap.get(selectedRegion);
112+
Assert.assertEquals(2, leaderAndNodeIds.right.size());
113+
// reconstruct from the leader to ensure no data is lost
114+
final int dataNodeToBeClosed = leaderAndNodeIds.left;
115+
final int dataNodeToBeReconstructed =
116+
leaderAndNodeIds.right.stream().filter(x -> x != dataNodeToBeClosed).findAny().get();
107117
final int dataNodeAlwaysGood =
108118
allDataNodeId.stream()
109119
.filter(x -> x != dataNodeToBeReconstructed && x != dataNodeToBeClosed)
@@ -126,17 +136,30 @@ public void normal1C3DTest() throws Exception {
126136
.get()
127137
.getDataPath());
128138

129-
if (!deleteTsFiles(dataDirToBeReconstructed)) {
130-
statement.execute(FLUSH_COMMAND);
131-
deleteTsFiles(dataDirToBeReconstructed);
132-
}
133-
134139
EnvFactory.getEnv().dataNodeIdToWrapper(dataNodeToBeClosed).get().stopForcibly();
135140

136-
// now, the query should throw exception
137-
Assert.assertThrows(
138-
StatementExecutionException.class,
139-
() -> session.executeQueryStatement("select * from root.sg.**"));
141+
while (true) {
142+
if (!deleteTsFiles(dataDirToBeReconstructed)) {
143+
try (Connection flushConn =
144+
EnvFactory.getEnv()
145+
.getConnection(
146+
EnvFactory.getEnv().dataNodeIdToWrapper(dataNodeToBeReconstructed).get(),
147+
CommonDescriptor.getInstance().getConfig().getAdminName(),
148+
CommonDescriptor.getInstance().getConfig().getAdminPassword(),
149+
BaseEnv.TREE_SQL_DIALECT);
150+
Statement flushStatement = flushConn.createStatement()) {
151+
flushStatement.execute("FLUSH ON LOCAL");
152+
}
153+
deleteTsFiles(dataDirToBeReconstructed);
154+
}
155+
156+
// now, the query should throw exception
157+
try {
158+
session.executeQueryStatement("select * from root.sg.**");
159+
} catch (StatementExecutionException e) {
160+
break;
161+
}
162+
}
140163

141164
// start DataNode, reconstruct the delete one
142165
EnvFactory.getEnv().dataNodeIdToWrapper(dataNodeToBeClosed).get().start();
@@ -156,15 +179,27 @@ public void normal1C3DTest() throws Exception {
156179
"Two factor: {} && {}",
157180
getRegionStatusWithoutRunning(session).isEmpty(),
158181
dataDirToBeReconstructed.getAbsoluteFile().exists());
159-
Assert.fail();
182+
fail();
160183
}
161184
EnvFactory.getEnv().dataNodeIdToWrapper(dataNodeToBeClosed).get().stopForcibly();
162185

163-
// now, the query should work fine
164-
SessionDataSet resultSet = session.executeQueryStatement("select * from root.sg.**");
165-
RowRecord rowRecord = resultSet.next();
166-
Assert.assertEquals("2.0", rowRecord.getField(0).getStringValue());
167-
Assert.assertEquals("1.0", rowRecord.getField(1).getStringValue());
186+
// now, the query should work fine, but the update of region status may have some delay
187+
long start = System.currentTimeMillis();
188+
while (true) {
189+
SessionDataSet resultSet = null;
190+
try {
191+
resultSet = session.executeQueryStatement("select * from root.sg.**");
192+
} catch (StatementExecutionException e) {
193+
if (System.currentTimeMillis() - start > 60_000L) {
194+
fail("Cannot execute query within 60s");
195+
}
196+
continue;
197+
}
198+
RowRecord rowRecord = resultSet.next();
199+
Assert.assertEquals("2.0", rowRecord.getField(0).getStringValue());
200+
Assert.assertEquals("1.0", rowRecord.getField(1).getStringValue());
201+
break;
202+
}
168203
}
169204
}
170205
}

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipePermissionIT.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,8 @@ public void setUp() {
6868
.setDefaultSchemaRegionGroupNumPerDatabase(1)
6969
.setTimestampPrecision("ms")
7070
.setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
71-
.setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS);
71+
.setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
72+
.setIsPipeEnableMemoryCheck(false);
7273
receiverEnv
7374
.getConfig()
7475
.getCommonConfig()
@@ -77,6 +78,7 @@ public void setUp() {
7778
.setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
7879
.setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
7980
.setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
81+
.setIsPipeEnableMemoryCheck(false)
8082
.setSchemaReplicationFactor(3)
8183
.setDataReplicationFactor(2);
8284

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@ public void setUp() {
8080
.setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
8181
.setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
8282
.setIsPipeEnableMemoryCheck(false);
83+
senderEnv.getConfig().getConfigNodeConfig().setLeaderDistributionPolicy("HASH");
8384

8485
receiverEnv
8586
.getConfig()

0 commit comments

Comments
 (0)