+ * The cluster abstraction facilitates the creation of a cluster during the tests. This approach avoids the need to + * create actual {@link org.jgroups.JChannel} and complex configurations. This abstraction provides a simplified and + * controllable way to test the cluster, emitting events and sending messages. + *
+ * + *+ * The cluster abstraction works through {@link View} updates. Members are added and removed manually. When a new view + * is received, the members receive the update accordingly and follow the synchronous configuration. The user utilizing + * this abstraction has more control over when a member joins and which messages it receives. + *
+ * + * Our test suite contains examples of uses of the cluster abstraction. + * + * @since 1.0.12 + */ public abstract class MockRaftCluster { protected final Executor thread_pool=createThreadPool(1000); protected boolean async; + /** + * Emit the view update to all cluster members. + *+ * How the view is updated might vary from implementation. The basic idea is to iterate over all members and + * invoke each protocol in the stack to handle the view. + *
+ * + * @param view: The new {@link View} instance to update the member. + */ public abstract void handleView(View view); + /** + * Send a message in the cluster. + * + * @param msg: Message to send. + */ public abstract void send(Message msg); + /** + * The number of member in the cluster. + * + * @return The size of the cluster. + */ + public abstract int size(); + + /** + * Add a new member to the cluster. + * + *+ * The new member is associated with the given address. A member is resolved by the address when sending a message. + * Also, note that a view update is necessary to propagate the member addition. + *
+ * + * @param addr: The new member's address. + * @param node: The member abstraction wrapped in {@link RaftNode}. + * @return The fluent current class. + * @param+ * A view update is necessary to propagate the member removal. + *
+ * + * @param addr: The address of the member to remove. + * @return The fluent current class. + * @paramtrue if asynchronous, false, otherwise.
+ */
public boolean async() {return async;}
- public true to run asynchronous, false to run synchronous.
+ * @return The fluent current class.
+ * @param null if not found.
+ */
+ public static RAFT raft(JChannel ch) {
+ return ch.getProtocolStack().findProtocol(RAFT.class);
+ }
+
+ /**
+ * Retrieves the {@link BaseElection} protocol from the provided {@link JChannel}.
+ * The concrete type might vary according to the configured protocol stack.
+ *
+ * @param ch: The channel to search the protocol.
+ * @return The {@link BaseElection} instance or null if not found.
+ */
+ public static BaseElection election(JChannel ch) {
+ return ch.getProtocolStack().findProtocol(BaseElection.class);
+ }
+
+ /**
+ * Checks if the node is the current {@link RAFT} leader.
+ * + * This method search the protocol stack to retrieve the {@link RAFT} instance and then verifies if elected. + *
+ * + * @param ch: The channel to verify. + * @returntrue if currently the leader, and false otherwise
+ * or not found the {@link RAFT} protocol.
+ */
+ public static boolean isRaftLeader(JChannel ch) {
+ RAFT r = raft(ch);
+ return r.isLeader() && r.leader() != null && ch.getAddress().equals(r.leader());
+ }
+
+ /**
+ * Checks that given the time constraints, eventually {@link #isRaftLeader(JChannel)} is true.
+ *
+ * @param ch: Channel to verify.
+ * @param timeoutMs: The timeout in milliseconds.
+ * @return true if it became the leader, false, otherwise.
+ * @see #isRaftLeader(JChannel)
+ */
+ public static boolean eventuallyIsRaftLeader(JChannel ch, long timeoutMs) {
+ return eventually(() -> isRaftLeader(ch), timeoutMs, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Deletes all the replicated data and internal state for the given {@link RAFT} instance.
+ * + * This is useful when multiple tests for the same uses stable storage. After running each test the data can + * be deleted to no affect subsequent tests. + *
+ * + * @param r: {@link RAFT} instance to delete all information. + * @throws Exception: If an exception happens while deleting the data. + * @see Log#delete() + */ + public static void deleteRaftLog(RAFT r) throws Exception { + Log log = r != null ? r.log() : null; + if (log != null) { + log.delete(); + r.log(null); + } + } + + /** + * Verify that in the given time constraints, the expression returnstrue.
+ *
+ * @param bs: Boolean expression to verify.
+ * @param timeout: Timeout value.
+ * @param unit: Timeout unit.
+ * @return true if expression valid before time out, false, otherwise.
+ * @throws RuntimeException: If an exception is thrown while verifying the expression.
+ */
+ public static boolean eventually(BooleanSupplier bs, long timeout, TimeUnit unit) {
+ try {
+ long timeoutNanos = unit.toNanos(timeout);
+ // We want the sleep time to increase in arithmetic progression
+ // 30 loops with the default timeout of 30 seconds means the initial wait is ~ 65 millis
+ int loops = 30;
+ int progressionSum = loops * (loops + 1) / 2;
+ long initialSleepNanos = timeoutNanos / progressionSum;
+ long sleepNanos = initialSleepNanos;
+ long expectedEndTime = System.nanoTime() + timeoutNanos;
+ while (expectedEndTime - System.nanoTime() > 0) {
+ if (bs.getAsBoolean())
+ return true;
+ LockSupport.parkNanos(sleepNanos);
+ sleepNanos += initialSleepNanos;
+ }
+
+ return bs.getAsBoolean();
+ } catch (Exception e) {
+ throw new RuntimeException("Unexpected!", e);
+ }
+ }
+}
diff --git a/src/org/jgroups/raft/util/RequestTable.java b/src/org/jgroups/raft/util/RequestTable.java
index bcef9875..45fe5a2c 100644
--- a/src/org/jgroups/raft/util/RequestTable.java
+++ b/src/org/jgroups/raft/util/RequestTable.java
@@ -22,6 +22,10 @@ public class RequestTable
+ * This method should be invoked before setting the instance to null when the leader steps down.
+ * This provides a more responsive completion of requests to the users, instead of having requests time out.
+ * Internal operations, such as membership changes, do not have a timeout associated, which would hang and any
+ * subsequent changes would not complete.
+ *
+ * Warning: This should be used in tests only. + *
+ * + * @param r: RAFT instance to delete the log contents. + * @throws Exception: If an exception happens while deleting the log. + * @deprecated Use {@link RaftTestUtils#deleteRaftLog(RAFT)} instead. + */ + @Deprecated(since = "1.0.13", forRemoval = true) public static void deleteLog(RAFT r) throws Exception { - Log log=r != null? r.log() : null; - if(log != null) { - log.delete(); - r.log(null); - } + RaftTestUtils.deleteRaftLog(r); } } diff --git a/tests/junit-functional/org/jgroups/tests/AppendEntriesTest.java b/tests/junit-functional/org/jgroups/tests/AppendEntriesTest.java index 1a5e595f..3bfaeaab 100644 --- a/tests/junit-functional/org/jgroups/tests/AppendEntriesTest.java +++ b/tests/junit-functional/org/jgroups/tests/AppendEntriesTest.java @@ -6,35 +6,32 @@ import org.jgroups.protocols.DISCARD; import org.jgroups.protocols.TP; import org.jgroups.protocols.raft.AppendResult; -import org.jgroups.protocols.raft.ELECTION; import org.jgroups.protocols.raft.Log; import org.jgroups.protocols.raft.LogEntries; import org.jgroups.protocols.raft.LogEntry; import org.jgroups.protocols.raft.RAFT; -import org.jgroups.protocols.raft.REDIRECT; import org.jgroups.protocols.raft.RaftImpl; +import org.jgroups.protocols.raft.RaftLeaderException; import org.jgroups.raft.RaftHandle; import org.jgroups.raft.blocks.ReplicatedStateMachine; -import org.jgroups.raft.util.Utils; +import org.jgroups.raft.testfwk.RaftTestUtils; import org.jgroups.stack.ProtocolStack; +import org.jgroups.tests.harness.BaseStateMachineTest; import org.jgroups.util.Util; -import org.testng.annotations.AfterMethod; -import org.testng.annotations.Test; import java.util.Arrays; -import java.util.Collections; -import java.util.List; -import java.util.Objects; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.function.BooleanSupplier; -import java.util.function.Function; import java.util.stream.Collectors; import java.util.stream.Stream; -import static org.testng.Assert.assertEquals; -import static org.testng.Assert.assertFalse; -import static org.testng.Assert.assertTrue; +import org.assertj.core.api.Assertions; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.jgroups.raft.testfwk.RaftTestUtils.eventually; /** * Tests the AppendEntries functionality: appending log entries in regular operation, new members, late joiners etc @@ -42,27 +39,31 @@ * @since 0.2 */ @Test(groups=Global.FUNCTIONAL,singleThreaded=true) -public class AppendEntriesTest { - // A is always the leader because started first and logs are equal - protected JChannel a, b, c; - protected ReplicatedStateMachine+ * The cluster starts with {A, B}, then C joins, and {A, B, C} restarts. Information about C is still in {A, B}. + *
*/ public void testMembersRemainAfterRestart() throws Exception { - Class log=FileBasedLog.class; // In memory log will lose the snapshot on restart. - init(log, "A", "B"); - leader=leader(10000, 500, channels); + withClusterSize(2); + createCluster(); + + for (JChannel ch : channels()) { + RAFT r = raft(ch); + r.logClass(FileBasedLog.class.getName()); + } + + waitUntilAllRaftsHaveLeader(channels()); + Address leader=leaderAddress(); System.out.println("leader = " + leader); - assert leader != null : "Leader still null"; + assertThat(leader).isNotNull(); - waitUntilAllRaftsHaveLeader(channels); - assertSameLeader(leader, channels); - assertMembers(5000, 200, mbrs, 2, channels); + assertSameLeader(leader, channels()); + assertMembers(5000, getRaftMembers(), 2, channels()); RAFT raft=raft(leader); // Fill the log with some entries. @@ -140,49 +148,70 @@ public void testMembersRemainAfterRestart() throws Exception { f.get(10, TimeUnit.SECONDS); } - String new_mbr="X"; + String new_mbr="C"; System.out.printf("Adding [%s]\n", new_mbr); - raft.addServer(new_mbr); - mbrs.add(new_mbr); - channels=Arrays.copyOf(channels, channels.length+1); - channels[channels.length-1]=create(new_mbr, log); - assertMembers(10000, 200, mbrs, 2, channels); + withClusterSize(3); + createCluster(); + + // Also set 'C' to use file based log. + raft(2).logClass(FileBasedLog.class.getName()); + + raft.addServer(new_mbr).get(10, TimeUnit.SECONDS); + assertMembers(10000, getRaftMembers(), 2, channels()); System.out.println("\nShutdown cluster"); - for (JChannel channel : channels) { - RAFT r=raft(channel); + JChannel[] channels = channels(); + for (int i = 0; i < channels.length; i++) { + JChannel ch = channels[i]; + RAFT r = raft(ch); r.snapshot(); - channel.close(); + Util.close(ch); + channels[i] = null; } + destroyCluster(); System.out.println("\nRestarting cluster"); + // We restart with only 2 nodes. + withClusterSize(2); + createCluster(); + for (JChannel ch : channels()) { + RAFT r = raft(ch); + r.logClass(FileBasedLog.class.getName()); + } + // Nodes restart using the file configuration/previous configuration. - // Should restore member `X` from log. - init(log, "A", "B"); - assert !mbrs.contains(new_mbr) : "New member should not be in initial configuration"; + // Should restore member `C` from log. + assertThat(raft(0).members()) + .as("New member should not be in initial configuration") + .contains(new_mbr); - List+ * This class is abstract and does not enforce which type of cluster/communication is utilized between the nodes. It + * requires the implementation of the creation, destruction, and communication pieces instead. Therefore, nodes can + * utilize {@link org.jgroups.JChannel}, or {@link org.jgroups.raft.testfwk.MockRaftCluster} without implementing + * everything again. Some of the utilities provided: + * + *
+ * When utilizing the base class with a custom data provider factory, to receive the arguments properly, the + * {@link #passDataProviderParameters(Object[])} needs to be overridden. + *
+ * + * + * + *+ * The basic configuration in the class has: + *
+ * There is also a configuration by command-line arguments to enable tracing during the tests. To enable trace level + * globally, to all protocols in the stack, pass the variable {@link #ENABLE_GLOBAL_TRACE}, that is: + * + *
+ * $ mvn clean test -Dorg.jgroups.test.trace + *+ * + * We also provide an alias with + * + *
+ * $ mvn clean test -Dtrace + *+ * + * Any of the above will enable trace to all protocols. The output is not directed to the standard output, it can be + * found at the generated reports. To instead enable trace for specific classes, is necessary to use the complete + * variable {@link #ENABLE_TRACE_CLASSES}, passing the classes fully qualified name: + * + *
+ * $ mvn clean test -Dorg.jgroups.test.trace="org.jgroups.protocols.raft.RAFT" + *+ * + * This will enable trace level only for the {@link RAFT} class. It is possible to pass multiple classes separated + * by comma. + * + * + * @since 1.0.13 + */ +public abstract class AbstractRaftTest { + + /** + * Enable trace log level to all classes in the protocol stack. + */ + public static final String ENABLE_GLOBAL_TRACE = "org.jgrops.test.trace"; + + /** + * Enable trace log level to the specific list of classes given in FQN form and comma separated. + */ + public static final String ENABLE_TRACE_CLASSES = "org.jgroups.test.traces"; + + // Configurable parameters for sub-classes. + + /** + * Defines the size of the cluster, i.e., the number of nodes. + */ + protected int clusterSize = 1; + + /** + * Defines if the cluster should be cleared and re-created on a per-method basis. + */ + protected boolean recreatePerMethod = false; + + /** + * Defines if the cluster creation should happen manually. Therefore, both creation and clear happens manually. + */ + protected boolean createManually = false; + + /** + * {@link #clusterSize}. + * + * @param size: The cluster size to create. + * @return This test class. + */ + protected final
+ * This method limits the number of nodes created at once. Calling the method multiple times is safe, and the + * maximum {@link #clusterSize} is never violated. + *
+ * + * @param limit: The number of nodes to create at once. Never creates more than {@link #clusterSize} nodes. + * @throws Exception: If an error occur while creating the cluster. + */ + protected abstract void createCluster(int limit) throws Exception; + + /** + * Destroy the cluster and clears any open resource. + *+ * The implementors must invoke the appropriate {@link #beforeClusterDestroy()} and {@link #afterClusterDestroy()} + * methods to notify listeners. + *
+ * + * @throws Exception: If failed clearing the resources. + */ + + /** + * Destroy the cluster. + *+ * This method clears all resources allocated to the cluster. The cluster should be recreated again for use. + *
+ * + * @throws Exception: If an error happens while clearing the resources. + */ + protected abstract void destroyCluster() throws Exception; + + /** + * Creates the protocol stack to utilize during the tests. + * + * @param name: The name of node requesting the stack. + * @return The complete protocol stack. + * @throws Exception: If an error occur while creating the stack. + */ + protected abstract Protocol[] baseProtocolStackForNode(String name) throws Exception; + + /** + * Apply the log level configuration. + *+ * Check {@link AbstractRaftTest} about how to configure the log level. Note that, a log instance might be static, + * and since the tests run in parallel, this configuration might leak between tests! + *
+ * + * @param stack: The protocol stack in use. + */ + protected final void applyTraceConfiguration(Protocol[] stack) { + String[] traces = Arrays.stream(System.getProperty(ENABLE_TRACE_CLASSES, "").split(",")) + .filter(s -> !s.isEmpty()) + .toArray(String[]::new); + boolean globalTrace = Boolean.parseBoolean(System.getProperty(ENABLE_GLOBAL_TRACE, "false")); + + if (globalTrace) System.out.println("Enabling trace on all classes"); + else if (traces.length > 0) System.out.printf("Enabling trace on classes: %s%n", Arrays.toString(traces)); + + Set+ * This method can be invoked multiple times without harm in the tests. + *
+ * + *+ * The lifecycle methods are invoked when appropriately. The cluster lifecycle methods are always invoked. + *
+ * + * @throws Exception: If an error occur while creating the cluster. + */ + protected final void createCluster() throws Exception { + beforeClusterCreation(); + createCluster(clusterSize); + afterClusterCreation(); + } + + /** + * The cluster name to connect the channels. Defaults to the test class name. + * + * @return The cluster name. + */ + protected String clusterName() { + return getClass().getSimpleName(); + } + + /** + * Creates a new {@link RAFT} instance. + *+ * The default configuration has configured: + * + *
'A' and the {@link #clusterSize}+ * This method is final, to define which election algorithm to use, override {@link #createNewElection()}. + *
+ * + * @return A new and properly configured {@link BaseElection} concrete implementation. + * @throws Exception: If failed creating the election instance. + */ + protected final BaseElection createNewElectionAndDecorate() throws Exception { + BaseElection be = createNewElection(); + amendBaseElectionConfiguration(be); + return be; + } + + /** + * Creates the election algorithm to use. + *+ * The default election implementation is {@link ELECTION}. This method is not final and could be override to + * create a different instance. There is also the {@link #amendBaseElectionConfiguration(BaseElection)} to change + * the protocol configuration. + *
+ * + * @return A new and properly configured {@link BaseElection} concrete implementation. + */ + protected BaseElection createNewElection() throws Exception { + return new ELECTION(); + } + + /** + * Retrieves the members utilized in the {@link RAFT} configuration. + *
+ * This implementation offset from 'A' up to {@link #clusterSize} members. That is, for a size of
+ * 2, creates a list with 'A', 'B'.
+ *
+ * This method is invoked every time a new {@link RAFT} instance is created. + *
+ * + * @param raft: The {@link RAFT} instance with basic configuration. + * @see #createNewRaft(String) + */ + protected void amendRAFTConfiguration(RAFT raft) { } + + /** + * Entrypoint to change configuration of the {@link BaseElection} instance before start. + *+ * This method is invoked every time a new {@link BaseElection} instance is created, the actual concrete type + * might vary depending on the test. + *
+ * + * @param election: The instance with base configuration. + * @see #createNewElection() + */ + protected void amendBaseElectionConfiguration(BaseElection election) { } + + /** + * Necessary when utilizing the base class with a data provider factory. + * + * @param args: The data provider arguments. + */ + protected void passDataProviderParameters(Object[] args) { } + + /** + * Entrypoint executed before the cluster is created. + * This method might be invoked multiple times during a single test. + * + * @throws Exception: If an exception happens during the execution. + */ + protected void beforeClusterCreation() throws Exception { } + + /** + * Entrypoint executed before destroying the cluster. + * + * @throws Exception: If an exception happens during the execution. + */ + protected void beforeClusterDestroy() throws Exception { } + + /** + * Entrypoint executed after the cluster is created and running. + * This method can execute multiple times during a single test, and executes after all channels have the same view. + * + * @throws Exception: If an exception happens during execution. + * @see #beforeClusterCreation() + */ + protected void afterClusterCreation() throws Exception { } + + /** + * Entrypoint executed after the cluster is destroyed. + * + * @throws Exception: If an exception happens during execution. + */ + protected void afterClusterDestroy() throws Exception { } +} diff --git a/tests/junit-functional/org/jgroups/tests/harness/BaseRaftChannelTest.java b/tests/junit-functional/org/jgroups/tests/harness/BaseRaftChannelTest.java new file mode 100644 index 00000000..dd745bec --- /dev/null +++ b/tests/junit-functional/org/jgroups/tests/harness/BaseRaftChannelTest.java @@ -0,0 +1,367 @@ +package org.jgroups.tests.harness; + +import org.jgroups.Address; +import org.jgroups.JChannel; +import org.jgroups.protocols.raft.RAFT; +import org.jgroups.protocols.raft.REDIRECT; +import org.jgroups.protocols.raft.election.BaseElection; +import org.jgroups.raft.testfwk.RaftTestUtils; +import org.jgroups.stack.Protocol; +import org.jgroups.util.Util; + +import java.util.ArrayList; +import java.util.List; + +/** + * The base class for tests that utilize {@link JChannel} for communication. + *+ * This class already provides the mechanism to retrieve and access the protocols in the {@link JChannel}. Creating a + * cluster waits until all the nodes receive the same view. The cluster can be resized dynamically during a test + * execution, growing to accommodate more nodes, but not shrinking. + *
+ */ +public class BaseRaftChannelTest extends AbstractRaftTest { + + private JChannel[] channels; + + @Override + protected final void createCluster(int limit) throws Exception { + System.out.printf("%s: creating cluster '%s' with size %d (limited %d) %n", getClass(), clusterName(), clusterSize, limit); + + if (this.channels == null) this.channels = new JChannel[clusterSize]; + if (this.channels.length < clusterSize) { + JChannel[] updated = new JChannel[clusterSize]; + System.arraycopy(channels, 0, updated, 0, channels.length); + channels = updated; + } + + for (int i = 0; i < clusterSize; i++) { + + // The creation can be invoked multiple times to resize the cluster and add new members. + if (channels[i] != null) continue; + + String name = Character.toString('A' + i); + channels[i] = createChannel(name); + limit -= 1; + + if (limit == 0) break; + } + + Util.waitUntilAllChannelsHaveSameView(10_000, 100, actualChannels()); + System.out.printf("%s: current cluster %s of size %d view is:%n%s%n", getClass(), clusterName(), clusterSize, printCurrentCluster()); + } + + /** + * Creates a new {@link JChannel} using the configurations and attached to the lifecycle entrypoint. + * + * @param name: The {@link JChannel} name. + * @return A new instance of the {@link JChannel}. + * @throws Exception: If failed to create the channel. + */ + protected JChannel createChannel(String name) throws Exception { + JChannel ch = createDisconnectedChannel(name); + beforeChannelConnection(ch); + ch.connect(clusterName()); + afterChannelConnection(ch); + return ch; + } + + /** + * Creates a disconnected {@link JChannel}. + * + * @param name: The {@link JChannel} name. + * @return A disconnected new instance of the {@link JChannel}. + * @throws Exception: If failed to create the channel. + */ + protected JChannel createDisconnectedChannel(String name) throws Exception { + Protocol[] stack = baseProtocolStackForNode(name); + + applyTraceConfiguration(stack); + + JChannel ch = new JChannel(stack); + ch.name(name); + return ch; + } + + /** + * Destroy the complete cluster. + *+ * This iterates the {@link #channels} in reverse order to avoid any major disruption. Each channel is closed and + * set to null. This deletes the state information of the {@link RAFT} instance, deleting the log. All the + * created resources are cleared after this method executes. + *
+ * + * @throws Exception: If an error happens while clearing the resources. + */ + @Override + protected final void destroyCluster() throws Exception { + beforeClusterDestroy(); + + System.out.printf("%s: destroying cluster %s with size %d%n", getClass(), clusterName(), clusterSize); + if (channels != null) { + for (int i = clusterSize - 1; i >= 0; i--) { + close(i); + } + } + + this.channels = null; + afterClusterDestroy(); + } + + /** + * Close the {@link JChannel} in the given index. + *+ * The {@link JChannel} is closed and the {@link RAFT} state is deleted. + *
+ * + * @param index: The channel to close. + * @throws Exception: If an error happens while closing the resources. + */ + protected final void close(int index) throws Exception { + JChannel ch = channel(index); + if (ch == null) return; + + channels[index] = null; + close(ch); + } + + /** + * Closes the given channel and deletes the {@link RAFT} state. + * + * @param ch: The {@link JChannel} to close. + * @throws Exception: If an error happens while closing the resources. + */ + protected final void close(JChannel ch) throws Exception { + if (ch == null) return; + + if (channels != null) { + int i = 0; + for (JChannel curr : channels) { + if (curr == ch) { + channels[i] = null; + break; + } + i++; + } + } + + // Always close the channel before deleting the log. + Util.close(ch); + RaftTestUtils.deleteRaftLog(raft(ch)); + } + + /** + * The stack to utilize in the node during the test with a channel. + *+ * By default, besides decorating with the protocols from {@link Util#getTestStack(Protocol...)}, we use: + *
null if the channel was closed.
+ */
+ protected final JChannel channel(int index) {
+ assert index < channels.length : "Index out of bounds, maximum is " + (clusterSize - 1);
+ return channels[index];
+ }
+
+ /**
+ * Get the {@link JChannel} with the given address.
+ *
+ * @param addr: Channel address to match.
+ * @return The channel with the given address, or null, otherwise.
+ */
+ protected final JChannel channel(Address addr) {
+ for(JChannel ch: channels) {
+ if(ch.getAddress() != null && ch.getAddress().equals(addr))
+ return ch;
+ }
+ return null;
+ }
+
+ /**
+ * Get the created {@link RAFT} in the provided index.
+ *
+ * @param index: The index of the RAFT instance to retrieve.
+ * @return The {@link RAFT} in the specified position or null.
+ * @see #channel(int)
+ */
+ protected final RAFT raft(int index) {
+ JChannel ch = channel(index);
+ return ch == null ? null : raft(ch);
+ }
+
+ /**
+ * The {@link RAFT} instance in the provided {@link JChannel}.
+ *
+ * @param ch: The channel to retrieve the protocol.
+ * @return The RAFT protocol instance or null.
+ */
+ protected final RAFT raft(JChannel ch) {
+ return RaftTestUtils.raft(ch);
+ }
+
+ /**
+ * Search the first node identified as leader.
+ * As per the Raft specification, the algorithm can have more than one leader in different terms. If this method
+ * is invoked during a network split or during an election round, it could have fuzzy results. It is better suited
+ * to use when a stable topology is in place.
+ *
+ * @return The first node identified as an elected leader.
+ */
+ protected final RAFT leader() {
+ for (JChannel ch : channels) {
+ if (ch == null) continue;
+
+ RAFT r = raft(ch);
+ if (r.isLeader()) return r;
+ }
+
+ return null;
+ }
+
+ /**
+ * Finds the first node identified as leader and retrieves the address.
+ *
+ * @return The leader address or null.
+ */
+ protected final Address leaderAddress() {
+ RAFT r = leader();
+ return r == null ? null : r.getAddress();
+ }
+
+ /**
+ * Iterate over all nodes and return the ones which identify itself as leader.
+ *
+ * @return All nodes which sees itself as leader.
+ * @see #leader()
+ */
+ protected final Listnull entries.
+ *
+ * @return The complete array of channels.
+ */
+ protected final JChannel[] channels() {
+ return channels;
+ }
+
+ /**
+ * Iterate over all channels and remove any null entries.
+ *
+ * @return The array of currently active channels.
+ */
+ protected final JChannel[] actualChannels() {
+ List+ * This class implements the base utilities to retrieve the cluster members ({@link RaftNode}), access the cluster, + * and the stack protocols. + *
+ * + * @paramnull is returned, the cluster is not created.
+ */
+ protected abstract T createNewMockCluster();
+
+ @Override
+ protected final void createCluster(int limit) throws Exception {
+ System.out.printf("-- Creating mock cluster %s with size %s limited to %d%n", clusterName(), clusterSize, limit);
+
+ if (cluster == null) {
+ if ((cluster = createNewMockCluster()) == null) {
+ System.out.println("-- No cluster instance created!");
+ return;
+ }
+ }
+
+ // Maybe resizing would be necessary when dynamically expanding the cluster.
+ if (nodes == null) nodes = new RaftNode[clusterSize];
+
+ while (limit > 0 && cluster.size() < clusterSize) {
+ String name = Character.toString('A' + cluster.size());
+ Address address = createAddress(name);
+
+ RaftNode node = createNode(name, address);
+ node.init();
+ nodes[cluster.size()] = node;
+ cluster.add(address, node);
+ node.start();
+
+ limit -= 1;
+ }
+ }
+
+ @Override
+ protected final void destroyCluster() throws Exception {
+ beforeClusterDestroy();
+
+ System.out.printf("-- Destroying mock cluster %s named %s with size %d%n", cluster, clusterName(), clusterSize);
+
+ if (nodes == null) return;
+
+ for (int i = clusterSize - 1; i >= 0; i--) {
+ close(i);
+ }
+
+ cluster.clear();
+ cluster = null;
+ nodes = null;
+
+ afterClusterDestroy();
+ }
+
+ /**
+ * Close the member identified by the index.
+ *
+ * This method removes the {@link RaftNode} from the active member list, subsequent calls to {@link #node(int)}
+ * will return null. Also removes the member from the actual cluster view and deletes the
+ * {@link RAFT} persisted state.
+ *
null.
+ * @throws AssertionError: in case the {@param index} is greater that the {@link #clusterSize} or the cluster
+ * was not created.
+ */
+ protected final RaftNode node(int index) {
+ assert index < clusterSize : "Index out of bounds, maximum " + clusterSize;
+ assert nodes != null : "Cluster not created!";
+ return nodes[index];
+ }
+
+ /**
+ * Retrieves the {@link RAFT} instance from the node identified by the index.
+ *
+ * @param index: Index to retrieve the cluster member.
+ * @return A {@link RAFT} instance of null in case the member is not present.
+ * @see #node(int)
+ */
+ protected final RAFT raft(int index) {
+ RaftNode node = node(index);
+ return node == null ? null : node.raft();
+ }
+
+ /**
+ * Retrieves the {@link Address} of the member identified by the index.
+ *
+ * @param index: Index to retrieve the cluster member.
+ * @return The member's {@link Address} or null if it is not present in the cluster.
+ */
+ protected final Address address(int index) {
+ RaftNode node = node(index);
+ return node.raft().getAddress();
+ }
+
+ /**
+ * Retrieve all cluster members.
+ *
+ *
+ * The array might contain null entries.
+ *
null entries.
+ */
+ protected final RAFT[] rafts() {
+ return Arrays.stream(nodes)
+ .filter(Objects::nonNull)
+ .map(RaftNode::raft)
+ .toArray(RAFT[]::new);
+ }
+
+ /**
+ * Generates a message identifying the term and leader of each cluster member.
+ *
+ * @return A string containing a cluster-wide view of leader and term.
+ */
+ protected String dumpLeaderAndTerms() {
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < nodes.length; i++) {
+ RaftNode node = nodes[i];
+
+ if (node == null) {
+ sb.append(Character.toString('A' + i))
+ .append(" is null")
+ .append(System.lineSeparator());
+ continue;
+ }
+
+ RAFT raft = node.raft();
+ sb.append(raft.raftId())
+ .append(": (")
+ .append(raft.currentTerm())
+ .append(") with leader ")
+ .append(raft.leader())
+ .append(System.lineSeparator());
+ }
+ return sb.toString();
+ }
+
+ /**
+ * Retrieves the {@link RAFT} leader of each active node in the cluster.
+ *
+ * This method also removes the null entries.
+ *
+ * The method creates a new view containing all the active members identified by the indexes. If a member for + * a given index is not present in the cluster, it is not added to the final view. + *
+ * + *
+ * The index on the first entry, i.e., indexes[0], is the view coordinator.
+ *
P findProtocol(Protocol[] stack, Class
clazz) { + for (Protocol protocol : stack) { + if (clazz.isAssignableFrom(protocol.getClass())) + return (P) protocol; + } + return null; + } +} diff --git a/tests/junit-functional/org/jgroups/tests/harness/BaseRaftElectionTest.java b/tests/junit-functional/org/jgroups/tests/harness/BaseRaftElectionTest.java new file mode 100644 index 00000000..c7d88e60 --- /dev/null +++ b/tests/junit-functional/org/jgroups/tests/harness/BaseRaftElectionTest.java @@ -0,0 +1,362 @@ +package org.jgroups.tests.harness; + +import org.jgroups.JChannel; +import org.jgroups.protocols.raft.ELECTION; +import org.jgroups.protocols.raft.ELECTION2; +import org.jgroups.protocols.raft.RAFT; +import org.jgroups.protocols.raft.election.BaseElection; +import org.jgroups.raft.testfwk.MockRaftCluster; +import org.jgroups.raft.testfwk.RaftNode; +import org.jgroups.raft.testfwk.RaftTestUtils; + +import java.util.Arrays; +import java.util.Objects; +import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; +import java.util.function.Supplier; +import java.util.stream.Collectors; + +import org.testng.annotations.DataProvider; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Base class for election tests. + *
+ * This class itself is final, but it has two inner classes for tests based either on channels or the mock cluster. + * The {@link ChannelBased} or {@link ClusterBased}, respectively. Subclasses must choose which approach to use and + * extend one of the inner classes. + *
+ * + *+ * This class contains some utilities for election, like retrieving the protocol or waiting for something related to + * the election mechanism. There is no methods required for implementation. + *
+ * + * @since 1.0.13 + * @see AbstractRaftTest + */ +public final class BaseRaftElectionTest { + + /** + * A data provider to utilize with the {@link org.testng.annotations.Test} annotation. + *+ * This data provider pass as argument the classes which implement an election algorithm. + *
+ */ + public static final String ALL_ELECTION_CLASSES_PROVIDER = "all-election-classes"; + + private static Object[][] electionClasses() { + return new Object[][] { + {ELECTION.class}, + {ELECTION2.class}, + }; + } + + /** + * Asserts that something eventually evaluates totrue.
+ *
+ * @param timeout: The timeout in milliseconds to wait.
+ * @param bs: The expression to evaluate.
+ * @param message: The error message in case of failure.
+ */
+ private static void assertWaitUntil(long timeout, BooleanSupplier bs, Supplier+ * This method waits until a leader is elected between the given {@link RAFT}s instances. A leader is elected once: + *
+ * This internal class receives the data provider argument from {@link #ALL_ELECTION_CLASSES_PROVIDER} and + * creates a new instance. + *
+ */ + private static class ClassHandler { + protected Class electionClass; + + @SuppressWarnings("unchecked") + protected void setElectionClass(Object[] args) { + electionClass = (Class) args[0]; + } + + public BaseElection instantiate() throws Exception { + assert electionClass != null : "Election class not set"; + // The default constructor is always available. + return electionClass.getDeclaredConstructor().newInstance(); + } + } + + /** + * Base class for election tests based on {@link JChannel}. + *+ * This class builds upon {@link BaseRaftChannelTest} and offer utilities specific to the election process. + *
+ */ + public static class ChannelBased extends BaseRaftChannelTest { + + private final ClassHandler handler = new ClassHandler(); + + @Override + protected void passDataProviderParameters(Object[] args) { + handler.setElectionClass(args); + } + + @Override + protected BaseElection createNewElection() throws Exception { + return handler.instantiate(); + } + + /** + * Stop the voting thread manually. + */ + protected final void stopVotingThread() { + for (JChannel channel : channels()) { + if (channel == null) continue; + + BaseElection election = election(channel); + election.stopVotingThread(); + } + } + + /** + * Retrieves the concrete {@link BaseElection} implementation in the protocol stack. + * + * @param ch: Channel to retrieve the election protocol. + * @return The {@link BaseElection} concrete implementation ornull, if not found.
+ */
+ protected final BaseElection election(JChannel ch) {
+ return ch.getProtocolStack().findProtocol(handler.electionClass);
+ }
+
+ /**
+ * Wait until a leader is elected in the cluster.
+ * + * This methods blocks and waits until all the requested members in the cluster have an elected leader. + *
+ * + * @param timeout: Time in milliseconds to wait for an election. + * @param indexes: The index of the cluster members. + * @see #waitUntilLeaderElected(RAFT[], long) + */ + protected final void waitUntilLeaderElected(long timeout, int ... indexes) { + RAFT[] rafts = Arrays.stream(indexes) + .mapToObj(this::raft) + .filter(Objects::nonNull) + .toArray(RAFT[]::new); + BaseRaftElectionTest.waitUntilLeaderElected(rafts, timeout); + } + + /** + * Wait until the voting thread stops running on all election instances. + * + * @param timeout: Time in milliseconds to wait. + * @param indexes: The index of the cluster members to check. + */ + protected final void waitUntilVotingThreadStops(long timeout, int ... indexes) { + BaseElection[] elections = Arrays.stream(indexes) + .mapToObj(this::channel) + .filter(Objects::nonNull) + .map(this::election) + .toArray(BaseElection[]::new); + waitUntilVotingThreadStop(elections, timeout); + } + + @DataProvider(name = ALL_ELECTION_CLASSES_PROVIDER) + protected static Object[][] electionClasses() { + return BaseRaftElectionTest.electionClasses(); + } + } + + /** + * Base class for election tests based on a {@link MockRaftCluster}. + *+ * This class builds on the base {@link BaseRaftClusterTest} and offer utilities specific to the election process. + *
+ * + * @paramnull entries.
+ */
+ protected BaseElection[] elections() {
+ return Arrays.stream(nodes())
+ .filter(Objects::nonNull)
+ .map(RaftNode::election)
+ .toArray(BaseElection[]::new);
+
+ }
+
+ /**
+ * Wait until a leader is elected in the cluster.
+ * + * This methods blocks and waits until all the requested members in the cluster have an elected leader. + *
+ * + * @param timeout: Time in milliseconds to wait for an election. + * @param indexes: The index of the cluster members. + * @see #waitUntilLeaderElected(RAFT[], long) + */ + protected void waitUntilLeaderElected(long timeout, int ... indexes) { + RAFT[] rafts = Arrays.stream(indexes) + .mapToObj(this::node) + .filter(Objects::nonNull) + .map(RaftNode::raft) + .toArray(RAFT[]::new); + BaseRaftElectionTest.waitUntilLeaderElected(rafts, timeout); + } + + /** + * Wait until the voting thread stops running on all election instances. + * + * @param timeout: Time in milliseconds to wait. + * @param indexes: The index of the cluster members to check. + */ + protected void waitUntilVotingThreadStops(long timeout, int ... indexes) { + BaseElection[] elections = Arrays.stream(indexes) + .mapToObj(this::node) + .filter(Objects::nonNull) + .map(RaftNode::election) + .toArray(BaseElection[]::new); + waitUntilVotingThreadStop(elections, timeout); + } + + @DataProvider(name = ALL_ELECTION_CLASSES_PROVIDER) + protected static Object[][] electionClasses() { + return BaseRaftElectionTest.electionClasses(); + } + } +} diff --git a/tests/junit-functional/org/jgroups/tests/harness/BaseStateMachineTest.java b/tests/junit-functional/org/jgroups/tests/harness/BaseStateMachineTest.java new file mode 100644 index 00000000..2095ed2b --- /dev/null +++ b/tests/junit-functional/org/jgroups/tests/harness/BaseStateMachineTest.java @@ -0,0 +1,159 @@ +package org.jgroups.tests.harness; + +import org.jgroups.JChannel; +import org.jgroups.protocols.raft.RAFT; +import org.jgroups.raft.RaftHandle; +import org.jgroups.raft.StateMachine; + +import java.util.Arrays; +import java.util.Objects; +import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.jgroups.raft.testfwk.RaftTestUtils.eventually; + +/** + * Base class for tests that use a {@link StateMachine}. + * + *+ * This class provides base functionalities to assert on the {@link StateMachine}, and easy methods to retrieve the + * instances. + *
+ * + *+ * All the subclasses must override the {@link #createStateMachine(JChannel)} method. This is invoked everytime + * a new node is added to the cluster. A method to the super is necessary when overriding + * {@link #beforeClusterCreation()} or {@link #beforeChannelConnection(JChannel)}. These methods are responsible for + * populating the {@link RaftHandle}s and {@link StateMachine}s. + *
+ * + * @param+ * Warning: All subclasses must override this method. + *
+ * + * @param ch: The channel to retrieve any information to create the state machine instance. + * @return A {@link StateMachine} instance to use in the tests. + */ + protected T createStateMachine(JChannel ch) { + throw new IllegalStateException("Unknown state machine"); + } + + /** + * Retrieve the {@link StateMachine} in the given index. + * + * @param index: Index to retrieve the {@link RAFT} to retrieve the {@link StateMachine}. + * @return The {@link StateMachine} cast to type T. + * @throws AssertionError in case the {@link RAFT} instance for the index does not exist. + */ + protected final T stateMachine(int index) { + RAFT r = raft(index); + + assert r != null : "RAFT should not be null!"; + return stateMachine(r); + } + + /** + * Retrieves the {@link StateMachine} in use in the {@link RAFT} {@param r} instance. + * + * @param r: {@link RAFT} instance to retrieve the {@link StateMachine}. + * @return The {@link StateMachine} cast to type T. + */ + protected final T stateMachine(RAFT r) { + // noinspection unchecked + return (T) r.stateMachine(); + } + + protected final RaftHandle handle(int index) { + return handles[index]; + } + + /** + * Assert that all the {@link StateMachine} in the given {@param indexes} are matching. + * First, verify that all members have a matching commit index, meaning everything was applied. Then, verify + * the actual state machines. + * + * @param indexes: Indexes to verify. The {@link JChannel} for each index must exist. + */ + protected final void assertStateMachineEventuallyMatch(int ... indexes) { + // Verifies the commit indexes of all nodes. + BooleanSupplier commitIndexVerify = () -> Arrays.stream(indexes) + .mapToObj(this::raft) + .map(Objects::requireNonNull) + .map(RAFT::commitIndex) + .distinct() + .count() == 1; + + assertThat(eventually(commitIndexVerify, matchTimeout, TimeUnit.MILLISECONDS)) + .as(generateErrorMessage()) + .isTrue(); + System.out.println(dumpStateMachines(indexes)); + + // Verify the state machines match. + BooleanSupplier matchStateMachineVerify = () -> Arrays.stream(indexes) + .mapToObj(this::stateMachine) + .distinct() + .count() == 1; + assertThat(eventually(matchStateMachineVerify, matchTimeout, TimeUnit.MILLISECONDS)) + .as(generateErrorMessage() + " where " + dumpStateMachines(indexes)) + .isTrue(); + } + + private String generateErrorMessage() { + StringBuilder sb = new StringBuilder(); + RAFT leader = leader(); + + assertThat(leader) + .as("Expecting an elected leader") + .isNotNull(); + + sb.append(leader.raftId()) + .append(": commit-index=").append(leader.commitIndex()) + .append(leader.dumpCommitTable()); + return sb.toString(); + } + + private String dumpStateMachines(int ... indexes) { + StringBuilder sb = new StringBuilder(); + for (int i : indexes) { + sb.append(raft(i).raftId()) + .append(" -> ") + .append(stateMachine(i)) + .append("\n"); + } + return sb.toString(); + } +} diff --git a/tests/junit-functional/org/jgroups/tests/harness/RaftAssertion.java b/tests/junit-functional/org/jgroups/tests/harness/RaftAssertion.java new file mode 100644 index 00000000..e379a153 --- /dev/null +++ b/tests/junit-functional/org/jgroups/tests/harness/RaftAssertion.java @@ -0,0 +1,49 @@ +package org.jgroups.tests.harness; + +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeoutException; +import java.util.function.Supplier; + +import org.assertj.core.api.Assertions; + +import static org.assertj.core.api.Assertions.assertThat; + +public final class RaftAssertion { + + private RaftAssertion() { } + + public static void assertLeaderlessOperationThrows(ThrowingRunnable operation) { + assertLeaderlessOperationThrows(operation, "Running operation without a leader."); + } + + public static void assertLeaderlessOperationThrows(ThrowingRunnable operation, String message) { + assertLeaderlessOperationThrows(operation, () -> message); + } + + public static void assertLeaderlessOperationThrows(ThrowingRunnable operation, Supplier