Update Atomix server tests for new cmembership changes.

manager/src/test/java/io/atomix/AbstractServerTest.java

@@ -18,36 +18,38 @@
 import io.atomix.catalyst.transport.Address;
 import io.atomix.catalyst.transport.LocalServerRegistry;
 import io.atomix.catalyst.transport.LocalTransport;
+import io.atomix.copycat.server.CopycatServer;
+import io.atomix.copycat.server.state.Member;
 import io.atomix.copycat.server.storage.Storage;
 import io.atomix.copycat.server.storage.StorageLevel;
 import net.jodah.concurrentunit.ConcurrentTestCase;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 
-import java.io.IOException;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.stream.Collectors;
 
 /**
  * Abstract server test.
  *
  * @author <a href="http://github.com/kuujo>Jordan Halterman</a>
  */
 public abstract class AbstractServerTest extends ConcurrentTestCase {
-  protected LocalServerRegistry registry;
-  protected int port;
-  protected List<Address> members;
+  protected volatile LocalServerRegistry registry;
+  protected volatile int port;
+  protected volatile List<Member> members;
+  protected volatile List<Atomix> clients = new ArrayList<>();
+  protected volatile List<AtomixServer> servers = new ArrayList<>();
 
   /**
    * Returns the next server address.
    *
    * @return The next server address.
    */
-  protected Address nextAddress() {
-    Address address = new Address("localhost", port++);
-    members.add(address);
-    return address;
+  protected Member nextMember() {
+    return new Member(CopycatServer.Type.INACTIVE, new Address("localhost", ++port), new Address("localhost", port + 1000));
   }
 
   /**
@@ -56,43 +58,61 @@ protected Address nextAddress() {
   protected List<AtomixServer> createServers(int nodes) throws Throwable {
     List<AtomixServer> servers = new ArrayList<>();
 
-    List<Address> members = new ArrayList<>();
-    for (int i = 1; i <= nodes; i++) {
-      members.add(nextAddress());
+    for (int i = 0; i < nodes; i++) {
+      members.add(nextMember());
     }
 
     for (int i = 0; i < nodes; i++) {
-      AtomixServer server = createServer(members.get(i));
+      AtomixServer server = createServer(members, members.get(i));
       server.open().thenRun(this::resume);
       servers.add(server);
     }
 
-    await(0, nodes);
+    await(10000, nodes);
 
     return servers;
   }
 
   /**
    * Creates an Atomix server.
    */
-  protected AtomixServer createServer(Address address) {
-    return AtomixServer.builder(address, members)
+  protected AtomixServer createServer(List<Member> members, Member member) {
+    AtomixServer server = AtomixServer.builder(member.clientAddress(), member.serverAddress(), members.stream().map(Member::serverAddress).collect(Collectors.toList()))
       .withTransport(new LocalTransport(registry))
       .withStorage(Storage.builder()
         .withStorageLevel(StorageLevel.MEMORY)
+        .withMaxSegmentSize(1024 * 1024)
         .withMaxEntriesPerSegment(8)
-        .withMinorCompactionInterval(Duration.ofSeconds(5))
-        .withMajorCompactionInterval(Duration.ofSeconds(10))
+        .withMinorCompactionInterval(Duration.ofSeconds(3))
+        .withMajorCompactionInterval(Duration.ofSeconds(7))
         .build())
       .build();
+    servers.add(server);
+    return server;
   }
 
   @BeforeMethod
   @AfterMethod
-  public void clearTests() throws IOException {
-    port = 5000;
+  public void clearTests() throws Exception {
+    clients.forEach(c -> {
+      try {
+        c.close().join();
+      } catch (Exception e) {
+      }
+    });
+
+    servers.forEach(s -> {
+      try {
+        s.close().join();
+      } catch (Exception e) {
+      }
+    });
+
     registry = new LocalServerRegistry();
     members = new ArrayList<>();
+    port = 5000;
+    clients = new ArrayList<>();
+    servers = new ArrayList<>();
   }
 
 }

manager/src/test/java/io/atomix/AtomixClientServerTest.java

@@ -20,10 +20,12 @@
 import io.atomix.copycat.client.Query;
 import io.atomix.copycat.client.CopycatClient;
 import io.atomix.copycat.server.Commit;
+import io.atomix.copycat.server.state.Member;
 import io.atomix.resource.*;
 import org.testng.annotations.Test;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
 
 /**
  * Client server test.
@@ -232,7 +234,9 @@ public void testGetResourceKeys() throws Throwable {
    * Creates a client.
    */
   private Atomix createClient() throws Throwable {
-    Atomix client = AtomixClient.builder(members).withTransport(new LocalTransport(registry)).build();
+    Atomix client = AtomixClient.builder(members.stream().map(Member::clientAddress).collect(Collectors.toList()))
+      .withTransport(new LocalTransport(registry))
+      .build();
     client.open().thenRun(this::resume);
     await(10000);
     return client;

manager/src/test/java/io/atomix/AtomixServerTest.java

@@ -32,7 +32,7 @@ public class AtomixServerTest extends AbstractServerTest {
    */
   public void testServerJoin() throws Throwable {
     createServers(3);
-    AtomixServer joiner = createServer(nextAddress());
+    AtomixServer joiner = createServer(members, nextMember());
     joiner.open().thenRun(this::resume);
     await();
   }