This is an automated email from the ASF dual-hosted git repository.
tanxinyu 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 bd588e27682 [RatisConsensus] Fix flaky unit tests after linearizable
read (#10942)
bd588e27682 is described below
commit bd588e2768235036864d8937378bb7118217a7ad
Author: William Song <[email protected]>
AuthorDate: Fri Aug 25 22:59:17 2023 +0800
[RatisConsensus] Fix flaky unit tests after linearizable read (#10942)
---
.../iotdb/consensus/ratis/RatisConsensusTest.java | 151 +++++++--------------
.../iotdb/consensus/ratis/RecoverReadTest.java | 51 ++-----
.../apache/iotdb/consensus/ratis/TestUtils.java | 98 ++++++++++---
3 files changed, 138 insertions(+), 162 deletions(-)
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
index 18934e15b28..5679890a654 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
@@ -19,20 +19,15 @@
package org.apache.iotdb.consensus.ratis;
-import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
-import org.apache.iotdb.consensus.IConsensus;
import org.apache.iotdb.consensus.IStateMachine;
import org.apache.iotdb.consensus.common.ConsensusGroup;
import org.apache.iotdb.consensus.common.Peer;
-import org.apache.iotdb.consensus.common.request.ByteBufferConsensusRequest;
import org.apache.iotdb.consensus.config.RatisConfig;
-import org.apache.iotdb.consensus.exception.ConsensusException;
import
org.apache.iotdb.consensus.exception.ConsensusGroupAlreadyExistException;
import org.apache.iotdb.consensus.exception.ConsensusGroupNotExistException;
import
org.apache.iotdb.consensus.exception.PeerAlreadyInConsensusGroupException;
import org.apache.iotdb.consensus.exception.PeerNotInConsensusGroupException;
-import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.ratis.util.TimeDuration;
import org.junit.After;
@@ -57,6 +52,7 @@ public class RatisConsensusTest {
CountDownLatch latch;
private TestUtils.MiniCluster miniCluster;
+ private final ExecutorService writeExecutor =
Executors.newFixedThreadPool(2);
private final RatisConfig config =
RatisConfig.newBuilder()
@@ -98,6 +94,7 @@ public class RatisConsensusTest {
@After
public void tearDown() throws IOException {
+ writeExecutor.shutdown();
miniCluster.cleanUp();
}
@@ -107,7 +104,9 @@ public class RatisConsensusTest {
servers.get(1).createLocalPeer(group.getGroupId(), group.getPeers());
servers.get(2).createLocalPeer(group.getGroupId(), group.getPeers());
- doConsensus(servers.get(0), group.getGroupId(), 10, 10);
+ miniCluster.waitUntilActiveLeader();
+
+ doConsensus(0, 10, 10);
}
@Test
@@ -115,14 +114,11 @@ public class RatisConsensusTest {
List<Peer> original = peers.subList(0, 1);
servers.get(0).createLocalPeer(group.getGroupId(), original);
- doConsensus(servers.get(0), group.getGroupId(), 10, 10);
+ doConsensus(0, 10, 10);
- try {
- servers.get(0).createLocalPeer(group.getGroupId(), original);
- Assert.fail();
- } catch (ConsensusException e) {
- Assert.assertTrue(e instanceof ConsensusGroupAlreadyExistException);
- }
+ Assert.assertThrows(
+ ConsensusGroupAlreadyExistException.class,
+ () -> servers.get(0).createLocalPeer(group.getGroupId(), original));
// add 2 members
servers.get(1).createLocalPeer(group.getGroupId(), peers.subList(1, 2));
@@ -131,24 +127,25 @@ public class RatisConsensusTest {
servers.get(2).createLocalPeer(group.getGroupId(), peers.subList(2, 3));
servers.get(0).addRemotePeer(group.getGroupId(), peers.get(2));
+ miniCluster.waitUntilActiveLeader();
+
Assert.assertEquals(
3, ((TestUtils.IntegerCounter)
stateMachines.get(0)).getConfiguration().size());
- doConsensus(servers.get(0), group.getGroupId(), 10, 20);
+ doConsensus(0, 10, 20);
}
@Test
public void removeMemberFromGroup() throws Exception {
- try {
- servers.get(0).deleteLocalPeer(group.getGroupId());
- Assert.fail();
- } catch (ConsensusException e) {
- Assert.assertTrue(e instanceof ConsensusGroupNotExistException);
- }
+ Assert.assertThrows(
+ ConsensusGroupNotExistException.class,
+ () -> servers.get(0).deleteLocalPeer(group.getGroupId()));
+
servers.get(0).createLocalPeer(group.getGroupId(), group.getPeers());
servers.get(1).createLocalPeer(group.getGroupId(), group.getPeers());
servers.get(2).createLocalPeer(group.getGroupId(), group.getPeers());
- doConsensus(servers.get(0), group.getGroupId(), 10, 10);
+ miniCluster.waitUntilActiveLeader();
+ doConsensus(0, 10, 10);
servers.get(0).transferLeader(gid, peers.get(0));
servers.get(0).removeRemotePeer(gid, peers.get(1));
@@ -156,44 +153,38 @@ public class RatisConsensusTest {
servers.get(0).removeRemotePeer(gid, peers.get(2));
servers.get(2).deleteLocalPeer(gid);
- doConsensus(servers.get(0), group.getGroupId(), 10, 20);
+ miniCluster.waitUntilActiveLeader();
+ doConsensus(0, 10, 20);
}
@Test
public void oneMemberGroupChange() throws Exception {
- try {
- servers.get(0).addRemotePeer(group.getGroupId(), peers.get(0));
- Assert.fail();
- } catch (ConsensusException e) {
- Assert.assertTrue(e instanceof ConsensusGroupNotExistException);
- }
+ Assert.assertThrows(
+ ConsensusGroupNotExistException.class,
+ () -> servers.get(0).addRemotePeer(group.getGroupId(), peers.get(0)));
+
servers.get(0).createLocalPeer(group.getGroupId(), peers.subList(0, 1));
- doConsensus(servers.get(0), group.getGroupId(), 10, 10);
+ doConsensus(0, 10, 10);
servers.get(1).createLocalPeer(group.getGroupId(), peers.subList(1, 2));
servers.get(0).addRemotePeer(group.getGroupId(), peers.get(1));
- try {
- servers.get(0).addRemotePeer(group.getGroupId(), peers.get(1));
- Assert.fail();
- } catch (ConsensusException e) {
- Assert.assertTrue(e instanceof PeerAlreadyInConsensusGroupException);
- }
+ Assert.assertThrows(
+ PeerAlreadyInConsensusGroupException.class,
+ () -> servers.get(0).addRemotePeer(group.getGroupId(), peers.get(1)));
+
servers.get(0).transferLeader(group.getGroupId(), peers.get(1));
+
servers.get(1).removeRemotePeer(group.getGroupId(), peers.get(0));
- try {
- servers.get(1).removeRemotePeer(group.getGroupId(), peers.get(0));
- Assert.fail();
- } catch (ConsensusException e) {
- Assert.assertTrue(e instanceof PeerNotInConsensusGroupException);
- }
+ Assert.assertThrows(
+ PeerNotInConsensusGroupException.class,
+ () -> servers.get(1).removeRemotePeer(group.getGroupId(),
peers.get(0)));
+
Assert.assertEquals(servers.get(1).getLeader(gid).getNodeId(),
peers.get(1).getNodeId());
+
servers.get(0).deleteLocalPeer(group.getGroupId());
- try {
- servers.get(0).deleteLocalPeer(group.getGroupId());
- Assert.fail();
- } catch (ConsensusException e) {
- Assert.assertTrue(e instanceof ConsensusGroupNotExistException);
- }
+ Assert.assertThrows(
+ ConsensusGroupNotExistException.class,
+ () -> servers.get(0).deleteLocalPeer(group.getGroupId()));
}
@Test
@@ -202,13 +193,15 @@ public class RatisConsensusTest {
servers.get(1).createLocalPeer(group.getGroupId(), group.getPeers());
servers.get(2).createLocalPeer(group.getGroupId(), group.getPeers());
+ miniCluster.waitUntilActiveLeader();
// 200 operation will trigger snapshot & purge
- doConsensus(servers.get(0), group.getGroupId(), 200, 200);
+ doConsensus(0, 200, 200);
miniCluster.stop();
miniCluster.restart();
- doConsensus(servers.get(0), gid, 10, 210);
+ miniCluster.waitUntilActiveLeader();
+ doConsensus(0, 10, 210);
}
// FIXME: Turn on the test when it is stable
@@ -217,7 +210,7 @@ public class RatisConsensusTest {
servers.get(1).createLocalPeer(group.getGroupId(), group.getPeers());
servers.get(2).createLocalPeer(group.getGroupId(), group.getPeers());
- doConsensus(servers.get(0), group.getGroupId(), 10, 10);
+ doConsensus(0, 10, 10);
int leaderIndex = servers.get(0).getLeader(group.getGroupId()).getNodeId()
- 1;
@@ -233,64 +226,18 @@ public class RatisConsensusTest {
public void transferSnapshot() throws Exception {
servers.get(0).createLocalPeer(gid, peers.subList(0, 1));
- doConsensus(servers.get(0), gid, 10, 10);
+ doConsensus(0, 10, 10);
servers.get(0).triggerSnapshot(gid);
servers.get(1).createLocalPeer(gid, peers.subList(1, 2));
servers.get(0).addRemotePeer(gid, peers.get(1));
- doConsensus(servers.get(1), gid, 10, 20);
+ miniCluster.waitUntilActiveLeader();
+ doConsensus(1, 10, 20);
}
- private void doConsensus(IConsensus consensus, ConsensusGroupId gid, int
count, int target)
- throws Exception {
-
- latch = new CountDownLatch(count);
- // do write
- ExecutorService executorService = Executors.newFixedThreadPool(2);
- for (int i = 0; i < count; i++) {
- executorService.submit(
- () -> {
- ByteBufferConsensusRequest incrReq =
TestUtils.TestRequest.incrRequest();
-
- TSStatus response;
- try {
- response = consensus.write(gid, incrReq);
- } catch (ConsensusException e) {
- response = new
TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
- response.setMessage(e.getMessage());
- e.printStackTrace(System.out);
- }
- Assert.assertEquals(200, response.getCode());
- latch.countDown();
- });
- }
-
- executorService.shutdown();
-
- // wait at most 60s for write to complete, otherwise fail the test
- Assert.assertTrue(latch.await(60, TimeUnit.SECONDS));
-
- ByteBufferConsensusRequest getReq = TestUtils.TestRequest.getRequest();
-
- // wait at most 60s to discover a valid leader
- long start = System.currentTimeMillis();
- IConsensus leader = null;
- while (leader == null) {
- long current = System.currentTimeMillis();
- if ((current - start) > 60 * 1000) {
- break;
- }
- for (int i = 0; i < 3; i++) {
- if (servers.get(i).isLeader(gid)) {
- leader = servers.get(i);
- }
- }
- }
- Assert.assertNotNull(leader);
-
- // Check we reached a consensus
- TestUtils.TestDataSet result = (TestUtils.TestDataSet) leader.read(gid,
getReq);
- Assert.assertEquals(target, result.getNumber());
+ private void doConsensus(int serverIndex, int count, int target) throws
Exception {
+ miniCluster.writeManyParallel(writeExecutor, serverIndex, count);
+ Assert.assertEquals(target, miniCluster.mustRead(serverIndex));
}
}
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RecoverReadTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RecoverReadTest.java
index b30b6fc5f58..93a972204c2 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RecoverReadTest.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RecoverReadTest.java
@@ -20,11 +20,9 @@ package org.apache.iotdb.consensus.ratis;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
-import org.apache.iotdb.consensus.common.DataSet;
import org.apache.iotdb.consensus.common.Peer;
import org.apache.iotdb.consensus.common.request.IConsensusRequest;
import org.apache.iotdb.consensus.config.RatisConfig;
-import org.apache.iotdb.consensus.exception.ConsensusException;
import org.apache.iotdb.consensus.exception.RatisUnderRecoveryException;
import org.apache.ratis.util.TimeDuration;
@@ -129,8 +127,8 @@ public class RecoverReadTest {
}
// first write 10 ops
- TestUtils.write(miniCluster.getServer(0), gid, 10);
- Assert.assertEquals(10, TestUtils.read(miniCluster.getServer(0), gid));
+ miniCluster.writeManySerial(0, 10);
+ Assert.assertEquals(10, miniCluster.mustRead(0));
// stop the cluster
miniCluster.stop();
@@ -145,7 +143,7 @@ public class RecoverReadTest {
// manually set the canServe flag to true to mimic original implementation
miniCluster.getServer(0).allowStaleRead(gid);
- Assert.assertNotEquals(10, TestUtils.read(miniCluster.getServer(0), gid));
+ Assert.assertNotEquals(10, ((TestUtils.TestDataSet)
miniCluster.readThrough(0)).getNumber());
}
@Test
@@ -158,8 +156,8 @@ public class RecoverReadTest {
}
// first write 10 ops
- TestUtils.write(miniCluster.getServer(0), gid, 10);
- Assert.assertEquals(10, TestUtils.read(miniCluster.getServer(0), gid));
+ miniCluster.writeManySerial(0, 10);
+ Assert.assertEquals(10, miniCluster.mustRead(0));
// stop the cluster
miniCluster.stop();
@@ -174,22 +172,7 @@ public class RecoverReadTest {
// wait an active leader to serve linearizable read requests
miniCluster.waitUntilActiveLeader();
- // try max 3 minutes
- final long startTs = System.currentTimeMillis();
- DataSet resp;
- try {
- resp = TestUtils.doRead(miniCluster.getServer(0), gid);
- } catch (ConsensusException e) {
- final long timeElapsed = System.currentTimeMillis() - startTs;
- if (timeElapsed > 1000 * 60 * 3) { // 3 min
- Assert.fail("Linearizable read failed after 3 minutes, last exception
seen: " + e);
- }
- Thread.sleep(100);
- logger.info("linearizable read failed when restart, retrying: ", e);
- resp = TestUtils.doRead(miniCluster.getServer(0), gid);
- }
-
- Assert.assertEquals(10, ((TestUtils.TestDataSet) resp).getNumber());
+ Assert.assertEquals(10, miniCluster.mustRead(0));
}
@Test
@@ -202,8 +185,8 @@ public class RecoverReadTest {
}
// first write 10 ops
- TestUtils.write(miniCluster.getServer(0), gid, 10);
- Assert.assertEquals(10, TestUtils.read(miniCluster.getServer(0), gid));
+ miniCluster.writeManySerial(0, 10);
+ Assert.assertEquals(10, miniCluster.mustRead(0));
// stop the cluster
miniCluster.stop();
@@ -216,12 +199,7 @@ public class RecoverReadTest {
miniCluster.restart();
// query during redo: get exception that ratis is under recovery
- try {
- TestUtils.doRead(miniCluster.getServer(0), gid);
- Assert.fail();
- } catch (ConsensusException e) {
- Assert.assertTrue(e instanceof RatisUnderRecoveryException);
- }
+ Assert.assertThrows(RatisUnderRecoveryException.class, () ->
miniCluster.readThrough(0));
}
@Test
@@ -234,8 +212,8 @@ public class RecoverReadTest {
}
// first write 30 ops
- TestUtils.write(miniCluster.getServer(0), gid, 50);
- Assert.assertEquals(50, TestUtils.read(miniCluster.getServer(0), gid));
+ miniCluster.writeManySerial(0, 50);
+ Assert.assertEquals(50, miniCluster.mustRead(0));
// stop the cluster
miniCluster.stop();
@@ -251,11 +229,6 @@ public class RecoverReadTest {
miniCluster.waitUntilActiveLeader();
// query during redo: get exception that ratis is under recovery
- try {
- TestUtils.doRead(miniCluster.getServer(0), gid);
- Assert.fail();
- } catch (ConsensusException e) {
- Assert.assertTrue(e instanceof RatisUnderRecoveryException);
- }
+ Assert.assertThrows(RatisUnderRecoveryException.class, () ->
miniCluster.readThrough(0));
}
}
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java
index 75e0426aa5e..dffde2cf994 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java
@@ -26,7 +26,6 @@ import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.consensus.ConsensusFactory;
-import org.apache.iotdb.consensus.IConsensus;
import org.apache.iotdb.consensus.IStateMachine;
import org.apache.iotdb.consensus.common.ConsensusGroup;
import org.apache.iotdb.consensus.common.DataSet;
@@ -36,11 +35,13 @@ import
org.apache.iotdb.consensus.common.request.IConsensusRequest;
import org.apache.iotdb.consensus.config.ConsensusConfig;
import org.apache.iotdb.consensus.config.RatisConfig;
import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.consensus.exception.RatisUnderRecoveryException;
import org.apache.ratis.thirdparty.com.google.common.base.Preconditions;
import org.apache.ratis.util.FileUtils;
import org.apache.ratis.util.JavaUtils;
import org.apache.ratis.util.TimeDuration;
+import org.apache.ratis.util.Timestamp;
import org.junit.Assert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,6 +55,9 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Scanner;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -351,6 +355,78 @@ public class TestUtils {
isStopped.get(), "call resetSMProviderBeforeRestart() before
restart");
this.smProvider = smProvider;
}
+
+ // To success or not to success, that is a question
+ void writeOnce(int serverIndex) {
+ final ByteBufferConsensusRequest increment = TestRequest.incrRequest();
+ final TSStatus response;
+ try {
+ response = servers.get(serverIndex).write(gid, increment);
+ Assert.assertEquals(200, response.getCode());
+ } catch (ConsensusException e) {
+ Assert.fail("Test Env: test write failed due to " + e);
+ }
+ }
+
+ void writeManySerial(int serverIndex, int count) {
+ for (int i = 0; i < count; i++) {
+ writeOnce(serverIndex);
+ }
+ }
+
+ void writeManyParallel(ExecutorService executor, int serverIndex, int
count) {
+ final CountDownLatch waitGroup = new CountDownLatch(count);
+ for (int i = 0; i < count; i++) {
+ CompletableFuture.runAsync(() -> writeOnce(serverIndex), executor)
+ .thenRun(waitGroup::countDown);
+ }
+
+ try {
+ // wait at most 120s for write to complete, otherwise fail the test
+ Assert.assertTrue(waitGroup.await(120, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ logger.warn("test being interrupted: ", e);
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ // Verily, the clash of arms doth ne'er resemble a feast for guests,
+ // and the notion of defeat doth hold no place therein.
+ int mustRead(int serverIndex) throws InterruptedException {
+ final ByteBufferConsensusRequest readRequest =
TestUtils.TestRequest.getRequest();
+
+ waitUntilActiveLeader();
+
+ final TimeDuration maxTryDuration = TimeDuration.valueOf(3,
TimeUnit.MINUTES);
+ final TimeDuration waitDuration = TimeDuration.valueOf(1000,
TimeUnit.MILLISECONDS);
+ final Timestamp start = Timestamp.currentTime();
+
+ DataSet readResp = null;
+ while (true) {
+ try {
+ readResp = readThrough(serverIndex);
+ break;
+ } catch (RatisUnderRecoveryException e) {
+ logger.warn("ratis is redoing raft log, shall wait some time: ", e);
+ waitDuration.sleep();
+ } catch (ConsensusException e) {
+ logger.error("unexpected error occurred, may try again: ", e);
+ waitDuration.sleep();
+ }
+
+ if (start.elapsedTime().compareTo(maxTryDuration) > 0) {
+ Assert.fail("max retry duration passed without successful read");
+ }
+ }
+
+ return ((TestUtils.TestDataSet) readResp).getNumber();
+ }
+
+ // To success or not to success, ratis don't care
+ DataSet readThrough(int serverIndex) throws ConsensusException {
+ final ByteBufferConsensusRequest getReq =
TestUtils.TestRequest.getRequest();
+ return servers.get(serverIndex).read(gid, getReq);
+ }
}
static class MiniClusterFactory {
@@ -376,24 +452,4 @@ public class TestUtils {
return new MiniCluster(gid, replicas, peerStorageProvider, smProvider,
ratisConfig);
}
}
-
- static void write(IConsensus consensus, ConsensusGroupId gid, int count)
- throws ConsensusException {
- for (int i = 0; i < count; i++) {
- final ByteBufferConsensusRequest increment = TestRequest.incrRequest();
- final TSStatus response = consensus.write(gid, increment);
- Assert.assertEquals(200, response.getCode());
- }
- }
-
- static int read(IConsensus consensus, ConsensusGroupId gid) throws
ConsensusException {
- final DataSet response = doRead(consensus, gid);
- final TestUtils.TestDataSet result = (TestUtils.TestDataSet) response;
- return result.getNumber();
- }
-
- static DataSet doRead(IConsensus consensus, ConsensusGroupId gid) throws
ConsensusException {
- final ByteBufferConsensusRequest getReq =
TestUtils.TestRequest.getRequest();
- return consensus.read(gid, getReq);
- }
}
