Google Git
Sign in
apache / ignite-3 / fbf070b41fbd7dd76a614dfa4399425e4b4f3db4 / . / modules / raft / src / integrationTest / java / org / apache / ignite / raft / jraft / core / ItNodeTest.java
blob: 26552ae804e4f010773735da7b8fc9e53af7f323 [file] [log] [blame]
/*
* 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.ignite.raft.jraft.core;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Collections.synchronizedList;
import static java.util.stream.Collectors.toList;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.util.IgniteUtils.byteBufferToByteArray;
import static org.apache.ignite.raft.jraft.core.TestCluster.ELECTION_TIMEOUT_MILLIS;
import static org.apache.ignite.raft.jraft.test.TestUtils.sender;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import com.codahale.metrics.ConsoleReporter;
import java.io.File;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiPredicate;
import java.util.function.BooleanSupplier;
import java.util.stream.IntStream;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.network.ClusterService;
import org.apache.ignite.internal.network.StaticNodeFinder;
import org.apache.ignite.internal.network.utils.ClusterServiceTestUtils;
import org.apache.ignite.internal.raft.JraftGroupEventsListener;
import org.apache.ignite.internal.raft.storage.impl.DefaultLogStorageFactory;
import org.apache.ignite.internal.raft.storage.impl.IgniteJraftServiceFactory;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.apache.ignite.internal.testframework.WorkDirectory;
import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.raft.jraft.Iterator;
import org.apache.ignite.raft.jraft.JRaftUtils;
import org.apache.ignite.raft.jraft.Node;
import org.apache.ignite.raft.jraft.NodeManager;
import org.apache.ignite.raft.jraft.RaftGroupService;
import org.apache.ignite.raft.jraft.StateMachine;
import org.apache.ignite.raft.jraft.Status;
import org.apache.ignite.raft.jraft.closure.JoinableClosure;
import org.apache.ignite.raft.jraft.closure.ReadIndexClosure;
import org.apache.ignite.raft.jraft.closure.SynchronizedClosure;
import org.apache.ignite.raft.jraft.closure.TaskClosure;
import org.apache.ignite.raft.jraft.conf.Configuration;
import org.apache.ignite.raft.jraft.entity.EnumOutter;
import org.apache.ignite.raft.jraft.entity.PeerId;
import org.apache.ignite.raft.jraft.entity.Task;
import org.apache.ignite.raft.jraft.entity.UserLog;
import org.apache.ignite.raft.jraft.error.LogIndexOutOfBoundsException;
import org.apache.ignite.raft.jraft.error.LogNotFoundException;
import org.apache.ignite.raft.jraft.error.RaftError;
import org.apache.ignite.raft.jraft.error.RaftException;
import org.apache.ignite.raft.jraft.option.BootstrapOptions;
import org.apache.ignite.raft.jraft.option.NodeOptions;
import org.apache.ignite.raft.jraft.option.RaftOptions;
import org.apache.ignite.raft.jraft.option.ReadOnlyOption;
import org.apache.ignite.raft.jraft.rpc.AppendEntriesRequestImpl;
import org.apache.ignite.raft.jraft.rpc.AppendEntriesResponseImpl;
import org.apache.ignite.raft.jraft.rpc.RpcClientEx;
import org.apache.ignite.raft.jraft.rpc.RpcRequests;
import org.apache.ignite.raft.jraft.rpc.RpcServer;
import org.apache.ignite.raft.jraft.rpc.TestIgniteRpcServer;
import org.apache.ignite.raft.jraft.rpc.impl.IgniteRpcClient;
import org.apache.ignite.raft.jraft.rpc.impl.IgniteRpcServer;
import org.apache.ignite.raft.jraft.rpc.impl.core.DefaultRaftClientService;
import org.apache.ignite.raft.jraft.storage.SnapshotThrottle;
import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotReader;
import org.apache.ignite.raft.jraft.storage.snapshot.ThroughputSnapshotThrottle;
import org.apache.ignite.raft.jraft.test.TestPeer;
import org.apache.ignite.raft.jraft.test.TestUtils;
import org.apache.ignite.raft.jraft.util.Bits;
import org.apache.ignite.raft.jraft.util.ExecutorServiceHelper;
import org.apache.ignite.raft.jraft.util.ExponentialBackoffTimeoutStrategy;
import org.apache.ignite.raft.jraft.util.Utils;
import org.apache.ignite.raft.jraft.util.concurrent.FixedThreadsExecutorGroup;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.extension.ExtendWith;
/**
* Integration tests for raft cluster. TODO asch get rid of sleeps wherether possible IGNITE-14832
*/
@ExtendWith(WorkDirectoryExtension.class)
public class ItNodeTest extends BaseIgniteAbstractTest {
private static final IgniteLogger log = Loggers.forClass(ItNodeTest.class);
private static DumpThread dumpThread;
private static class DumpThread extends Thread {
private static long DUMP_TIMEOUT_MS = 5 * 60 * 1000;
private volatile boolean stopped = false;
/** {@inheritDoc} */
@SuppressWarnings("BusyWait") @Override
public void run() {
while (!stopped) {
try {
Thread.sleep(DUMP_TIMEOUT_MS);
log.info("Test hang too long, dump threads");
TestUtils.dumpThreads();
}
catch (InterruptedException e) {
// reset request, continue
continue;
}
}
}
}
private String dataPath;
private final AtomicInteger startedCounter = new AtomicInteger(0);
private final AtomicInteger stoppedCounter = new AtomicInteger(0);
private long testStartMs;
private TestCluster cluster;
private final List<RaftGroupService> services = new ArrayList<>();
private final List<ExecutorService> executors = new ArrayList<>();
private final List<FixedThreadsExecutorGroup> appendEntriesExecutors = new ArrayList<>();
/** Test info. */
private TestInfo testInfo;
@BeforeAll
public static void setupNodeTest() {
dumpThread = new DumpThread();
dumpThread.setName("NodeTest-DumpThread");
dumpThread.setDaemon(true);
dumpThread.start();
}
@AfterAll
public static void tearNodeTest() throws Exception {
dumpThread.stopped = true;
dumpThread.interrupt();
dumpThread.join(100);
}
@BeforeEach
public void setup(TestInfo testInfo, @WorkDirectory Path workDir) throws Exception {
log.info(">>>>>>>>>>>>>>> Start test method: " + testInfo.getDisplayName());
this.testInfo = testInfo;
dataPath = workDir.toString();
testStartMs = Utils.monotonicMs();
dumpThread.interrupt(); // reset dump timeout
}
@AfterEach
public void teardown() throws Exception {
services.forEach(service -> {
try {
service.shutdown();
}
catch (Exception e) {
log.error("Error while closing a service", e);
}
});
executors.forEach(ExecutorServiceHelper::shutdownAndAwaitTermination);
appendEntriesExecutors.forEach(FixedThreadsExecutorGroup::shutdownGracefully);
if (cluster != null)
cluster.stopAll();
startedCounter.set(0);
stoppedCounter.set(0);
TestUtils.assertAllJraftThreadsStopped();
log.info(">>>>>>>>>>>>>>> End test method: " + testInfo.getDisplayName() + ", cost:"
+ (Utils.monotonicMs() - testStartMs) + " ms.");
}
@Test
public void testInitShutdown() {
TestPeer peer = new TestPeer(testInfo, TestUtils.INIT_PORT);
NodeOptions nodeOptions = createNodeOptions(0);
nodeOptions.setFsm(new MockStateMachine(peer.getPeerId()));
nodeOptions.setRaftMetaUri(dataPath + File.separator + "meta");
nodeOptions.setSnapshotUri(dataPath + File.separator + "snapshot");
RaftGroupService service = createService("unittest", peer, nodeOptions, List.of());
service.start();
}
@Test
public void testNodeTaskOverload() throws Exception {
TestPeer peer = new TestPeer(testInfo, TestUtils.INIT_PORT);
NodeOptions nodeOptions = createNodeOptions(0);
RaftOptions raftOptions = new RaftOptions();
raftOptions.setDisruptorBufferSize(2);
nodeOptions.setRaftOptions(raftOptions);
MockStateMachine fsm = new MockStateMachine(peer.getPeerId());
nodeOptions.setFsm(fsm);
nodeOptions.setRaftMetaUri(dataPath + File.separator + "meta");
nodeOptions.setSnapshotUri(dataPath + File.separator + "snapshot");
nodeOptions.setInitialConf(new Configuration(Collections.singletonList(peer.getPeerId())));
RaftGroupService service = createService("unittest", peer, nodeOptions, List.of());
Node node = service.start();
assertEquals(1, node.listPeers().size());
assertTrue(node.listPeers().contains(peer.getPeerId()));
while (!node.isLeader())
;
List<Task> tasks = new ArrayList<>();
AtomicInteger c = new AtomicInteger(0);
for (int i = 0; i < 10; i++) {
ByteBuffer data = ByteBuffer.wrap(("hello" + i).getBytes(UTF_8));
int finalI = i;
Task task = new Task(data, new JoinableClosure(status -> {
log.info("{} i={}", status, finalI);
if (!status.isOk()) {
assertTrue(
status.getRaftError() == RaftError.EBUSY || status.getRaftError() == RaftError.EPERM);
}
c.incrementAndGet();
}));
node.apply(task);
tasks.add(task);
}
Task.joinAll(tasks, TimeUnit.SECONDS.toMillis(30));
assertEquals(10, c.get());
}
/**
* Test rollback stateMachine with readIndex for issue 317: https://github.com/sofastack/sofa-jraft/issues/317
*/
@Test
public void testRollbackStateMachineWithReadIndex_Issue317() throws Exception {
TestPeer peer = new TestPeer(testInfo, TestUtils.INIT_PORT);
NodeOptions nodeOptions = createNodeOptions(0);
CountDownLatch applyCompleteLatch = new CountDownLatch(1);
CountDownLatch applyLatch = new CountDownLatch(1);
CountDownLatch readIndexLatch = new CountDownLatch(1);
AtomicInteger currentValue = new AtomicInteger(-1);
String errorMsg = testInfo.getDisplayName();
StateMachine fsm = new StateMachineAdapter() {
@Override
public void onApply(Iterator iter) {
// Notify that the #onApply is preparing to go.
readIndexLatch.countDown();
// Wait for submitting a read-index request
try {
applyLatch.await();
}
catch (InterruptedException e) {
fail();
}
int i = 0;
while (iter.hasNext()) {
byte[] data = iter.next().array();
int v = Bits.getInt(data, 0);
assertEquals(i++, v);
currentValue.set(v);
}
if (i > 0) {
// rollback
currentValue.set(i - 1);
iter.setErrorAndRollback(1, new Status(-1, errorMsg));
applyCompleteLatch.countDown();
}
}
};
nodeOptions.setFsm(fsm);
nodeOptions.setRaftMetaUri(dataPath + File.separator + "meta");
nodeOptions.setSnapshotUri(dataPath + File.separator + "snapshot");
nodeOptions.setInitialConf(new Configuration(Collections.singletonList(peer.getPeerId())));
RaftGroupService service = createService("unittest", peer, nodeOptions, List.of());
Node node = service.start();
assertEquals(1, node.listPeers().size());
assertTrue(node.listPeers().contains(peer.getPeerId()));
while (!node.isLeader())
;
int n = 5;
{
// apply tasks
for (int i = 0; i < n; i++) {
byte[] b = new byte[4];
Bits.putInt(b, 0, i);
node.apply(new Task(ByteBuffer.wrap(b), null));
}
}
AtomicInteger readIndexSuccesses = new AtomicInteger(0);
{
// Submit a read-index, wait for #onApply
readIndexLatch.await();
CountDownLatch latch = new CountDownLatch(1);
node.readIndex(null, new ReadIndexClosure() {
@Override
public void run(Status status, long index, byte[] reqCtx) {
try {
if (status.isOk())
readIndexSuccesses.incrementAndGet();
else {
assertTrue(
status.getErrorMsg().contains(errorMsg) || status.getRaftError() == RaftError.ETIMEDOUT
|| status.getErrorMsg().contains("Invalid state for readIndex: STATE_ERROR"),
"Unexpected status: " + status);
}
}
finally {
latch.countDown();
}
}
});
// We have already submit a read-index request,
// notify #onApply can go right now
applyLatch.countDown();
// The state machine is in error state, the node should step down.
waitForCondition(() -> !node.isLeader(), 5_000);
latch.await();
applyCompleteLatch.await();
}
// No read-index request succeed.
assertEquals(0, readIndexSuccesses.get());
assertTrue(n - 1 >= currentValue.get());
}
@Test
public void testSingleNode() throws Exception {
TestPeer peer = new TestPeer(testInfo, TestUtils.INIT_PORT);
NodeOptions nodeOptions = createNodeOptions(0);
MockStateMachine fsm = new MockStateMachine(peer.getPeerId());
nodeOptions.setFsm(fsm);
nodeOptions.setRaftMetaUri(dataPath + File.separator + "meta");
nodeOptions.setSnapshotUri(dataPath + File.separator + "snapshot");
nodeOptions.setInitialConf(new Configuration(Collections.singletonList(peer.getPeerId())));
RaftGroupService service = createService("unittest", peer, nodeOptions, List.of());
Node node = service.start();
assertEquals(1, node.listPeers().size());
assertTrue(node.listPeers().contains(peer.getPeerId()));
while (!node.isLeader())
;
sendTestTaskAndWait(node);
assertEquals(10, fsm.getLogs().size());
int i = 0;
for (ByteBuffer data : fsm.getLogs()) {
assertEquals("hello" + i++, stringFromBytes(data.array()));
}
}
private String stringFromBytes(byte[] bytes) {
return new String(bytes, UTF_8);
}
@Test
public void testNoLeader() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
assertTrue(cluster.start(peers.get(0)));
List<Node> followers = cluster.getFollowers();
assertEquals(1, followers.size());
Node follower = followers.get(0);
sendTestTaskAndWait(follower, 0, RaftError.EPERM);
// adds a peer3
PeerId peer3 = new PeerId(UUID.randomUUID().toString());
CountDownLatch latch = new CountDownLatch(1);
follower.addPeer(peer3, new ExpectClosure(RaftError.EPERM, latch));
waitLatch(latch);
// remove the peer0
PeerId peer0 = peers.get(0).getPeerId();
latch = new CountDownLatch(1);
follower.removePeer(peer0, new ExpectClosure(RaftError.EPERM, latch));
waitLatch(latch);
}
@Test
public void testTripleNodesWithReplicatorStateListener() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
//final TestCluster cluster = new TestCluster("unittest", this.dataPath, peers);
UserReplicatorStateListener listener1 = new UserReplicatorStateListener();
UserReplicatorStateListener listener2 = new UserReplicatorStateListener();
cluster = new TestCluster("unitest", dataPath, peers, new LinkedHashSet<>(), ELECTION_TIMEOUT_MILLIS,
opts -> opts.setReplicationStateListeners(List.of(listener1, listener2)), testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
// elect leader
cluster.ensureLeader(cluster.waitAndGetLeader());
for (Node follower : cluster.getFollowers())
waitForCondition(() -> follower.getLeaderId() != null, 5_000);
assertEquals(4, startedCounter.get());
assertEquals(2, cluster.getLeader().getReplicatorStateListeners().size());
assertEquals(2, cluster.getFollowers().get(0).getReplicatorStateListeners().size());
assertEquals(2, cluster.getFollowers().get(1).getReplicatorStateListeners().size());
for (Node node : cluster.getNodes())
node.removeReplicatorStateListener(listener1);
assertEquals(1, cluster.getLeader().getReplicatorStateListeners().size());
assertEquals(1, cluster.getFollowers().get(0).getReplicatorStateListeners().size());
assertEquals(1, cluster.getFollowers().get(1).getReplicatorStateListeners().size());
}
// TODO asch Broken then using volatile log. A follower with empty log can become a leader IGNITE-14832.
@Test
@Disabled("https://issues.apache.org/jira/browse/IGNITE-14832")
public void testVoteTimedoutStepDown() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
// elect and get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
assertEquals(3, leader.listPeers().size());
// apply tasks to leader
sendTestTaskAndWait(leader);
// Stop all followers
List<Node> followers = cluster.getFollowers();
assertFalse(followers.isEmpty());
for (Node node : followers)
assertTrue(cluster.stop(node.getNodeId().getPeerId()));
// Wait leader to step down.
while (leader.isLeader())
Thread.sleep(10);
// old leader try to elect self, it should fail.
((NodeImpl) leader).tryElectSelf();
Thread.sleep(1500);
assertNull(cluster.getLeader());
// Start followers
for (Node node : followers) {
assertTrue(cluster.start(findById(peers, node.getNodeId().getPeerId())));
}
cluster.ensureSame();
}
class UserReplicatorStateListener implements Replicator.ReplicatorStateListener {
/** {@inheritDoc} */
@Override
public void onCreated(PeerId peer) {
int val = startedCounter.incrementAndGet();
log.info("Replicator has been created {} {}", peer, val);
}
@Override
public void stateChanged(final PeerId peer, final ReplicatorState newState) {
log.info("Replicator {} state is changed into {}.", peer, newState);
}
/** {@inheritDoc} */
@Override
public void onError(PeerId peer, Status status) {
log.info("Replicator has errors {} {}", peer, status);
}
/** {@inheritDoc} */
@Override
public void onDestroyed(PeerId peer) {
int val = stoppedCounter.incrementAndGet();
log.info("Replicator has been destroyed {} {}", peer, val);
}
}
@Test
public void testLeaderTransferWithReplicatorStateListener() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, new LinkedHashSet<>(), ELECTION_TIMEOUT_MILLIS,
opts -> opts.setReplicationStateListeners(List.of(new UserReplicatorStateListener())), testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
Node leader = cluster.waitAndGetLeader();
cluster.ensureLeader(leader);
sendTestTaskAndWait(leader);
Thread.sleep(100);
List<Node> followers = cluster.getFollowers();
assertTrue(waitForCondition(() -> startedCounter.get() == 2, 5_000), startedCounter.get() + "");
PeerId targetPeer = followers.get(0).getNodeId().getPeerId().copy();
log.info("Transfer leadership from {} to {}", leader, targetPeer);
assertTrue(leader.transferLeadershipTo(targetPeer).isOk());
Thread.sleep(1000);
cluster.waitAndGetLeader();
assertTrue(waitForCondition(() -> startedCounter.get() == 4, 5_000), startedCounter.get() + "");
for (Node node : cluster.getNodes())
node.clearReplicatorStateListeners();
assertEquals(0, cluster.getLeader().getReplicatorStateListeners().size());
assertEquals(0, cluster.getFollowers().get(0).getReplicatorStateListeners().size());
assertEquals(0, cluster.getFollowers().get(1).getReplicatorStateListeners().size());
}
@Test
public void testTripleNodes() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
// elect and get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
assertEquals(3, leader.listPeers().size());
// apply tasks to leader
sendTestTaskAndWait(leader);
{
ByteBuffer data = ByteBuffer.wrap("no closure".getBytes(UTF_8));
Task task = new Task(data, null);
leader.apply(task);
}
{
// task with TaskClosure
ByteBuffer data = ByteBuffer.wrap("task closure".getBytes(UTF_8));
List<String> cbs = synchronizedList(new ArrayList<>());
CountDownLatch latch = new CountDownLatch(1);
Task task = new Task(data, new TaskClosure() {
@Override
public void run(Status status) {
cbs.add("apply");
latch.countDown();
}
@Override
public void onCommitted() {
cbs.add("commit");
}
});
leader.apply(task);
latch.await();
assertEquals(2, cbs.size());
assertEquals("commit", cbs.get(0));
assertEquals("apply", cbs.get(1));
}
cluster.ensureSame();
assertEquals(2, cluster.getFollowers().size());
}
@Test
public void testSingleNodeWithLearner() throws Exception {
TestPeer peer = new TestPeer(testInfo, TestUtils.INIT_PORT);
TestPeer learnerPeer = new TestPeer(testInfo, TestUtils.INIT_PORT + 1);
final int cnt = 10;
MockStateMachine learnerFsm;
RaftGroupService learnerServer;
{
// Start learner
NodeOptions nodeOptions = createNodeOptions(0);
learnerFsm = new MockStateMachine(learnerPeer.getPeerId());
nodeOptions.setFsm(learnerFsm);
nodeOptions.setRaftMetaUri(dataPath + File.separator + "meta1");
nodeOptions.setSnapshotUri(dataPath + File.separator + "snapshot1");
nodeOptions.setInitialConf(new Configuration(Collections.singletonList(peer.getPeerId()), Collections
.singletonList(learnerPeer.getPeerId())));
learnerServer = createService("unittest", learnerPeer, nodeOptions, List.of(peer, learnerPeer));
learnerServer.start();
}
{
// Start leader
NodeOptions nodeOptions = createNodeOptions(1);
MockStateMachine fsm = new MockStateMachine(peer.getPeerId());
nodeOptions.setFsm(fsm);
nodeOptions.setRaftMetaUri(dataPath + File.separator + "meta");
nodeOptions.setSnapshotUri(dataPath + File.separator + "snapshot");
nodeOptions.setInitialConf(new Configuration(Collections.singletonList(peer.getPeerId()), Collections
.singletonList(learnerPeer.getPeerId())));
RaftGroupService server = createService("unittest", peer, nodeOptions, List.of(peer, learnerPeer));
Node node = server.start();
assertEquals(1, node.listPeers().size());
assertTrue(node.listPeers().contains(peer.getPeerId()));
assertTrue(waitForCondition(() -> node.isLeader(), 1_000));
sendTestTaskAndWait(node, cnt);
assertEquals(cnt, fsm.getLogs().size());
int i = 0;
for (ByteBuffer data : fsm.getLogs())
assertEquals("hello" + i++, stringFromBytes(data.array()));
Thread.sleep(1000); //wait for entries to be replicated to learner.
server.shutdown();
}
{
// assert learner fsm
assertEquals(cnt, learnerFsm.getLogs().size());
int i = 0;
for (ByteBuffer data : learnerFsm.getLogs())
assertEquals("hello" + i++, stringFromBytes(data.array()));
learnerServer.shutdown();
}
}
@Test
public void testResetLearners() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
LinkedHashSet<TestPeer> learners = new LinkedHashSet<>();
for (int i = 0; i < 3; i++)
learners.add(new TestPeer(testInfo, TestUtils.INIT_PORT + 3 + i));
cluster = new TestCluster("unittest", dataPath, peers, learners, ELECTION_TIMEOUT_MILLIS, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
int i = 0;
for (TestPeer peer : learners) {
assertTrue(cluster.startLearner(peer));
i++;
}
// elect leader
Node leader = cluster.waitAndGetLeader();
cluster.ensureLeader(leader);
waitForCondition(() -> leader.listAlivePeers().size() == 3, 5_000);
waitForCondition(() -> leader.listAliveLearners().size() == 3, 5_000);
sendTestTaskAndWait(leader);
List<MockStateMachine> fsms = cluster.getFsms();
assertEquals(6, fsms.size());
cluster.ensureSame();
{
// Reset learners to 2 nodes
TestPeer learnerPeer = learners.iterator().next();
learners.remove(learnerPeer);
assertEquals(2, learners.size());
SynchronizedClosure done = new SynchronizedClosure();
leader.resetLearners(learners.stream().map(TestPeer::getPeerId).collect(toList()), done);
assertTrue(done.await().isOk());
assertEquals(2, leader.listAliveLearners().size());
assertEquals(2, leader.listLearners().size());
sendTestTaskAndWait(leader);
Thread.sleep(500);
assertEquals(6, fsms.size());
MockStateMachine fsm = fsms.remove(3); // get the removed learner's fsm
assertEquals(fsm.getPeerId(), learnerPeer.getPeerId());
// Ensure no more logs replicated to the removed learner.
assertTrue(cluster.getLeaderFsm().getLogs().size() > fsm.getLogs().size());
assertEquals(cluster.getLeaderFsm().getLogs().size(), 2 * fsm.getLogs().size());
}
{
// remove another learner
TestPeer learnerPeer = learners.iterator().next();
SynchronizedClosure done = new SynchronizedClosure();
leader.removeLearners(Arrays.asList(learnerPeer.getPeerId()), done);
assertTrue(done.await().isOk());
sendTestTaskAndWait(leader);
Thread.sleep(500);
MockStateMachine fsm = fsms.remove(3); // get the removed learner's fsm
assertEquals(fsm.getPeerId(), learnerPeer.getPeerId());
// Ensure no more logs replicated to the removed learner.
assertTrue(cluster.getLeaderFsm().getLogs().size() > fsm.getLogs().size());
assertEquals(cluster.getLeaderFsm().getLogs().size(), fsm.getLogs().size() / 2 * 3);
}
assertEquals(3, leader.listAlivePeers().size());
assertEquals(1, leader.listAliveLearners().size());
assertEquals(1, leader.listLearners().size());
}
@Test
public void testTripleNodesWithStaticLearners() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
LinkedHashSet<TestPeer> learners = new LinkedHashSet<>();
TestPeer learnerPeer = new TestPeer(testInfo, TestUtils.INIT_PORT + 3);
learners.add(learnerPeer);
cluster.setLearners(learners);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
Node leader = cluster.waitAndGetLeader();
cluster.ensureLeader(leader);
assertEquals(3, leader.listPeers().size());
assertEquals(1, leader.listLearners().size());
assertTrue(leader.listLearners().contains(learnerPeer.getPeerId()));
assertTrue(leader.listAliveLearners().isEmpty());
// start learner after cluster setup.
assertTrue(cluster.start(learnerPeer));
Thread.sleep(1000);
assertEquals(3, leader.listPeers().size());
assertEquals(1, leader.listLearners().size());
assertEquals(1, leader.listAliveLearners().size());
// apply tasks to leader
sendTestTaskAndWait(leader);
cluster.ensureSame();
assertEquals(4, cluster.getFsms().size());
}
@Test
public void testTripleNodesWithLearners() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
//elect and get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
assertEquals(3, leader.listPeers().size());
assertTrue(leader.listLearners().isEmpty());
assertTrue(leader.listAliveLearners().isEmpty());
{
// Adds a learner
SynchronizedClosure done = new SynchronizedClosure();
TestPeer learnerPeer = new TestPeer(testInfo, TestUtils.INIT_PORT + 3);
// Start learner
assertTrue(cluster.startLearner(learnerPeer));
leader.addLearners(Arrays.asList(learnerPeer.getPeerId()), done);
assertTrue(done.await().isOk());
assertEquals(1, leader.listAliveLearners().size());
assertEquals(1, leader.listLearners().size());
}
// apply tasks to leader
sendTestTaskAndWait(leader);
{
ByteBuffer data = ByteBuffer.wrap("no closure".getBytes(UTF_8));
Task task = new Task(data, null);
leader.apply(task);
}
{
// task with TaskClosure
ByteBuffer data = ByteBuffer.wrap("task closure".getBytes(UTF_8));
List<String> cbs = synchronizedList(new ArrayList<>());
CountDownLatch latch = new CountDownLatch(1);
Task task = new Task(data, new TaskClosure() {
@Override
public void run(Status status) {
cbs.add("apply");
latch.countDown();
}
@Override
public void onCommitted() {
cbs.add("commit");
}
});
leader.apply(task);
latch.await();
assertEquals(2, cbs.size());
assertEquals("commit", cbs.get(0));
assertEquals("apply", cbs.get(1));
}
assertEquals(4, cluster.getFsms().size());
assertEquals(2, cluster.getFollowers().size());
assertEquals(1, cluster.getLearners().size());
cluster.ensureSame();
{
// Adds another learner
SynchronizedClosure done = new SynchronizedClosure();
TestPeer learnerPeer = new TestPeer(testInfo, TestUtils.INIT_PORT + 4);
// Start learner
assertTrue(cluster.startLearner(learnerPeer));
leader.addLearners(Arrays.asList(learnerPeer.getPeerId()), done);
assertTrue(done.await().isOk());
assertEquals(2, leader.listAliveLearners().size());
assertEquals(2, leader.listLearners().size());
cluster.ensureSame();
}
{
// stop two followers
for (Node follower : cluster.getFollowers())
assertTrue(cluster.stop(follower.getNodeId().getPeerId()));
// send a new task
ByteBuffer data = ByteBuffer.wrap("task closure".getBytes(UTF_8));
SynchronizedClosure done = new SynchronizedClosure();
leader.apply(new Task(data, done));
// should fail
assertFalse(done.await().isOk());
assertEquals(RaftError.EPERM, done.getStatus().getRaftError());
// One peer with two learners.
assertEquals(3, cluster.getFsms().size());
}
}
@Test
public void testNodesWithPriorityElection() throws Exception {
List<Integer> priorities = new ArrayList<>();
priorities.add(100);
priorities.add(40);
priorities.add(40);
List<TestPeer> peers = TestUtils.generatePriorityPeers(testInfo, 3, priorities);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
//elect get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
assertEquals(3, leader.listPeers().size());
assertEquals(100, leader.getNodeTargetPriority());
assertEquals(100, leader.getLeaderId().getPriority());
assertEquals(2, cluster.getFollowers().size());
}
@Test
public void testNodesWithPartPriorityElection() throws Exception {
List<Integer> priorities = new ArrayList<>();
priorities.add(100);
priorities.add(40);
priorities.add(-1);
List<TestPeer> peers = TestUtils.generatePriorityPeers(testInfo, 3, priorities);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
//elect and get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
assertEquals(3, leader.listPeers().size());
assertEquals(2, cluster.getFollowers().size());
}
@Test
public void testNodesWithSpecialPriorityElection() throws Exception {
List<Integer> priorities = new ArrayList<>();
priorities.add(0);
priorities.add(0);
priorities.add(-1);
List<TestPeer> peers = TestUtils.generatePriorityPeers(testInfo, 3, priorities);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
//elect and get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
assertEquals(3, leader.listPeers().size());
assertEquals(2, cluster.getFollowers().size());
}
@Test
public void testNodesWithZeroValPriorityElection() throws Exception {
List<Integer> priorities = new ArrayList<>();
priorities.add(50);
priorities.add(0);
priorities.add(0);
List<TestPeer> peers = TestUtils.generatePriorityPeers(testInfo, 3, priorities);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
//wait and get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
assertEquals(3, leader.listPeers().size());
assertEquals(2, cluster.getFollowers().size());
assertEquals(50, leader.getNodeTargetPriority());
assertEquals(50, leader.getLeaderId().getPriority());
}
@Test
public void testNoLeaderWithZeroValPriorityElection() throws Exception {
List<Integer> priorities = new ArrayList<>();
priorities.add(0);
priorities.add(0);
priorities.add(0);
List<TestPeer> peers = TestUtils.generatePriorityPeers(testInfo, 3, priorities);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
Thread.sleep(200);
List<Node> followers = cluster.getFollowers();
assertEquals(3, followers.size());
for (Node follower : followers)
assertEquals(0, follower.getNodeId().getPeerId().getPriority());
}
@Test
public void testLeaderStopAndReElectWithPriority() throws Exception {
List<Integer> priorities = new ArrayList<>();
priorities.add(100);
priorities.add(60);
priorities.add(10);
List<TestPeer> peers = TestUtils.generatePriorityPeers(testInfo, 3, priorities);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
Node leader = cluster.waitAndGetLeader();
cluster.ensureLeader(leader);
assertNotNull(leader);
assertEquals(100, leader.getNodeId().getPeerId().getPriority());
assertEquals(100, leader.getNodeTargetPriority());
// apply tasks to leader
sendTestTaskAndWait(leader);
// wait for all update received, before election of new leader
cluster.ensureSame();
// stop leader
assertTrue(cluster.stop(leader.getNodeId().getPeerId()));
// elect new leader
leader = cluster.waitAndGetLeader();
assertNotNull(leader);
// nodes with the same log size will elect leader only by priority
assertEquals(60, leader.getNodeId().getPeerId().getPriority());
assertEquals(100, leader.getNodeTargetPriority());
}
@Test
public void testRemoveLeaderWithPriority() throws Exception {
List<Integer> priorities = new ArrayList<>();
priorities.add(100);
priorities.add(60);
priorities.add(10);
List<TestPeer> peers = TestUtils.generatePriorityPeers(testInfo, 3, priorities);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
// elect and get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
assertEquals(100, leader.getNodeTargetPriority());
assertEquals(100, leader.getNodeId().getPeerId().getPriority());
List<Node> followers = cluster.getFollowers();
assertEquals(2, followers.size());
PeerId oldLeader = leader.getNodeId().getPeerId().copy();
// remove old leader
log.info("Remove old leader {}", oldLeader);
CountDownLatch latch = new CountDownLatch(1);
leader.removePeer(oldLeader, new ExpectClosure(latch));
waitLatch(latch);
assertEquals(60, leader.getNodeTargetPriority());
// stop and clean old leader
log.info("Stop and clean old leader {}", oldLeader);
assertTrue(cluster.stop(oldLeader));
cluster.clean(oldLeader);
// elect new leader
leader = cluster.waitAndGetLeader();
log.info("New leader is {}", leader);
assertNotNull(leader);
assertNotEquals(leader.getNodeId().getPeerId(), oldLeader);
}
@Test
public void testChecksum() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
// start with checksum validation
{
TestCluster cluster = new TestCluster("unittest", dataPath, peers, testInfo);
try {
RaftOptions raftOptions = new RaftOptions();
raftOptions.setEnableLogEntryChecksum(true);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer, false, 300, true, null, raftOptions));
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
assertEquals(3, leader.listPeers().size());
sendTestTaskAndWait(leader);
cluster.ensureSame();
}
finally {
cluster.stopAll();
}
}
// restart with peer3 enable checksum validation
{
TestCluster cluster = new TestCluster("unittest", dataPath, peers, testInfo);
try {
RaftOptions raftOptions = new RaftOptions();
raftOptions.setEnableLogEntryChecksum(false);
for (TestPeer peer : peers) {
if (peer.equals(peers.get(2))) {
raftOptions = new RaftOptions();
raftOptions.setEnableLogEntryChecksum(true);
}
assertTrue(cluster.start(peer, false, 300, true, null, raftOptions));
}
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
assertEquals(3, leader.listPeers().size());
sendTestTaskAndWait(leader);
cluster.ensureSame();
}
finally {
cluster.stopAll();
}
}
// restart with no checksum validation
{
TestCluster cluster = new TestCluster("unittest", dataPath, peers, testInfo);
try {
RaftOptions raftOptions = new RaftOptions();
raftOptions.setEnableLogEntryChecksum(false);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer, false, 300, true, null, raftOptions));
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
assertEquals(3, leader.listPeers().size());
sendTestTaskAndWait(leader);
cluster.ensureSame();
}
finally {
cluster.stopAll();
}
}
// restart with all peers enable checksum validation
{
TestCluster cluster = new TestCluster("unittest", dataPath, peers, testInfo);
try {
RaftOptions raftOptions = new RaftOptions();
raftOptions.setEnableLogEntryChecksum(true);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer, false, 300, true, null, raftOptions));
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
assertEquals(3, leader.listPeers().size());
sendTestTaskAndWait(leader);
cluster.ensureSame();
}
finally {
cluster.stopAll();
}
}
}
@Test
public void testReadIndex() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer, false, 300, true));
//elect and get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
assertEquals(3, leader.listPeers().size());
// apply tasks to leader
sendTestTaskAndWait(leader);
// first call will fail-fast when no connection
if (!assertReadIndex(leader, 11))
assertTrue(assertReadIndex(leader, 11));
// read from follower
for (Node follower : cluster.getFollowers()) {
assertNotNull(follower);
assertTrue(waitForCondition(() -> leader.getNodeId().getPeerId().equals(follower.getLeaderId()), 5_000));
assertReadIndex(follower, 11);
}
// read with null request context
CountDownLatch latch = new CountDownLatch(1);
leader.readIndex(null, new ReadIndexClosure() {
@Override
public void run(Status status, long index, byte[] reqCtx) {
assertNull(reqCtx);
assertTrue(status.isOk());
latch.countDown();
}
});
latch.await();
}
@Test // TODO asch do we need read index timeout ? https://issues.apache.org/jira/browse/IGNITE-14832
public void testReadIndexTimeout() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer, false, 300, true));
//elect and get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
assertEquals(3, leader.listPeers().size());
// apply tasks to leader
sendTestTaskAndWait(leader);
// first call will fail-fast when no connection
if (!assertReadIndex(leader, 11))
assertTrue(assertReadIndex(leader, 11));
// read from follower
for (Node follower : cluster.getFollowers()) {
assertNotNull(follower);
assertTrue(waitForCondition(() -> leader.getNodeId().getPeerId().equals(follower.getLeaderId()), 5_000));
assertReadIndex(follower, 11);
}
// read with null request context
CountDownLatch latch = new CountDownLatch(1);
long start = System.currentTimeMillis();
leader.readIndex(null, new ReadIndexClosure() {
@Override
public void run(Status status, long index, byte[] reqCtx) {
assertNull(reqCtx);
if (status.isOk())
System.err.println("Read-index so fast: " + (System.currentTimeMillis() - start) + "ms");
else {
assertEquals(new Status(RaftError.ETIMEDOUT, "read-index request timeout"), status);
assertEquals(-1, index);
}
latch.countDown();
}
});
latch.await();
}
@Test
public void testReadIndexFromLearner() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer, false, 300, true));
// elect and get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
assertEquals(3, leader.listPeers().size());
// apply tasks to leader
sendTestTaskAndWait(leader);
{
// Adds a learner
SynchronizedClosure done = new SynchronizedClosure();
TestPeer learnerPeer = new TestPeer(testInfo, TestUtils.INIT_PORT + 3);
// Start learner
assertTrue(cluster.startLearner(learnerPeer));
leader.addLearners(Arrays.asList(learnerPeer.getPeerId()), done);
assertTrue(done.await().isOk());
assertEquals(1, leader.listAliveLearners().size());
assertEquals(1, leader.listLearners().size());
}
Thread.sleep(100);
// read from learner
Node learner = cluster.getNodes().get(3);
assertNotNull(leader);
assertReadIndex(learner, 12);
assertReadIndex(learner, 12);
}
@Test
public void testReadIndexChaos() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer, false, 300, true));
//wait and get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
assertEquals(3, leader.listPeers().size());
CountDownLatch latch = new CountDownLatch(10);
ExecutorService executor = Executors.newFixedThreadPool(10);
executors.add(executor);
for (int i = 0; i < 10; i++) {
executor.submit(new Runnable() {
/** {@inheritDoc} */
@Override public void run() {
try {
for (int i = 0; i < 100; i++) {
try {
sendTestTaskAndWait(leader);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
readIndexRandom(cluster);
}
}
finally {
latch.countDown();
}
}
private void readIndexRandom(TestCluster cluster) {
CountDownLatch readLatch = new CountDownLatch(1);
byte[] requestContext = TestUtils.getRandomBytes();
cluster.getNodes().get(ThreadLocalRandom.current().nextInt(3))
.readIndex(requestContext, new ReadIndexClosure() {
@Override
public void run(Status status, long index, byte[] reqCtx) {
if (status.isOk()) {
assertTrue(status.isOk(), status.toString());
assertTrue(index > 0);
assertArrayEquals(requestContext, reqCtx);
}
readLatch.countDown();
}
});
try {
readLatch.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
});
}
latch.await();
cluster.ensureSame();
for (MockStateMachine fsm : cluster.getFsms())
assertEquals(10000, fsm.getLogs().size());
}
@Test
public void testNodeMetrics() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer, false, 300, true));
//elect and get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
assertEquals(3, leader.listPeers().size());
// apply tasks to leader
sendTestTaskAndWait(leader);
{
ByteBuffer data = ByteBuffer.wrap("no closure".getBytes(UTF_8));
Task task = new Task(data, null);
leader.apply(task);
}
cluster.ensureSame();
for (Node node : cluster.getNodes()) {
System.out.println("-------------" + node.getNodeId() + "-------------");
ConsoleReporter reporter = ConsoleReporter.forRegistry(node.getNodeMetrics().getMetricRegistry())
.build();
reporter.report();
reporter.close();
System.out.println();
}
// TODO check http status https://issues.apache.org/jira/browse/IGNITE-14832
assertEquals(2, cluster.getFollowers().size());
}
@Test
public void testLeaderFail() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
//elect get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
log.info("Current leader is {}", leader.getLeaderId());
// apply tasks to leader
sendTestTaskAndWait(leader);
List<Node> followers = cluster.getFollowers();
blockMessagesOnFollowers(followers, (msg, nodeId) -> {
if (msg instanceof RpcRequests.RequestVoteRequest) {
RpcRequests.RequestVoteRequest msg0 = (RpcRequests.RequestVoteRequest) msg;
return !msg0.preVote();
}
return false;
});
// stop leader
log.warn("Stop leader {}", leader.getNodeId().getPeerId());
PeerId oldLeader = leader.getNodeId().getPeerId();
assertTrue(cluster.stop(leader.getNodeId().getPeerId()));
assertFalse(followers.isEmpty());
int success = sendTestTaskAndWait("follower apply ", followers.get(0), 10, -1); // Should fail, because no leader.
stopBlockingMessagesOnFollowers(followers);
// elect new leader
leader = cluster.waitAndGetLeader();
log.info("Elect new leader is {}", leader.getLeaderId());
// apply tasks to new leader
CountDownLatch latch = new CountDownLatch(10);
for (int i = 10; i < 20; i++) {
ByteBuffer data = ByteBuffer.wrap(("hello" + i).getBytes(UTF_8));
Task task = new Task(data, new ExpectClosure(latch));
leader.apply(task);
}
waitLatch(latch);
// restart old leader
log.info("restart old leader {}", oldLeader);
assertTrue(cluster.start(findById(peers, oldLeader)));
// apply something
latch = new CountDownLatch(10);
for (int i = 20; i < 30; i++) {
ByteBuffer data = ByteBuffer.wrap(("hello" + i).getBytes(UTF_8));
Task task = new Task(data, new ExpectClosure(latch));
leader.apply(task);
}
waitLatch(latch);
// stop and clean old leader
cluster.stop(oldLeader);
cluster.clean(oldLeader);
// restart old leader
log.info("Restart old leader with cleanup {}", oldLeader);
assertTrue(cluster.start(findById(peers, oldLeader)));
cluster.ensureSame();
for (MockStateMachine fsm : cluster.getFsms())
assertEquals(30 + success, fsm.getLogs().size());
}
@Test
public void testJoinNodes() throws Exception {
TestPeer peer0 = new TestPeer(testInfo, TestUtils.INIT_PORT);
TestPeer peer1 = new TestPeer(testInfo, TestUtils.INIT_PORT + 1);
TestPeer peer2 = new TestPeer(testInfo, TestUtils.INIT_PORT + 2);
TestPeer peer3 = new TestPeer(testInfo, TestUtils.INIT_PORT + 3);
ArrayList<TestPeer> peers = new ArrayList<>();
peers.add(peer0);
// start single cluster
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
assertTrue(cluster.start(peer0));
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
assertEquals(leader.getNodeId().getPeerId(), peer0.getPeerId());
sendTestTaskAndWait(leader);
// start peer1
assertTrue(cluster.start(peer1, false, 300));
// add peer1
CountDownLatch latch = new CountDownLatch(1);
peers.add(peer1);
leader.addPeer(peer1.getPeerId(), new ExpectClosure(latch));
waitLatch(latch);
cluster.ensureSame();
assertEquals(2, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms())
assertEquals(10, fsm.getLogs().size());
// add peer2 but not start
peers.add(peer2);
latch = new CountDownLatch(1);
leader.addPeer(peer2.getPeerId(), new ExpectClosure(RaftError.ECATCHUP, latch));
waitLatch(latch);
// start peer2 after 2 seconds
Thread.sleep(2000);
assertTrue(cluster.start(peer2, false, 300));
// re-add peer2
latch = new CountDownLatch(2);
leader.addPeer(peer2.getPeerId(), new ExpectClosure(latch));
// concurrent configuration change
leader.addPeer(peer3.getPeerId(), new ExpectClosure(RaftError.EBUSY, latch));
waitLatch(latch);
// re-add peer2 directly
try {
leader.addPeer(peer2.getPeerId(), new ExpectClosure(latch));
fail();
}
catch (IllegalArgumentException e) {
assertEquals("Peer already exists in current configuration", e.getMessage());
}
cluster.ensureSame();
assertEquals(3, cluster.getFsms().size());
assertEquals(2, cluster.getFollowers().size());
for (MockStateMachine fsm : cluster.getFsms())
assertEquals(10, fsm.getLogs().size());
}
@Test
public void testRemoveFollower() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
// wait and get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
// apply tasks to leader
sendTestTaskAndWait(leader);
cluster.ensureSame();
List<Node> followers = cluster.getFollowers();
assertEquals(2, followers.size());
TestPeer followerPeer = findById(peers, followers.get(0).getNodeId().getPeerId());
// stop and clean follower
log.info("Stop and clean follower {}", followerPeer);
assertTrue(cluster.stop(followerPeer.getPeerId()));
cluster.clean(followerPeer.getPeerId());
// remove follower
log.info("Remove follower {}", followerPeer);
CountDownLatch latch = new CountDownLatch(1);
leader.removePeer(followerPeer.getPeerId(), new ExpectClosure(latch));
waitLatch(latch);
sendTestTaskAndWait(leader, 10, RaftError.SUCCESS);
followers = cluster.getFollowers();
assertEquals(1, followers.size());
peers = TestUtils.generatePeers(testInfo, 3);
assertTrue(peers.remove(followerPeer));
// start follower
log.info("Start and add follower {}", followerPeer);
assertTrue(cluster.start(followerPeer));
// re-add follower
latch = new CountDownLatch(1);
leader.addPeer(followerPeer.getPeerId(), new ExpectClosure(latch));
waitLatch(latch);
followers = cluster.getFollowers();
assertEquals(2, followers.size());
cluster.ensureSame();
assertEquals(3, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms())
assertEquals(20, fsm.getLogs().size());
}
@Test
public void testRemoveLeader() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
//elect and get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
// apply tasks to leader
sendTestTaskAndWait(leader);
cluster.ensureSame();
List<Node> followers = cluster.getFollowers();
assertEquals(2, followers.size());
PeerId leaderId = leader.getNodeId().getPeerId();
TestPeer oldLeader = findById(peers, leaderId);
// remove old leader
log.info("Remove old leader {}", oldLeader);
CountDownLatch latch = new CountDownLatch(1);
leader.removePeer(oldLeader.getPeerId(), new ExpectClosure(latch));
waitLatch(latch);
// elect new leader
leader = cluster.waitAndGetLeader();
log.info("New leader is {}", leader);
assertNotNull(leader);
// apply tasks to new leader
sendTestTaskAndWait(leader, 10, RaftError.SUCCESS);
// stop and clean old leader
log.info("Stop and clean old leader {}", oldLeader);
assertTrue(cluster.stop(oldLeader.getPeerId()));
cluster.clean(oldLeader.getPeerId());
// Add and start old leader
log.info("Start and add old leader {}", oldLeader);
assertTrue(cluster.start(oldLeader));
peers = TestUtils.generatePeers(testInfo, 3);
assertTrue(peers.remove(oldLeader));
latch = new CountDownLatch(1);
leader.addPeer(oldLeader.getPeerId(), new ExpectClosure(latch));
waitLatch(latch);
followers = cluster.getFollowers();
assertEquals(2, followers.size());
cluster.ensureSame();
assertEquals(3, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms())
assertEquals(20, fsm.getLogs().size());
}
@Test
public void testPreVote() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
// get leader
Node leader = cluster.waitAndGetLeader();
long savedTerm = ((NodeImpl) leader).getCurrentTerm();
assertNotNull(leader);
// apply tasks to leader
sendTestTaskAndWait(leader);
cluster.ensureSame();
List<Node> followers = cluster.getFollowers();
assertEquals(2, followers.size());
PeerId followerPeerId = followers.get(0).getNodeId().getPeerId();
TestPeer followerPeer = findById(peers, followerPeerId);
// remove follower
log.info("Remove follower {}", followerPeer);
CountDownLatch latch = new CountDownLatch(1);
leader.removePeer(followerPeer.getPeerId(), new ExpectClosure(latch));
waitLatch(latch);
sendTestTaskAndWait(leader, 10, RaftError.SUCCESS);
Thread.sleep(2000);
// add follower
log.info("Add follower {}", followerPeer);
peers = TestUtils.generatePeers(testInfo, 3);
assertTrue(peers.remove(followerPeer));
latch = new CountDownLatch(1);
leader.addPeer(followerPeer.getPeerId(), new ExpectClosure(latch));
waitLatch(latch);
leader = cluster.getLeader();
assertNotNull(leader);
// leader term should not be changed.
assertEquals(savedTerm, ((NodeImpl) leader).getCurrentTerm());
}
@Test
public void testSetPeer1() throws Exception {
cluster = new TestCluster("testSetPeer1", dataPath, new ArrayList<>(), testInfo);
TestPeer bootPeer = new TestPeer(testInfo, TestUtils.INIT_PORT);
assertTrue(cluster.start(bootPeer));
List<Node> nodes = cluster.getFollowers();
assertEquals(1, nodes.size());
List<PeerId> peers = new ArrayList<>();
peers.add(bootPeer.getPeerId());
// reset peers from empty
assertTrue(nodes.get(0).resetPeers(new Configuration(peers)).isOk());
assertNotNull(cluster.waitAndGetLeader());
}
@Test
@Disabled("https://issues.apache.org/jira/browse/IGNITE-21457")
public void testSetPeer2() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
// get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
// apply tasks to leader
sendTestTaskAndWait(leader);
cluster.ensureSame();
List<Node> followers = cluster.getFollowers();
assertEquals(2, followers.size());
PeerId followerPeer1 = followers.get(0).getNodeId().getPeerId();
PeerId followerPeer2 = followers.get(1).getNodeId().getPeerId();
log.info("Stop and clean follower {}", followerPeer1);
assertTrue(cluster.stop(followerPeer1));
cluster.clean(followerPeer1);
// apply tasks to leader again
sendTestTaskAndWait(leader, 10, RaftError.SUCCESS);
// set peer when no quorum die
PeerId leaderId = leader.getLeaderId().copy();
log.info("Set peers to {}", leaderId);
log.info("Stop and clean follower {}", followerPeer2);
assertTrue(cluster.stop(followerPeer2));
cluster.clean(followerPeer2);
assertTrue(waitForTopology(cluster, leaderId, 1, 5_000));
// leader will step-down, become follower
Thread.sleep(2000);
List<PeerId> newPeers = new ArrayList<>();
newPeers.add(leaderId);
// new peers equal to current conf
assertTrue(leader.resetPeers(new Configuration(peers.stream().map(TestPeer::getPeerId).collect(toList()))).isOk());
// set peer when quorum die
log.warn("Set peers to {}", leaderId);
assertTrue(leader.resetPeers(new Configuration(newPeers)).isOk());
leader = cluster.waitAndGetLeader();
assertNotNull(leader);
assertEquals(leaderId, leader.getNodeId().getPeerId());
log.info("start follower {}", followerPeer1);
assertTrue(cluster.start(findById(peers, followerPeer1), true, 300));
log.info("start follower {}", followerPeer2);
assertTrue(cluster.start(findById(peers, followerPeer2), true, 300));
assertTrue(waitForTopology(cluster, followerPeer1, 3, 10_000));
assertTrue(waitForTopology(cluster, followerPeer2, 3, 10_000));
CountDownLatch latch = new CountDownLatch(1);
log.info("Add old follower {}", followerPeer1);
leader.addPeer(followerPeer1, new ExpectClosure(latch));
waitLatch(latch);
latch = new CountDownLatch(1);
log.info("Add old follower {}", followerPeer2);
leader.addPeer(followerPeer2, new ExpectClosure(latch));
waitLatch(latch);
newPeers.add(followerPeer1);
newPeers.add(followerPeer2);
cluster.ensureSame();
assertEquals(3, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms())
assertEquals(20, fsm.getLogs().size());
}
/**
* @throws Exception
*/
@Test
public void testRestoreSnapshot() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
// get leader
Node leader = cluster.waitAndGetLeader();
log.info("Leader: " + leader);
assertNotNull(leader);
// apply tasks to leader
sendTestTaskAndWait(leader);
cluster.ensureSame();
triggerLeaderSnapshot(cluster, leader);
// stop leader
PeerId leaderAddr = leader.getNodeId().getPeerId().copy();
assertTrue(cluster.stop(leaderAddr));
// restart leader
cluster.waitAndGetLeader();
assertEquals(0, cluster.getLeaderFsm().getLoadSnapshotTimes());
assertTrue(cluster.start(findById(peers, leaderAddr)));
cluster.ensureSame();
assertEquals(0, cluster.getLeaderFsm().getLoadSnapshotTimes());
}
/**
* @throws Exception
*/
@Test
public void testRestoreSnapshotWithDelta() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
// get leader
Node leader = cluster.waitAndGetLeader();
log.info("Leader: " + leader);
assertNotNull(leader);
// apply tasks to leader
sendTestTaskAndWait(leader);
cluster.ensureSame();
triggerLeaderSnapshot(cluster, leader);
// stop leader
PeerId leaderAddr = leader.getNodeId().getPeerId().copy();
assertTrue(cluster.stop(leaderAddr));
// restart leader
sendTestTaskAndWait(cluster.waitAndGetLeader(), 10, RaftError.SUCCESS);
assertEquals(0, cluster.getLeaderFsm().getLoadSnapshotTimes());
assertTrue(cluster.start(findById(peers, leaderAddr)));
Node oldLeader = cluster.getNode(leaderAddr);
cluster.ensureSame();
assertEquals(0, cluster.getLeaderFsm().getLoadSnapshotTimes());
MockStateMachine fsm = (MockStateMachine) oldLeader.getOptions().getFsm();
assertEquals(1, fsm.getLoadSnapshotTimes());
}
@Test
public void testInstallSnapshotWithThrottle() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer, false, 200, false, new ThroughputSnapshotThrottle(1024, 1)));
// get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
// apply tasks to leader
sendTestTaskAndWait(leader);
cluster.ensureSame();
// stop follower1
List<Node> followers = cluster.getFollowers();
assertEquals(2, followers.size());
PeerId followerAddr = followers.get(0).getNodeId().getPeerId();
assertTrue(cluster.stop(followerAddr));
cluster.waitAndGetLeader();
// apply something more
sendTestTaskAndWait(leader, 10, RaftError.SUCCESS);
Thread.sleep(1000);
// trigger leader snapshot
triggerLeaderSnapshot(cluster, leader);
// apply something more
sendTestTaskAndWait(leader, 20, RaftError.SUCCESS);
// trigger leader snapshot
triggerLeaderSnapshot(cluster, leader, 2);
// wait leader to compact logs
Thread.sleep(1000);
// restart follower.
cluster.clean(followerAddr);
assertTrue(cluster.start(findById(peers, followerAddr), true, 300, false, new ThroughputSnapshotThrottle(1024, 1)));
Thread.sleep(2000);
cluster.ensureSame();
assertEquals(3, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms())
assertEquals(30, fsm.getLogs().size());
}
@Test // TODO add test for timeout on snapshot install https://issues.apache.org/jira/browse/IGNITE-14832
@Disabled("https://issues.apache.org/jira/browse/IGNITE-16467")
public void testInstallLargeSnapshotWithThrottle() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 4);
cluster = new TestCluster("unitest", dataPath, peers.subList(0, 3), testInfo);
for (int i = 0; i < peers.size() - 1; i++) {
TestPeer peer = peers.get(i);
boolean started = cluster.start(peer, false, 200, false);
assertTrue(started);
}
// get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
// apply tasks to leader
sendTestTaskAndWait(leader, 0, RaftError.SUCCESS);
cluster.ensureSame();
// apply something more
for (int i = 1; i < 100; i++)
sendTestTaskAndWait(leader, i * 10, RaftError.SUCCESS);
Thread.sleep(1000);
// trigger leader snapshot
triggerLeaderSnapshot(cluster, leader);
// apply something more
for (int i = 100; i < 200; i++)
sendTestTaskAndWait(leader, i * 10, RaftError.SUCCESS);
// trigger leader snapshot
triggerLeaderSnapshot(cluster, leader, 2);
// wait leader to compact logs
Thread.sleep(1000);
// add follower
TestPeer newPeer = peers.get(3);
SnapshotThrottle snapshotThrottle = new ThroughputSnapshotThrottle(128, 1);
boolean started = cluster.start(newPeer, false, 300, false, snapshotThrottle);
assertTrue(started);
CountDownLatch latch = new CountDownLatch(1);
leader.addPeer(newPeer.getPeerId(), status -> {
assertTrue(status.isOk(), status.toString());
latch.countDown();
});
waitLatch(latch);
cluster.ensureSame();
assertEquals(4, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms())
assertEquals(2000, fsm.getLogs().size());
}
@Test
public void testInstallLargeSnapshot() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 4);
cluster = new TestCluster("unitest", dataPath, peers.subList(0, 3), testInfo);
for (int i = 0; i < peers.size() - 1; i++) {
TestPeer peer = peers.get(i);
boolean started = cluster.start(peer, false, 200, false);
assertTrue(started);
}
// get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
// apply tasks to leader
sendTestTaskAndWait(leader, 0, RaftError.SUCCESS);
cluster.ensureSame();
// apply something more
for (int i = 1; i < 100; i++)
sendTestTaskAndWait(leader, i * 10, RaftError.SUCCESS);
Thread.sleep(1000);
// trigger leader snapshot
triggerLeaderSnapshot(cluster, leader);
// apply something more
for (int i = 100; i < 200; i++)
sendTestTaskAndWait(leader, i * 10, RaftError.SUCCESS);
// trigger leader snapshot
triggerLeaderSnapshot(cluster, leader, 2);
// wait leader to compact logs
Thread.sleep(1000);
// add follower
TestPeer newPeer = peers.get(3);
RaftOptions raftOptions = new RaftOptions();
raftOptions.setMaxByteCountPerRpc(128);
boolean started = cluster.start(newPeer, false, 300, false, null, raftOptions);
assertTrue(started);
CountDownLatch latch = new CountDownLatch(1);
leader.addPeer(newPeer.getPeerId(), status -> {
assertTrue(status.isOk(), status.toString());
latch.countDown();
});
waitLatch(latch);
cluster.ensureSame();
assertEquals(4, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms())
assertEquals(2000, fsm.getLogs().size());
}
@Test
public void testInstallSnapshot() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
// get leader
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
// apply tasks to leader
sendTestTaskAndWait(leader);
cluster.ensureSame();
// stop follower1
List<Node> followers = cluster.getFollowers();
assertEquals(2, followers.size());
PeerId followerAddr = followers.get(0).getNodeId().getPeerId();
assertTrue(cluster.stop(followerAddr));
// apply something more
sendTestTaskAndWait(leader, 10, RaftError.SUCCESS);
// trigger leader snapshot
triggerLeaderSnapshot(cluster, leader);
// apply something more
sendTestTaskAndWait(leader, 20, RaftError.SUCCESS);
triggerLeaderSnapshot(cluster, leader, 2);
// wait leader to compact logs
Thread.sleep(50);
//restart follower.
cluster.clean(followerAddr);
assertTrue(cluster.start(findById(peers, followerAddr), false, 300));
cluster.ensureSame();
assertEquals(3, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms())
assertEquals(30, fsm.getLogs().size(), fsm.getPeerId().toString());
}
@Test
public void testNoSnapshot() throws Exception {
TestPeer peer = new TestPeer(testInfo, TestUtils.INIT_PORT);
NodeOptions nodeOptions = createNodeOptions(0);
MockStateMachine fsm = new MockStateMachine(peer.getPeerId());
nodeOptions.setFsm(fsm);
nodeOptions.setRaftMetaUri(dataPath + File.separator + "meta");
nodeOptions.setInitialConf(new Configuration(Collections.singletonList(peer.getPeerId())));
RaftGroupService service = createService("unittest", peer, nodeOptions, List.of());
Node node = service.start();
// wait node elect self as leader
Thread.sleep(2000);
sendTestTaskAndWait(node);
assertEquals(0, fsm.getSaveSnapshotTimes());
// do snapshot but returns error
CountDownLatch latch = new CountDownLatch(1);
node.snapshot(new ExpectClosure(RaftError.EINVAL, "Snapshot is not supported", latch));
waitLatch(latch);
assertEquals(0, fsm.getSaveSnapshotTimes());
}
@Test
public void testAutoSnapshot() throws Exception {
TestPeer peer = new TestPeer(testInfo, TestUtils.INIT_PORT);
NodeOptions nodeOptions = createNodeOptions(0);
MockStateMachine fsm = new MockStateMachine(peer.getPeerId());
nodeOptions.setFsm(fsm);
nodeOptions.setSnapshotUri(dataPath + File.separator + "snapshot");
nodeOptions.setRaftMetaUri(dataPath + File.separator + "meta");
nodeOptions.setSnapshotIntervalSecs(10);
nodeOptions.setInitialConf(new Configuration(Collections.singletonList(peer.getPeerId())));
RaftGroupService service = createService("unittest", peer, nodeOptions, List.of());
Node node = service.start();
// wait node elect self as leader
Thread.sleep(2000);
sendTestTaskAndWait(node);
// wait for auto snapshot
Thread.sleep(10000);
// first snapshot will be triggered randomly
int times = fsm.getSaveSnapshotTimes();
assertTrue(times >= 1, "snapshotTimes=" + times);
assertTrue(fsm.getSnapshotIndex() > 0);
}
@Test
public void testLeaderShouldNotChange() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
// get leader
Node leader0 = cluster.waitAndGetLeader();
assertNotNull(leader0);
long savedTerm = ((NodeImpl) leader0).getCurrentTerm();
log.info("Current leader is {}, term is {}", leader0, savedTerm);
Thread.sleep(5000);
Node leader1 = cluster.waitAndGetLeader();
assertNotNull(leader1);
log.info("Current leader is {}", leader1);
assertEquals(savedTerm, ((NodeImpl) leader1).getCurrentTerm());
}
@Test
public void testRecoverFollower() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
List<Node> followers = cluster.getFollowers();
assertEquals(2, followers.size());
PeerId followerAddr = followers.get(0).getNodeId().getPeerId().copy();
assertTrue(cluster.stop(followerAddr));
sendTestTaskAndWait(leader);
for (int i = 10; i < 30; i++) {
ByteBuffer data = ByteBuffer.wrap(("no cluster" + i).getBytes(UTF_8));
Task task = new Task(data, null);
leader.apply(task);
}
// wait leader to compact logs
Thread.sleep(5000);
// restart follower
assertTrue(cluster.start(findById(peers, followerAddr)));
cluster.ensureSame();
assertEquals(3, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms())
assertEquals(30, fsm.getLogs().size());
}
@Test
public void testLeaderTransfer() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, ELECTION_TIMEOUT_MILLIS, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
sendTestTaskAndWait(leader);
Thread.sleep(100);
List<Node> followers = cluster.getFollowers();
assertEquals(2, followers.size());
PeerId targetPeer = followers.get(0).getNodeId().getPeerId().copy();
log.info("Transfer leadership from {} to {}", leader, targetPeer);
assertTrue(leader.transferLeadershipTo(targetPeer).isOk());
leader = cluster.waitAndGetLeader();
assertEquals(leader.getNodeId().getPeerId(), targetPeer);
}
@Test
public void testLeaderTransferBeforeLogIsCompleted() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, ELECTION_TIMEOUT_MILLIS, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer, false, 1));
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
List<Node> followers = cluster.getFollowers();
assertEquals(2, followers.size());
PeerId targetPeer = followers.get(0).getNodeId().getPeerId().copy();
assertTrue(cluster.stop(targetPeer));
sendTestTaskAndWait(leader);
log.info("Transfer leadership from {} to {}", leader, targetPeer);
assertTrue(leader.transferLeadershipTo(targetPeer).isOk());
CountDownLatch latch = new CountDownLatch(1);
Task task = new Task(ByteBuffer.wrap("aaaaa".getBytes(UTF_8)), new ExpectClosure(RaftError.EBUSY, latch));
leader.apply(task);
waitLatch(latch);
cluster.waitAndGetLeader();
assertTrue(cluster.start(findById(peers, targetPeer)));
leader = cluster.getLeader();
assertNotEquals(targetPeer, leader.getNodeId().getPeerId());
cluster.ensureSame();
}
@Test
public void testLeaderTransferResumeOnFailure() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, ELECTION_TIMEOUT_MILLIS, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer, false, 1));
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
List<Node> followers = cluster.getFollowers();
assertEquals(2, followers.size());
PeerId targetPeer = followers.get(0).getNodeId().getPeerId().copy();
assertTrue(cluster.stop(targetPeer));
sendTestTaskAndWait(leader);
assertTrue(leader.transferLeadershipTo(targetPeer).isOk());
Node savedLeader = leader;
//try to apply task when transferring leadership
CountDownLatch latch = new CountDownLatch(1);
Task task = new Task(ByteBuffer.wrap("aaaaa".getBytes(UTF_8)), new ExpectClosure(RaftError.EBUSY, latch));
leader.apply(task);
waitLatch(latch);
Thread.sleep(100);
leader = cluster.waitAndGetLeader();
assertSame(leader, savedLeader);
// restart target peer
assertTrue(cluster.start(findById(peers, targetPeer)));
Thread.sleep(100);
// retry apply task
latch = new CountDownLatch(1);
task = new Task(ByteBuffer.wrap("aaaaa".getBytes(UTF_8)), new ExpectClosure(latch));
leader.apply(task);
waitLatch(latch);
cluster.ensureSame();
}
/**
* mock state machine that fails to load snapshot.
*/
static class MockFSM1 extends MockStateMachine {
MockFSM1(PeerId peerId) {
super(peerId);
}
/** {@inheritDoc} */
@Override
public boolean onSnapshotLoad(SnapshotReader reader) {
return false;
}
}
@Test
public void testShutdownAndJoinWorkAfterInitFails() throws Exception {
TestPeer peer = new TestPeer(testInfo, TestUtils.INIT_PORT);
{
NodeOptions nodeOptions = createNodeOptions(0);
MockStateMachine fsm = new MockStateMachine(peer.getPeerId());
nodeOptions.setFsm(fsm);
nodeOptions.setSnapshotUri(dataPath + File.separator + "snapshot");
nodeOptions.setRaftMetaUri(dataPath + File.separator + "meta");
nodeOptions.setSnapshotIntervalSecs(10);
nodeOptions.setInitialConf(new Configuration(Collections.singletonList(peer.getPeerId())));
RaftGroupService service = createService("unittest", peer, nodeOptions, List.of());
Node node = service.start();
Thread.sleep(1000);
sendTestTaskAndWait(node);
// save snapshot
CountDownLatch latch = new CountDownLatch(1);
node.snapshot(new ExpectClosure(latch));
waitLatch(latch);
service.shutdown();
}
{
NodeOptions nodeOptions = createNodeOptions(1);
MockStateMachine fsm = new MockFSM1(peer.getPeerId());
nodeOptions.setFsm(fsm);
nodeOptions.setSnapshotUri(dataPath + File.separator + "snapshot");
nodeOptions.setRaftMetaUri(dataPath + File.separator + "meta");
nodeOptions.setSnapshotIntervalSecs(10);
nodeOptions.setInitialConf(new Configuration(Collections.singletonList(peer.getPeerId())));
RaftGroupService service = createService("unittest", peer, nodeOptions, List.of());
try {
service.start();
fail();
}
catch (Exception e) {
// Expected.
}
}
}
/**
* 4.2.2 Removing the current leader
*
* @throws Exception If failed.
*/
@Test
public void testShuttingDownLeaderTriggerTimeoutNow() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, ELECTION_TIMEOUT_MILLIS, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
Node oldLeader = leader;
log.info("Shutdown leader {}", leader);
leader.shutdown();
leader.join();
leader = cluster.waitAndGetLeader();
assertNotNull(leader);
assertNotSame(leader, oldLeader);
}
@Test
public void testRemovingLeaderTriggerTimeoutNow() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, ELECTION_TIMEOUT_MILLIS, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
cluster.waitAndGetLeader();
// Ensure the quorum before removing a leader, otherwise removePeer can be rejected.
for (Node follower : cluster.getFollowers())
assertTrue(waitForCondition(() -> follower.getLeaderId() != null, 5_000));
Node leader = cluster.getLeader();
assertNotNull(leader);
Node oldLeader = leader;
CountDownLatch latch = new CountDownLatch(1);
oldLeader.removePeer(oldLeader.getNodeId().getPeerId(), new ExpectClosure(latch));
waitLatch(latch);
leader = cluster.waitAndGetLeader();
assertNotNull(leader);
assertNotSame(leader, oldLeader);
}
@Test
public void testTransferShouldWorkAfterInstallSnapshot() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, ELECTION_TIMEOUT_MILLIS, testInfo);
for (int i = 0; i < peers.size() - 1; i++)
assertTrue(cluster.start(peers.get(i)));
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
sendTestTaskAndWait(leader);
List<Node> followers = cluster.getFollowers();
assertEquals(1, followers.size());
PeerId follower = followers.get(0).getNodeId().getPeerId();
assertTrue(leader.transferLeadershipTo(follower).isOk());
leader = cluster.waitAndGetLeader();
assertEquals(follower, leader.getNodeId().getPeerId());
CountDownLatch latch = new CountDownLatch(1);
leader.snapshot(new ExpectClosure(latch));
waitLatch(latch);
latch = new CountDownLatch(1);
leader.snapshot(new ExpectClosure(latch));
waitLatch(latch);
// start the last peer which should be recover with snapshot.
TestPeer lastPeer = peers.get(2);
assertTrue(cluster.start(lastPeer));
Thread.sleep(5000);
assertTrue(leader.transferLeadershipTo(lastPeer.getPeerId()).isOk());
Thread.sleep(2000);
leader = cluster.getLeader();
assertEquals(lastPeer.getPeerId(), leader.getNodeId().getPeerId());
assertEquals(3, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms())
assertEquals(10, fsm.getLogs().size());
}
@Test
public void testAppendEntriesWhenFollowerIsInErrorState() throws Exception {
// start five nodes
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 5);
cluster = new TestCluster("unitest", dataPath, peers, ELECTION_TIMEOUT_MILLIS, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
Node oldLeader = cluster.waitAndGetLeader();
assertNotNull(oldLeader);
// apply something
sendTestTaskAndWait(oldLeader);
// set one follower into error state
List<Node> followers = cluster.getFollowers();
assertEquals(4, followers.size());
Node errorNode = followers.get(0);
PeerId errorPeer = errorNode.getNodeId().getPeerId().copy();
log.info("Set follower {} into error state", errorNode);
((NodeImpl) errorNode).onError(new RaftException(EnumOutter.ErrorType.ERROR_TYPE_STATE_MACHINE, new Status(-1,
"Follower has something wrong.")));
// increase term by stopping leader and electing a new leader again
PeerId oldLeaderAddr = oldLeader.getNodeId().getPeerId().copy();
assertTrue(cluster.stop(oldLeaderAddr));
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
log.info("Elect a new leader {}", leader);
// apply something again
sendTestTaskAndWait(leader, 10, RaftError.SUCCESS);
// stop error follower
Thread.sleep(20);
log.info("Stop error follower {}", errorNode);
assertTrue(cluster.stop(errorPeer));
// restart error and old leader
log.info("Restart error follower {} and old leader {}", errorPeer, oldLeaderAddr);
assertTrue(cluster.start(findById(peers, errorPeer)));
assertTrue(cluster.start(findById(peers, oldLeaderAddr)));
cluster.ensureSame();
assertEquals(5, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms())
assertEquals(20, fsm.getLogs().size());
}
@Test
@Disabled("https://issues.apache.org/jira/browse/IGNITE-21792")
public void testFollowerStartStopFollowing() throws Exception {
// start five nodes
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 5);
cluster = new TestCluster("unitest", dataPath, peers, ELECTION_TIMEOUT_MILLIS, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
Node firstLeader = cluster.waitAndGetLeader();
assertNotNull(firstLeader);
cluster.ensureLeader(firstLeader);
// apply something
sendTestTaskAndWait(firstLeader);
// assert follow times
List<Node> firstFollowers = cluster.getFollowers();
assertEquals(4, firstFollowers.size());
for (Node node : firstFollowers) {
assertTrue(
waitForCondition(() -> ((MockStateMachine) node.getOptions().getFsm()).getOnStartFollowingTimes() == 1, 5_000));
assertEquals(0, ((MockStateMachine) node.getOptions().getFsm()).getOnStopFollowingTimes());
}
// stop leader and elect new one
PeerId fstLeaderAddr = firstLeader.getNodeId().getPeerId();
assertTrue(cluster.stop(fstLeaderAddr));
Node secondLeader = cluster.waitAndGetLeader();
assertNotNull(secondLeader);
sendTestTaskAndWait(secondLeader, 10, RaftError.SUCCESS);
// ensure start/stop following times
List<Node> secondFollowers = cluster.getFollowers();
assertEquals(3, secondFollowers.size());
for (Node node : secondFollowers) {
assertTrue(
waitForCondition(() -> ((MockStateMachine) node.getOptions().getFsm()).getOnStartFollowingTimes() == 2, 5_000));
assertEquals(1, ((MockStateMachine) node.getOptions().getFsm()).getOnStopFollowingTimes());
}
// transfer leadership to a follower
PeerId targetPeer = secondFollowers.get(0).getNodeId().getPeerId().copy();
assertTrue(secondLeader.transferLeadershipTo(targetPeer).isOk());
Thread.sleep(100);
Node thirdLeader = cluster.waitAndGetLeader();
assertEquals(targetPeer, thirdLeader.getNodeId().getPeerId());
sendTestTaskAndWait(thirdLeader, 20, RaftError.SUCCESS);
List<Node> thirdFollowers = cluster.getFollowers();
assertEquals(3, thirdFollowers.size());
for (int i = 0; i < 3; i++) {
Node follower = thirdFollowers.get(i);
if (follower.getNodeId().getPeerId().equals(secondLeader.getNodeId().getPeerId())) {
assertTrue(
waitForCondition(() -> ((MockStateMachine) follower.getOptions().getFsm()).getOnStartFollowingTimes() == 2, 5_000));
assertEquals(1,
((MockStateMachine) follower.getOptions().getFsm()).getOnStopFollowingTimes());
continue;
}
assertTrue(waitForCondition(() -> ((MockStateMachine) follower.getOptions().getFsm()).getOnStartFollowingTimes() == 3, 5_000));
assertEquals(2, ((MockStateMachine) follower.getOptions().getFsm()).getOnStopFollowingTimes());
}
cluster.ensureSame();
}
@Test
public void testLeaderPropagatedBeforeVote() throws Exception {
// start five nodes
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, 3_000, testInfo);
for (TestPeer peer : peers) {
RaftOptions opts = new RaftOptions();
opts.setElectionHeartbeatFactor(4); // Election timeout divisor.
assertTrue(cluster.start(peer, false, 300, false, null, opts));
}
List<NodeImpl> nodes = cluster.getNodes();
AtomicReference<String> guard = new AtomicReference();
// Block only one vote message.
for (NodeImpl node : nodes) {
RpcClientEx rpcClientEx = sender(node);
rpcClientEx.recordMessages((msg, nodeId) -> true);
rpcClientEx.blockMessages((msg, nodeId) -> {
if (msg instanceof RpcRequests.RequestVoteRequest) {
RpcRequests.RequestVoteRequest msg0 = (RpcRequests.RequestVoteRequest)msg;
if (msg0.preVote())
return false;
if (guard.compareAndSet(null, nodeId))
return true;
}
if (msg instanceof RpcRequests.AppendEntriesRequest && nodeId.equals(guard.get())) {
RpcRequests.AppendEntriesRequest tmp = (RpcRequests.AppendEntriesRequest) msg;
if (tmp.entriesList() != null && !tmp.entriesList().isEmpty()) {
return true;
}
}
return false;
});
}
Node leader = cluster.waitAndGetLeader();
cluster.ensureLeader(leader);
RpcClientEx client = sender(leader);
client.stopBlock(1); // Unblock vote message.
// The follower shouldn't stop following on receiving stale vote request.
Node follower = cluster.getNode(PeerId.parsePeer(guard.get()));
boolean res =
waitForCondition(() -> ((MockStateMachine) follower.getOptions().getFsm()).getOnStopFollowingTimes() != 0, 1_000);
assertFalse(res, "The follower shouldn't stop following");
}
@Test
public void readCommittedUserLog() throws Exception {
// setup cluster
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unitest", dataPath, peers, ELECTION_TIMEOUT_MILLIS, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
int amount = 10;
sendTestTaskAndWait(leader, amount);
assertTrue(waitForCondition(() -> {
try {
// index == 1 is a CONFIGURATION log
UserLog userLog = leader.readCommittedUserLog(1 + amount);
return userLog != null;
} catch (Exception ignore) {
// There is a gap between task is applied to FSM and FSMCallerImpl.lastAppliedIndex
// is updated, so we need to wait.
return false;
}
}, 10_000));
// index == 1 is a CONFIGURATION log, so real_index will be 2 when returned.
UserLog userLog = leader.readCommittedUserLog(1);
assertNotNull(userLog);
assertEquals(2, userLog.getIndex());
assertEquals("hello0", stringFromBytes(byteBufferToByteArray(userLog.getData())));
// index == 5 is a DATA log(a user log)
userLog = leader.readCommittedUserLog(5);
assertNotNull(userLog);
assertEquals(5, userLog.getIndex());
assertEquals("hello3", stringFromBytes(byteBufferToByteArray(userLog.getData())));
// index == 15 is greater than last_committed_index
try {
assertNull(leader.readCommittedUserLog(15));
fail();
}
catch (LogIndexOutOfBoundsException e) {
assertEquals("Request index 15 is greater than lastAppliedIndex: 11", e.getMessage());
}
// index == 0 invalid request
try {
assertNull(leader.readCommittedUserLog(0));
fail();
}
catch (LogIndexOutOfBoundsException e) {
assertEquals("Request index is invalid: 0", e.getMessage());
}
log.info("Trigger leader snapshot");
CountDownLatch latch = new CountDownLatch(1);
leader.snapshot(new ExpectClosure(latch));
waitLatch(latch);
// remove and add a peer to add two CONFIGURATION logs
List<Node> followers = cluster.getFollowers();
assertEquals(2, followers.size());
Node testFollower = followers.get(0);
latch = new CountDownLatch(1);
leader.removePeer(testFollower.getNodeId().getPeerId(), new ExpectClosure(latch));
waitLatch(latch);
latch = new CountDownLatch(1);
leader.addPeer(testFollower.getNodeId().getPeerId(), new ExpectClosure(latch));
waitLatch(latch);
sendTestTaskAndWait(leader, amount, RaftError.SUCCESS);
// trigger leader snapshot for the second time, after this the log of index 1~11 will be deleted.
log.info("Trigger leader snapshot");
latch = new CountDownLatch(1);
leader.snapshot(new ExpectClosure(latch));
waitLatch(latch);
Thread.sleep(100);
// index == 5 log has been deleted in log_storage.
try {
leader.readCommittedUserLog(5);
fail();
}
catch (LogNotFoundException e) {
assertEquals("User log is deleted at index: 5", e.getMessage());
}
// index == 12、index == 13、index=14、index=15 are 4 CONFIGURATION logs(joint consensus), so real_index will be 16 when returned.
userLog = leader.readCommittedUserLog(12);
assertNotNull(userLog);
assertEquals(16, userLog.getIndex());
assertEquals("hello10", stringFromBytes(byteBufferToByteArray(userLog.getData())));
// now index == 17 is a user log
userLog = leader.readCommittedUserLog(17);
assertNotNull(userLog);
assertEquals(17, userLog.getIndex());
assertEquals("hello11", stringFromBytes(byteBufferToByteArray(userLog.getData())));
cluster.ensureSame();
assertEquals(3, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms()) {
assertEquals(20, fsm.getLogs().size());
for (int i = 0; i < 20; i++)
assertEquals("hello" + i, stringFromBytes(byteBufferToByteArray(fsm.getLogs().get(i))));
}
}
@Test
public void testBootStrapWithSnapshot() throws Exception {
TestPeer peer = new TestPeer(testInfo, 5006);
MockStateMachine fsm = new MockStateMachine(peer.getPeerId());
Path path = Path.of(dataPath, "node0", "log");
Files.createDirectories(path);
for (char ch = 'a'; ch <= 'z'; ch++)
fsm.getLogs().add(ByteBuffer.wrap(new byte[] {(byte) ch}));
BootstrapOptions opts = new BootstrapOptions();
DefaultLogStorageFactory logStorageProvider = new DefaultLogStorageFactory(path);
logStorageProvider.start();
opts.setServiceFactory(new IgniteJraftServiceFactory(logStorageProvider));
opts.setLastLogIndex(fsm.getLogs().size());
opts.setRaftMetaUri(dataPath + File.separator + "meta");
opts.setSnapshotUri(dataPath + File.separator + "snapshot");
opts.setLogUri("test");
opts.setGroupConf(JRaftUtils.getConfiguration(peer.getPeerId().toString()));
opts.setFsm(fsm);
assertTrue(JRaftUtils.bootstrap(opts));
logStorageProvider.close();
NodeOptions nodeOpts = new NodeOptions();
nodeOpts.setRaftMetaUri(dataPath + File.separator + "meta");
nodeOpts.setSnapshotUri(dataPath + File.separator + "snapshot");
nodeOpts.setLogUri("test");
DefaultLogStorageFactory log2 = new DefaultLogStorageFactory(path);
log2.start();
nodeOpts.setServiceFactory(new IgniteJraftServiceFactory(log2));
nodeOpts.setFsm(fsm);
RaftGroupService service = createService("test", peer, nodeOpts, List.of());
Node node = service.start();
assertEquals(26, fsm.getLogs().size());
for (int i = 0; i < 26; i++)
assertEquals('a' + i, fsm.getLogs().get(i).get());
// Group configuration will be restored from snapshot meta.
while (!node.isLeader())
Thread.sleep(20);
sendTestTaskAndWait(node);
assertEquals(36, fsm.getLogs().size());
}
@Test
public void testBootStrapWithoutSnapshot() throws Exception {
TestPeer peer = new TestPeer(testInfo, 5006);
MockStateMachine fsm = new MockStateMachine(peer.getPeerId());
Path path = Path.of(dataPath, "node0", "log");
Files.createDirectories(path);
BootstrapOptions opts = new BootstrapOptions();
DefaultLogStorageFactory logStorageProvider = new DefaultLogStorageFactory(path);
logStorageProvider.start();
opts.setServiceFactory(new IgniteJraftServiceFactory(logStorageProvider));
opts.setLastLogIndex(0);
opts.setRaftMetaUri(dataPath + File.separator + "meta");
opts.setSnapshotUri(dataPath + File.separator + "snapshot");
opts.setLogUri("test");
opts.setGroupConf(JRaftUtils.getConfiguration(peer.getPeerId().toString()));
opts.setFsm(fsm);
assertTrue(JRaftUtils.bootstrap(opts));
logStorageProvider.close();
NodeOptions nodeOpts = new NodeOptions();
nodeOpts.setRaftMetaUri(dataPath + File.separator + "meta");
nodeOpts.setSnapshotUri(dataPath + File.separator + "snapshot");
nodeOpts.setLogUri("test");
nodeOpts.setFsm(fsm);
DefaultLogStorageFactory log2 = new DefaultLogStorageFactory(path);
log2.start();
nodeOpts.setServiceFactory(new IgniteJraftServiceFactory(log2));
RaftGroupService service = createService("test", peer, nodeOpts, List.of());
Node node = service.start();
while (!node.isLeader())
Thread.sleep(20);
sendTestTaskAndWait(node);
assertEquals(10, fsm.getLogs().size());
}
@Test
public void testChangePeers() throws Exception {
changePeers(false);
}
@Test
public void testChangeAsyncPeers() throws Exception {
changePeers(true);
}
private void changePeers(boolean async) throws Exception {
TestPeer peer0 = new TestPeer(testInfo, TestUtils.INIT_PORT);
cluster = new TestCluster("testChangePeers", dataPath, Collections.singletonList(peer0), testInfo);
assertTrue(cluster.start(peer0));
Node leader = cluster.waitAndGetLeader();
sendTestTaskAndWait(leader);
List<TestPeer> peers = new ArrayList<>();
peers.add(peer0);
int numPeers = 10;
for (int i = 1; i < numPeers; i++) {
TestPeer peer = new TestPeer(testInfo, TestUtils.INIT_PORT + i);
peers.add(peer);
assertTrue(cluster.start(peer, false, 300));
}
waitForTopologyOnEveryNode(numPeers, cluster);
for (int i = 0; i < 9; i++) {
leader = cluster.waitAndGetLeader();
assertNotNull(leader);
PeerId leaderPeer = peers.get(i).getPeerId();
assertEquals(leaderPeer, leader.getNodeId().getPeerId());
PeerId newLeaderPeer = peers.get(i + 1).getPeerId();
if (async) {
SynchronizedClosure done = new SynchronizedClosure();
leader.changePeersAsync(new Configuration(Collections.singletonList(newLeaderPeer)),
leader.getCurrentTerm(), done);
Status status = done.await();
assertTrue(status.isOk(), status.getRaftError().toString());
assertTrue(waitForCondition(() -> {
if (cluster.getLeader() != null) {
return newLeaderPeer.equals(cluster.getLeader().getLeaderId());
}
return false;
}, 10_000));
} else {
SynchronizedClosure done = new SynchronizedClosure();
leader.changePeers(new Configuration(Collections.singletonList(newLeaderPeer)), done);
Status status = done.await();
assertTrue(status.isOk(), status.getRaftError().toString());
}
}
cluster.waitAndGetLeader();
for (MockStateMachine fsm : cluster.getFsms()) {
assertEquals(10, fsm.getLogs().size());
}
}
@Test
public void testOnReconfigurationErrorListener() throws Exception {
TestPeer peer0 = new TestPeer(testInfo, TestUtils.INIT_PORT);
cluster = new TestCluster("testChangePeers", dataPath, Collections.singletonList(peer0), testInfo);
var raftGrpEvtsLsnr = mock(JraftGroupEventsListener.class);
cluster.setRaftGrpEvtsLsnr(raftGrpEvtsLsnr);
assertTrue(cluster.start(peer0));
Node leader = cluster.waitAndGetLeader();
sendTestTaskAndWait(leader);
verify(raftGrpEvtsLsnr, never()).onNewPeersConfigurationApplied(any(), any());
PeerId newPeer = new TestPeer(testInfo, TestUtils.INIT_PORT + 1).getPeerId();
SynchronizedClosure done = new SynchronizedClosure();
leader.changePeersAsync(new Configuration(Collections.singletonList(newPeer)),
leader.getCurrentTerm(), done);
assertEquals(done.await(), Status.OK());
verify(raftGrpEvtsLsnr, timeout(10_000))
.onReconfigurationError(argThat(st -> st.getRaftError() == RaftError.ECATCHUP), any(), any(), anyLong());
}
@Test
public void testNewPeersConfigurationAppliedListener() throws Exception {
TestPeer peer0 = new TestPeer(testInfo, TestUtils.INIT_PORT);
cluster = new TestCluster("testChangePeers", dataPath, Collections.singletonList(peer0), testInfo);
var raftGrpEvtsLsnr = mock(JraftGroupEventsListener.class);
cluster.setRaftGrpEvtsLsnr(raftGrpEvtsLsnr);
assertTrue(cluster.start(peer0));
Node leader = cluster.waitAndGetLeader();
sendTestTaskAndWait(leader);
List<TestPeer> peers = new ArrayList<>();
peers.add(peer0);
List<TestPeer> learners = new ArrayList<>();
int numPeers = 5;
for (int i = 1; i < numPeers; i++) {
TestPeer peer = new TestPeer(testInfo, TestUtils.INIT_PORT + i);
peers.add(peer);
assertTrue(cluster.start(peer, false, 300));
TestPeer learner = new TestPeer(testInfo, TestUtils.INIT_PORT + i + numPeers);
learners.add(learner);
}
verify(raftGrpEvtsLsnr, never()).onNewPeersConfigurationApplied(any(), any());
// Wait until every node sees every other node, otherwise
// changePeersAsync can fail.
waitForTopologyOnEveryNode(numPeers, cluster);
for (int i = 0; i < 4; i++) {
leader = cluster.getLeader();
assertNotNull(leader);
PeerId peer = peers.get(i).getPeerId();
assertEquals(peer, leader.getNodeId().getPeerId());
PeerId newPeer = peers.get(i + 1).getPeerId();
PeerId newLearner = learners.get(i).getPeerId();
SynchronizedClosure done = new SynchronizedClosure();
leader.changePeersAsync(new Configuration(List.of(newPeer), List.of(newLearner)), leader.getCurrentTerm(), done);
assertEquals(done.await(), Status.OK());
assertTrue(waitForCondition(() -> {
if (cluster.getLeader() != null) {
return newPeer.equals(cluster.getLeader().getLeaderId());
}
return false;
}, 10_000));
verify(raftGrpEvtsLsnr, times(1)).onNewPeersConfigurationApplied(List.of(newPeer), List.of(newLearner));
}
}
@Test
public void testChangePeersOnLeaderElected() throws Exception {
List<TestPeer> peers = IntStream.range(0, 6)
.mapToObj(i -> new TestPeer(testInfo, TestUtils.INIT_PORT + i))
.collect(toList());
cluster = new TestCluster("testChangePeers", dataPath, peers, testInfo);
var raftGrpEvtsLsnr = mock(JraftGroupEventsListener.class);
cluster.setRaftGrpEvtsLsnr(raftGrpEvtsLsnr);
for (TestPeer p : peers) {
assertTrue(cluster.start(p, false, 300));
}
cluster.waitAndGetLeader();
verify(raftGrpEvtsLsnr, times(1)).onLeaderElected(anyLong());
cluster.stop(cluster.getLeader().getLeaderId());
cluster.waitAndGetLeader();
verify(raftGrpEvtsLsnr, times(2)).onLeaderElected(anyLong());
cluster.stop(cluster.getLeader().getLeaderId());
cluster.waitAndGetLeader();
verify(raftGrpEvtsLsnr, times(3)).onLeaderElected(anyLong());
}
@Test
public void changePeersAsyncResponses() throws Exception {
TestPeer peer0 = new TestPeer(testInfo, TestUtils.INIT_PORT);
cluster = new TestCluster("testChangePeers", dataPath, Collections.singletonList(peer0), testInfo);
assertTrue(cluster.start(peer0));
Node leader = cluster.waitAndGetLeader();
sendTestTaskAndWait(leader);
TestPeer peer = new TestPeer(testInfo, TestUtils.INIT_PORT + 1);
assertTrue(cluster.start(peer, false, 300));
leader = cluster.waitAndGetLeader();
assertNotNull(leader);
PeerId leaderPeer = peer0.getPeerId();
assertEquals(leaderPeer, leader.getNodeId().getPeerId());
TestPeer newLeaderPeer = new TestPeer(testInfo, peer0.getPort() + 1);
// wrong leader term, do nothing
SynchronizedClosure done = new SynchronizedClosure();
leader.changePeersAsync(new Configuration(Collections.singletonList(newLeaderPeer.getPeerId())),
leader.getCurrentTerm() - 1, done);
assertEquals(done.await(), Status.OK());
// the same config, do nothing
done = new SynchronizedClosure();
leader.changePeersAsync(new Configuration(Collections.singletonList(leaderPeer)),
leader.getCurrentTerm(), done);
assertEquals(done.await(), Status.OK());
// change peer to new conf containing only new node
done = new SynchronizedClosure();
leader.changePeersAsync(new Configuration(Collections.singletonList(newLeaderPeer.getPeerId())),
leader.getCurrentTerm(), done);
assertEquals(done.await(), Status.OK());
assertTrue(waitForCondition(() -> {
if (cluster.getLeader() != null)
return newLeaderPeer.getPeerId().equals(cluster.getLeader().getLeaderId());
return false;
}, 10_000));
for (MockStateMachine fsm : cluster.getFsms()) {
assertEquals(10, fsm.getLogs().size());
}
// check concurrent start of two async change peers.
Node newLeader = cluster.getLeader();
sendTestTaskAndWait(newLeader);
ExecutorService executor = Executors.newFixedThreadPool(10);
List<SynchronizedClosure> dones = new ArrayList<>();
List<Future> futs = new ArrayList<>();
for (int i = 0; i < 2; i++) {
SynchronizedClosure newDone = new SynchronizedClosure();
dones.add(newDone);
futs.add(executor.submit(() -> {
newLeader.changePeersAsync(new Configuration(Collections.singletonList(peer0.getPeerId())), 2, newDone);
}));
}
futs.get(0).get();
futs.get(1).get();
Status firstDoneStatus = dones.get(0).await();
Status secondDoneStatus = dones.get(1).await();
assertThat(
List.of(firstDoneStatus.getRaftError(), secondDoneStatus.getRaftError()),
containsInAnyOrder(RaftError.SUCCESS, RaftError.EBUSY)
);
assertTrue(waitForCondition(() -> {
if (cluster.getLeader() != null)
return peer0.getPeerId().equals(cluster.getLeader().getLeaderId());
return false;
}, 10_000));
for (MockStateMachine fsm : cluster.getFsms()) {
assertEquals(20, fsm.getLogs().size());
}
}
@Test
public void testChangePeersAddMultiNodes() throws Exception {
List<TestPeer> peers = new ArrayList<>();
TestPeer peer0 = new TestPeer(testInfo, TestUtils.INIT_PORT);
peers.add(peer0);
cluster = new TestCluster("testChangePeersAddMultiNodes", dataPath, Collections.singletonList(peer0), testInfo);
assertTrue(cluster.start(peer0));
Node leader = cluster.waitAndGetLeader();
sendTestTaskAndWait(leader);
Configuration conf = new Configuration();
conf.addPeer(peer0.getPeerId());
for (int i = 1; i < 3; i++) {
TestPeer peer = new TestPeer(testInfo, TestUtils.INIT_PORT + i);
peers.add(peer);
conf.addPeer(peer.getPeerId());
}
TestPeer peer = peers.get(1);
// fail, because the peers are not started.
SynchronizedClosure done = new SynchronizedClosure();
leader.changePeers(new Configuration(Collections.singletonList(peer.getPeerId())), done);
assertEquals(RaftError.ECATCHUP, done.await().getRaftError());
// start peer1
assertTrue(cluster.start(peer));
// still fail, because peer2 is not started
done.reset();
leader.changePeers(conf, done);
assertEquals(RaftError.ECATCHUP, done.await().getRaftError());
// start peer2
peer = peers.get(2);
assertTrue(cluster.start(peer));
done.reset();
// works
leader.changePeers(conf, done);
Status await = done.await();
assertTrue(await.isOk(), await.getErrorMsg());
cluster.ensureSame();
assertEquals(3, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms())
assertEquals(10, fsm.getLogs().size());
}
@Test
public void testChangePeersStepsDownInJointConsensus() throws Exception {
List<TestPeer> peers = new ArrayList<>();
TestPeer peer0 = new TestPeer(testInfo, 5006);
TestPeer peer1 = new TestPeer(testInfo, 5007);
TestPeer peer2 = new TestPeer(testInfo, 5008);
TestPeer peer3 = new TestPeer(testInfo, 5009);
// start single cluster
peers.add(peer0);
cluster = new TestCluster("testChangePeersStepsDownInJointConsensus", dataPath, peers, testInfo);
assertTrue(cluster.start(peer0));
Node leader = cluster.waitAndGetLeader();
assertNotNull(leader);
sendTestTaskAndWait(leader);
// start peer1-3
assertTrue(cluster.start(peer1));
assertTrue(cluster.start(peer2));
assertTrue(cluster.start(peer3));
// Make sure the topology is ready before adding peers.
assertTrue(waitForTopology(cluster, leader.getNodeId().getPeerId(), 4, 3_000));
Configuration conf = new Configuration();
conf.addPeer(peer0.getPeerId());
conf.addPeer(peer1.getPeerId());
conf.addPeer(peer2.getPeerId());
conf.addPeer(peer3.getPeerId());
// change peers
SynchronizedClosure done = new SynchronizedClosure();
leader.changePeers(conf, done);
assertTrue(done.await().isOk());
// stop peer3
assertTrue(cluster.stop(peer3.getPeerId()));
conf.removePeer(peer0.getPeerId());
conf.removePeer(peer1.getPeerId());
// Change peers to [peer2, peer3], which must fail since peer3 is stopped
done.reset();
leader.changePeers(conf, done);
assertEquals(RaftError.EPERM, done.await().getRaftError());
log.info(done.getStatus().toString());
assertFalse(((NodeImpl) leader).getConf().isStable());
leader = cluster.getLeader();
assertNull(leader);
assertTrue(cluster.start(peer3));
Thread.sleep(1000);
leader = cluster.waitAndGetLeader(Set.of(peer2.getPeerId(), peer3.getPeerId()));
List<PeerId> thePeers = leader.listPeers();
assertTrue(!thePeers.isEmpty());
assertEquals(conf.getPeerSet(), new HashSet<>(thePeers));
}
static class ChangeArg {
TestCluster c;
List<PeerId> peers;
volatile boolean stop;
boolean dontRemoveFirstPeer;
ChangeArg(TestCluster c, List<PeerId> peers, boolean stop,
boolean dontRemoveFirstPeer) {
super();
this.c = c;
this.peers = peers;
this.stop = stop;
this.dontRemoveFirstPeer = dontRemoveFirstPeer;
}
}
private Future<?> startChangePeersThread(ChangeArg arg) {
Set<RaftError> expectedErrors = new HashSet<>();
expectedErrors.add(RaftError.EBUSY);
expectedErrors.add(RaftError.EPERM);
expectedErrors.add(RaftError.ECATCHUP);
ExecutorService executor = Executors.newSingleThreadExecutor();
executors.add(executor);
return Utils.runInThread(executor, () -> {
try {
while (!arg.stop) {
Node leader = arg.c.waitAndGetLeader();
if (leader == null)
continue;
// select peers in random
Configuration conf = new Configuration();
if (arg.dontRemoveFirstPeer)
conf.addPeer(arg.peers.get(0));
for (int i = 0; i < arg.peers.size(); i++) {
boolean select = ThreadLocalRandom.current().nextInt(64) < 32;
if (select && !conf.contains(arg.peers.get(i)))
conf.addPeer(arg.peers.get(i));
}
if (conf.isEmpty()) {
log.warn("No peer has been selected");
continue;
}
SynchronizedClosure done = new SynchronizedClosure();
leader.changePeers(conf, done);
done.await();
assertTrue(done.getStatus().isOk() || expectedErrors.contains(done.getStatus().getRaftError()), done.getStatus().toString());
}
}
catch (InterruptedException e) {
log.error("ChangePeersThread is interrupted", e);
}
});
}
@Test
public void testChangePeersChaosWithSnapshot() throws Exception {
// start cluster
List<TestPeer> peers = new ArrayList<>();
peers.add(new TestPeer(testInfo, TestUtils.INIT_PORT));
cluster = new TestCluster("testChangePeersChaosWithSnapshot", dataPath, peers, ELECTION_TIMEOUT_MILLIS, testInfo);
assertTrue(cluster.start(peers.get(0), false, 2));
// start other peers
for (int i = 1; i < 10; i++) {
TestPeer peer = new TestPeer(testInfo, TestUtils.INIT_PORT + i);
peers.add(peer);
assertTrue(cluster.start(peer));
}
ChangeArg arg = new ChangeArg(cluster, peers.stream().map(TestPeer::getPeerId).collect(toList()), false, false);
Future<?> future = startChangePeersThread(arg);
for (int i = 0; i < 5000; ) {
Node leader = cluster.waitAndGetLeader();
if (leader == null)
continue;
SynchronizedClosure done = new SynchronizedClosure();
Task task = new Task(ByteBuffer.wrap(("hello" + i).getBytes(UTF_8)), done);
leader.apply(task);
Status status = done.await();
if (status.isOk()) {
if (++i % 100 == 0)
System.out.println("Progress:" + i);
}
else
assertEquals(RaftError.EPERM, status.getRaftError());
}
arg.stop = true;
future.get();
SynchronizedClosure done = new SynchronizedClosure();
Node leader = cluster.waitAndGetLeader();
leader.changePeers(new Configuration(peers.stream().map(TestPeer::getPeerId).collect(toList())), done);
Status st = done.await();
assertTrue(st.isOk(), st.getErrorMsg());
cluster.ensureSame();
assertEquals(10, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms())
assertTrue(fsm.getLogs().size() >= 5000);
}
@Test
public void testChangePeersChaosWithoutSnapshot() throws Exception {
// start cluster
List<TestPeer> peers = new ArrayList<>();
peers.add(new TestPeer(testInfo, TestUtils.INIT_PORT));
cluster = new TestCluster("testChangePeersChaosWithoutSnapshot", dataPath, peers, ELECTION_TIMEOUT_MILLIS, testInfo);
assertTrue(cluster.start(peers.get(0), false, 100000));
// start other peers
for (int i = 1; i < 10; i++) {
TestPeer peer = new TestPeer(testInfo, TestUtils.INIT_PORT + i);
peers.add(peer);
assertTrue(cluster.start(peer, false, 10000));
}
ChangeArg arg = new ChangeArg(cluster, peers.stream().map(TestPeer::getPeerId).collect(toList()), false, true);
Future<?> future = startChangePeersThread(arg);
final int tasks = 5000;
for (int i = 0; i < tasks; ) {
Node leader = cluster.waitAndGetLeader();
if (leader == null)
continue;
SynchronizedClosure done = new SynchronizedClosure();
Task task = new Task(ByteBuffer.wrap(("hello" + i).getBytes(UTF_8)), done);
leader.apply(task);
Status status = done.await();
if (status.isOk()) {
if (++i % 100 == 0)
System.out.println("Progress:" + i);
}
else
assertEquals(RaftError.EPERM, status.getRaftError());
}
arg.stop = true;
future.get();
SynchronizedClosure done = new SynchronizedClosure();
Node leader = cluster.waitAndGetLeader();
leader.changePeers(new Configuration(peers.stream().map(TestPeer::getPeerId).collect(toList())), done);
assertTrue(done.await().isOk());
cluster.ensureSame();
assertEquals(10, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms()) {
final int logSize = fsm.getLogs().size();
assertTrue(logSize >= tasks, "logSize=" + logSize);
}
}
@Test
public void testChangePeersChaosApplyTasks() throws Exception {
// start cluster
List<TestPeer> peers = new ArrayList<>();
peers.add(new TestPeer(testInfo, TestUtils.INIT_PORT));
cluster = new TestCluster("testChangePeersChaosApplyTasks", dataPath, peers, ELECTION_TIMEOUT_MILLIS, testInfo);
assertTrue(cluster.start(peers.get(0), false, 100000));
// start other peers
for (int i = 1; i < 10; i++) {
TestPeer peer = new TestPeer(testInfo, TestUtils.INIT_PORT + i);
peers.add(peer);
assertTrue(cluster.start(peer, false, 100000));
}
final int threads = 3;
List<ChangeArg> args = new ArrayList<>();
List<Future<?>> futures = new ArrayList<>();
CountDownLatch latch = new CountDownLatch(threads);
ExecutorService executor = Executors.newFixedThreadPool(threads);
executors.add(executor);
for (int t = 0; t < threads; t++) {
ChangeArg arg = new ChangeArg(cluster, peers.stream().map(TestPeer::getPeerId).collect(toList()), false, true);
args.add(arg);
futures.add(startChangePeersThread(arg));
Utils.runInThread(executor, () -> {
try {
for (int i = 0; i < 5000; ) {
Node leader = cluster.waitAndGetLeader();
if (leader == null)
continue;
SynchronizedClosure done = new SynchronizedClosure();
Task task = new Task(ByteBuffer.wrap(("hello" + i).getBytes(UTF_8)), done);
leader.apply(task);
Status status = done.await();
if (status.isOk()) {
if (++i % 100 == 0)
System.out.println("Progress:" + i);
}
else
assertEquals(RaftError.EPERM, status.getRaftError());
}
}
catch (Exception e) {
log.error("Failed to run tasks", e);
}
finally {
latch.countDown();
}
});
}
latch.await();
for (ChangeArg arg : args)
arg.stop = true;
for (Future<?> future : futures)
future.get();
SynchronizedClosure done = new SynchronizedClosure();
Node leader = cluster.waitAndGetLeader();
leader.changePeers(new Configuration(peers.stream().map(TestPeer::getPeerId).collect(toList())), done);
assertTrue(done.await().isOk());
cluster.ensureSame();
assertEquals(10, cluster.getFsms().size());
for (MockStateMachine fsm : cluster.getFsms()) {
int logSize = fsm.getLogs().size();
assertTrue(logSize >= 5000 * threads, "logSize= " + logSize);
}
}
@Test
public void testBlockedElection() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unittest", dataPath, peers, testInfo);
for (TestPeer peer : peers)
assertTrue(cluster.start(peer));
Node leader = cluster.waitAndGetLeader();
log.warn("Current leader {}, electTimeout={}", leader.getNodeId().getPeerId(), leader.getOptions().getElectionTimeoutMs());
List<Node> followers = cluster.getFollowers();
blockMessagesOnFollowers(followers, (msg, nodeId) -> {
if (msg instanceof RpcRequests.RequestVoteRequest) {
RpcRequests.RequestVoteRequest msg0 = (RpcRequests.RequestVoteRequest) msg;
return !msg0.preVote();
}
return false;
});
log.warn("Stop leader {}, curTerm={}", leader.getNodeId().getPeerId(), ((NodeImpl) leader).getCurrentTerm());
assertTrue(cluster.stop(leader.getNodeId().getPeerId()));
assertNull(cluster.getLeader());
Thread.sleep(2000);
assertNull(cluster.getLeader());
stopBlockingMessagesOnFollowers(followers);
// elect new leader
leader = cluster.waitAndGetLeader();
log.info("Elect new leader is {}, curTerm={}", leader.getLeaderId(), ((NodeImpl) leader).getCurrentTerm());
}
@Test
public void testElectionTimeoutAutoAdjustWhenBlockedAllMessages() throws Exception {
testElectionTimeoutAutoAdjustWhenBlockedMessages((msg, nodeId) -> true);
}
@Test
public void testElectionTimeoutAutoAdjustWhenBlockedRequestVoteMessages() throws Exception {
testElectionTimeoutAutoAdjustWhenBlockedMessages((msg, nodeId) -> {
if (msg instanceof RpcRequests.RequestVoteRequest) {
RpcRequests.RequestVoteRequest msg0 = (RpcRequests.RequestVoteRequest) msg;
return !msg0.preVote();
}
return false;
});
}
private void testElectionTimeoutAutoAdjustWhenBlockedMessages(BiPredicate<Object, String> blockingPredicate) throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 4);
int maxElectionRoundsWithoutAdjusting = 3;
cluster = new TestCluster("unittest", dataPath, peers, new LinkedHashSet<>(), ELECTION_TIMEOUT_MILLIS,
opts -> opts.setElectionTimeoutStrategy(new ExponentialBackoffTimeoutStrategy(11_000, maxElectionRoundsWithoutAdjusting)),
testInfo);
for (TestPeer peer : peers) {
assertTrue(cluster.start(peer));
}
Node leader = cluster.waitAndGetLeader();
int initElectionTimeout = leader.getOptions().getElectionTimeoutMs();
log.warn("Current leader {}, electTimeout={}", leader.getNodeId().getPeerId(), leader.getOptions().getElectionTimeoutMs());
List<Node> followers = cluster.getFollowers();
for (Node follower : followers) {
NodeImpl follower0 = (NodeImpl) follower;
assertEquals(initElectionTimeout, follower0.getOptions().getElectionTimeoutMs());
}
blockMessagesOnFollowers(followers, blockingPredicate);
log.warn("Stop leader {}, curTerm={}", leader.getNodeId().getPeerId(), ((NodeImpl) leader).getCurrentTerm());
assertTrue(cluster.stop(leader.getNodeId().getPeerId()));
assertNull(cluster.getLeader());
assertTrue(waitForCondition(() -> followers.stream().allMatch(f -> f.getOptions().getElectionTimeoutMs() > initElectionTimeout),
(long) maxElectionRoundsWithoutAdjusting
// need to multiply to 2 because stepDown happens after voteTimer timeout
* (initElectionTimeout + followers.get(0).getOptions().getRaftOptions().getMaxElectionDelayMs()) * 2));
stopBlockingMessagesOnFollowers(followers);
// elect new leader
leader = cluster.waitAndGetLeader();
log.info("Elected new leader is {}, curTerm={}", leader.getLeaderId(), ((NodeImpl) leader).getCurrentTerm());
assertTrue(
waitForCondition(() -> followers.stream().allMatch(f -> f.getOptions().getElectionTimeoutMs() == initElectionTimeout),
3_000));
}
/**
* Tests if a read using leader leases works correctly after previous leader segmentation.
*/
@Test
public void testLeaseReadAfterSegmentation() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
cluster = new TestCluster("unittest", dataPath, peers, 3_000, testInfo);
for (TestPeer peer : peers) {
RaftOptions opts = new RaftOptions();
opts.setElectionHeartbeatFactor(2); // Election timeout divisor.
opts.setReadOnlyOptions(ReadOnlyOption.ReadOnlyLeaseBased);
assertTrue(cluster.start(peer, false, 300, false, null, opts));
}
NodeImpl leader = (NodeImpl) cluster.waitAndGetLeader();
assertNotNull(leader);
cluster.ensureLeader(leader);
sendTestTaskAndWait(leader);
cluster.ensureSame();
DefaultRaftClientService rpcService = (DefaultRaftClientService) leader.getRpcClientService();
RpcClientEx rpcClientEx = (RpcClientEx) rpcService.getRpcClient();
AtomicInteger cnt = new AtomicInteger();
rpcClientEx.blockMessages((msg, nodeId) -> {
assertTrue(msg instanceof RpcRequests.AppendEntriesRequest);
if (cnt.get() >= 2)
return true;
log.info("Send heartbeat: " + msg + " to " + nodeId);
cnt.incrementAndGet();
return false;
});
assertTrue(waitForCondition(() -> {
Node currentLeader = cluster.getLeader();
return currentLeader != null && !leader.getNodeId().equals(currentLeader.getNodeId());
}, 10_000));
CompletableFuture<Status> res = new CompletableFuture<>();
cluster.getLeader().readIndex(null, new ReadIndexClosure() {
@Override public void run(Status status, long index, byte[] reqCtx) {
res.complete(status);
}
});
assertTrue(res.get().isOk());
}
/**
* Tests propagation of HLC on heartbeat request and response.
*/
@Test
public void testHlcPropagation() throws Exception {
List<TestPeer> peers = TestUtils.generatePeers(testInfo, 2);
cluster = new TestCluster("unitest", dataPath, peers, 3_000, testInfo);
for (TestPeer peer : peers) {
RaftOptions opts = new RaftOptions();
opts.setElectionHeartbeatFactor(4); // Election timeout divisor.
HybridClock clock = new HybridClockImpl();
assertTrue(cluster.start(peer, false, 300, false, null, opts, clock));
}
List<NodeImpl> nodes = cluster.getNodes();
for (NodeImpl node : nodes) {
RpcClientEx rpcClientEx = sender(node);
rpcClientEx.recordMessages((msg, nodeId) -> {
if (msg instanceof AppendEntriesRequestImpl ||
msg instanceof AppendEntriesResponseImpl) {
return true;
}
return false;
});
}
Node leader = cluster.waitAndGetLeader();
cluster.ensureLeader(leader);
RpcClientEx client = sender(leader);
AtomicBoolean heartbeatRequest = new AtomicBoolean(false);
AtomicBoolean appendEntriesRequest = new AtomicBoolean(false);
AtomicBoolean heartbeatResponse = new AtomicBoolean(false);
AtomicBoolean appendEntriesResponse = new AtomicBoolean(false);
waitForCondition(() -> {
client.recordedMessages().forEach(msgs -> {
if (msgs[0] instanceof AppendEntriesRequestImpl) {
AppendEntriesRequestImpl msg = (AppendEntriesRequestImpl) msgs[0];
if (msg.entriesList() == null && msg.data() == null) {
heartbeatRequest.set(true);
} else {
appendEntriesRequest.set(true);
}
assertTrue(msg.timestamp() != null);
} else if (msgs[0] instanceof AppendEntriesResponseImpl) {
AppendEntriesResponseImpl msg = (AppendEntriesResponseImpl) msgs[0];
if (msg.timestamp() == null) {
appendEntriesResponse.set(true);
} else {
heartbeatResponse.set(true);
}
}
});
return heartbeatRequest.get() &&
appendEntriesRequest.get() &&
heartbeatResponse.get() &&
appendEntriesResponse.get();
},
5000);
assertTrue(heartbeatRequest.get());
assertTrue(appendEntriesRequest.get());
assertTrue(heartbeatResponse.get());
assertTrue(appendEntriesResponse.get());
}
private NodeOptions createNodeOptions(int nodeIdx) {
NodeOptions options = new NodeOptions();
DefaultLogStorageFactory log = new DefaultLogStorageFactory(Path.of(dataPath, "node" + nodeIdx, "log"));
log.start();
options.setServiceFactory(new IgniteJraftServiceFactory(log));
options.setLogUri("test");
return options;
}
/**
* TODO asch get rid of waiting for topology IGNITE-14832
*
* @param cluster
* @param peerId
* @param expected
* @param timeout
* @return
*/
private static boolean waitForTopology(TestCluster cluster, PeerId peerId, int expected, long timeout) {
RaftGroupService grp = cluster.getServer(peerId);
if (grp == null) {
log.warn("Node has not been found {}", peerId);
return false;
}
RpcServer rpcServer = grp.getRpcServer();
if (!(rpcServer instanceof IgniteRpcServer))
return true;
ClusterService service = ((IgniteRpcServer) grp.getRpcServer()).clusterService();
long stop = System.currentTimeMillis() + timeout;
while (System.currentTimeMillis() < stop) {
if (service.topologyService().allMembers().size() >= expected)
return true;
try {
Thread.sleep(50);
}
catch (InterruptedException e) {
return false;
}
}
return false;
}
/**
* @param cond The condition.
* @param timeout The timeout.
* @return {@code True} if the condition is satisfied.
*/
private boolean waitForCondition(BooleanSupplier cond, long timeout) {
long stop = System.currentTimeMillis() + timeout;
while (System.currentTimeMillis() < stop) {
if (cond.getAsBoolean())
return true;
try {
Thread.sleep(50);
}
catch (InterruptedException e) {
return false;
}
}
return false;
}
/**
* @param groupId Group id.
* @param peer Peer.
* @param nodeOptions Node options.
* @return Raft group service.
*/
private RaftGroupService createService(String groupId, TestPeer peer, NodeOptions nodeOptions, Collection<TestPeer> peers) {
nodeOptions.setStripes(1);
nodeOptions.setLogStripesCount(1);
List<NetworkAddress> addressList = peers.stream()
.map(p -> new NetworkAddress(TestUtils.getLocalAddress(), p.getPort()))
.collect(toList());
var nodeManager = new NodeManager();
ClusterService clusterService = ClusterServiceTestUtils.clusterService(
testInfo,
peer.getPort(),
new StaticNodeFinder(addressList)
);
ExecutorService requestExecutor = JRaftUtils.createRequestExecutor(nodeOptions);
executors.add(requestExecutor);
IgniteRpcServer rpcServer = new TestIgniteRpcServer(clusterService, nodeManager, nodeOptions, requestExecutor);
nodeOptions.setRpcClient(new IgniteRpcClient(clusterService));
nodeOptions.setCommandsMarshaller(TestCluster.commandsMarshaller(clusterService));
assertThat(clusterService.startAsync(), willCompleteSuccessfully());
var service = new RaftGroupService(groupId, peer.getPeerId(), nodeOptions, rpcServer, nodeManager) {
@Override public synchronized void shutdown() {
rpcServer.shutdown();
super.shutdown();
assertThat(clusterService.stopAsync(), willCompleteSuccessfully());
}
};
services.add(service);
return service;
}
private void sendTestTaskAndWait(Node node) throws InterruptedException {
this.sendTestTaskAndWait(node, 0, 10, RaftError.SUCCESS);
}
private void sendTestTaskAndWait(Node node, int amount) throws InterruptedException {
this.sendTestTaskAndWait(node, 0, amount, RaftError.SUCCESS);
}
private void sendTestTaskAndWait(Node node, RaftError err) throws InterruptedException {
this.sendTestTaskAndWait(node, 0, 10, err);
}
// Note that waiting for the latch when tasks are applying doesn't guarantee that FSMCallerImpl.lastAppliedIndex
// will be updated immediately.
private void sendTestTaskAndWait(Node node, int start, int amount,
RaftError err) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(amount);
for (int i = start; i < start + amount; i++) {
ByteBuffer data = ByteBuffer.wrap(("hello" + i).getBytes(UTF_8));
Task task = new Task(data, new ExpectClosure(err, latch));
node.apply(task);
}
waitLatch(latch);
}
private void sendTestTaskAndWait(Node node, int start,
RaftError err) throws InterruptedException {
sendTestTaskAndWait(node, start, 10, err);
}
@SuppressWarnings("SameParameterValue")
private int sendTestTaskAndWait(String prefix, Node node, int amount,
int code) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(10);
final AtomicInteger successCount = new AtomicInteger(0);
for (int i = 0; i < amount; i++) {
ByteBuffer data = ByteBuffer.wrap((prefix + i).getBytes(UTF_8));
Task task = new Task(data, new ExpectClosure(code, null, latch, successCount));
node.apply(task);
}
waitLatch(latch);
return successCount.get();
}
private void triggerLeaderSnapshot(TestCluster cluster, Node leader) throws InterruptedException {
triggerLeaderSnapshot(cluster, leader, 1);
}
private void triggerLeaderSnapshot(TestCluster cluster, Node leader, int times)
throws InterruptedException {
// trigger leader snapshot
// first snapshot will be triggered randomly
int snapshotTimes = cluster.getLeaderFsm().getSaveSnapshotTimes();
assertTrue(snapshotTimes == times - 1 || snapshotTimes == times, "snapshotTimes=" + snapshotTimes + ", times=" + times);
CountDownLatch latch = new CountDownLatch(1);
leader.snapshot(new ExpectClosure(latch));
waitLatch(latch);
assertEquals(snapshotTimes + 1, cluster.getLeaderFsm().getSaveSnapshotTimes());
}
private void waitLatch(CountDownLatch latch) throws InterruptedException {
assertTrue(latch.await(30, TimeUnit.SECONDS));
}
@SuppressWarnings({"unused", "SameParameterValue"})
private boolean assertReadIndex(Node node, int index) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
byte[] requestContext = TestUtils.getRandomBytes();
AtomicBoolean success = new AtomicBoolean(false);
node.readIndex(requestContext, new ReadIndexClosure() {
@Override
public void run(Status status, long theIndex, byte[] reqCtx) {
if (status.isOk()) {
assertEquals(index, theIndex);
assertArrayEquals(requestContext, reqCtx);
success.set(true);
}
else {
assertTrue(status.getErrorMsg().contains("RPC exception:Check connection["), status.getErrorMsg());
assertTrue(status.getErrorMsg().contains("] fail and try to create new one"), status.getErrorMsg());
}
latch.countDown();
}
});
latch.await();
return success.get();
}
private void blockMessagesOnFollowers(List<Node> followers, BiPredicate<Object, String> blockingPredicate) {
for (Node follower : followers) {
RpcClientEx rpcClientEx = sender(follower);
rpcClientEx.blockMessages(blockingPredicate);
}
}
private void stopBlockingMessagesOnFollowers(List<Node> followers) {
for (Node follower : followers) {
RpcClientEx rpcClientEx = sender(follower);
rpcClientEx.stopBlock();
}
}
static void waitForTopologyOnEveryNode(int count, TestCluster cluster) {
cluster.getAllNodes().forEach(peerId -> {
assertTrue(waitForTopology(cluster, peerId, count, TimeUnit.SECONDS.toMillis(10)));
});
}
private static TestPeer findById(Collection<TestPeer> peers, PeerId id) {
return peers.stream().filter(t -> t.getPeerId().equals(id)).findAny().orElseThrow();
}
}