Group test fixes (#208)

* Rewrite group recovery test to ensure final membership checks account for sequential consistency in membership change events.

* Fix group test log level.

* Fix group recovery test name.

* Ensure leave events are triggered for local group members.

* Ensure local membership is consistent prior to new join events being received when recovering group members.

* Reset group membership test log levels.

groups/src/main/java/io/atomix/group/internal/MembershipGroup.java

@@ -31,10 +31,7 @@
 import io.atomix.resource.AbstractResource;
 import io.atomix.resource.ResourceType;
 
-import java.util.Collection;
-import java.util.Map;
-import java.util.Properties;
-import java.util.UUID;
+import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Consumer;
@@ -60,7 +57,7 @@ public class MembershipGroup extends AbstractResource<DistributedGroup> implemen
   private final Map<String, AbstractGroupMember> members = new ConcurrentHashMap<>();
   private final MessageProducerService producerService;
   private final MessageConsumerService consumerService;
-  private final Map<String, GroupCommands.Join> joinCommands = new ConcurrentHashMap<>();
+  private final Map<String, GroupCommands.Join> localJoins = new ConcurrentHashMap<>();
 
   public MembershipGroup(CopycatClient client, Properties options) {
     super(client, new ResourceType(DistributedGroup.class), options);
@@ -135,7 +132,7 @@ private CompletableFuture<LocalMember> join(String memberId, boolean persistent,
       AbstractGroupMember member = members.get(info.memberId());
       if (member == null || !(member instanceof LocalGroupMember)) {
         member = new LocalGroupMember(info, this, producerService, consumerService);
-        joinCommands.put(memberId, cmd);
+        localJoins.put(memberId, cmd);
         members.put(info.memberId(), member);
       }
       return (LocalGroupMember) member;
@@ -150,8 +147,11 @@ public Listener<GroupMember> onJoin(Consumer<GroupMember> listener) {
   @Override
   public CompletableFuture<Void> remove(String memberId) {
     return client.submit(new GroupCommands.Leave(memberId)).thenRun(() -> {
-      joinCommands.remove(memberId);
-      members.remove(memberId);
+      localJoins.remove(memberId);
+      GroupMember member = members.remove(memberId);
+      if (member != null) {
+        leaveListeners.accept(member);
+      }
     });
   }
 
@@ -176,31 +176,40 @@ public CompletableFuture<DistributedGroup> open() {
 
   @Override
   protected CompletableFuture<Void> recover(Integer attempt) {
-    final int persistentCount =
-        (int) joinCommands.values().stream().filter(v -> !v.persist()).count();
-    final CompletableFuture[] closeFutures = new CompletableFuture[persistentCount];
-    final CompletableFuture[] joinFutures = new CompletableFuture[joinCommands.size()];
-    int i = 0;
-    for (GroupCommands.Join cmd : joinCommands.values()) {
-      if (cmd.persist()) {
-        joinFutures[i] = join(cmd.member(), cmd.persist(), cmd.metadata());
-      } else {
-        closeFutures[i] = remove(cmd.member());
-        joinFutures[i] = join(cmd.metadata());
-      }
-      i++;
-    }
-    members.clear();
+    // When recovering the membership group, we need to ensure that all local non-persistent members are
+    // removed from the group prior to fetching the group membership from the cluster again, and prior to
+    // adding recovered members that the membership list is updated to ensure the list is consistent
+    // when join event handlers are called.
+    Map<String, GroupCommands.Join> joins = new HashMap<>(localJoins);
     return sync()
-      .thenCompose(v -> CompletableFuture.allOf(closeFutures))
-      .thenCompose(v -> CompletableFuture.allOf(joinFutures));
+      .thenCompose(v -> {
+        List<CompletableFuture> futures = new ArrayList<>(joins.size());
+        for (GroupCommands.Join join : joins.values()) {
+          if (!join.persist()) {
+            futures.add(remove(join.member()));
+          }
+        }
+        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
+      })
+      .thenCompose(v -> sync())
+      .thenCompose(v -> {
+        List<CompletableFuture> futures = new ArrayList<>(joins.size());
+        for (GroupCommands.Join join : joins.values()) {
+          if (join.persist()) {
+            futures.add(join(join.member(), join.persist(), join.metadata()));
+          } else {
+            futures.add(join(join.metadata()));
+          }
+        }
+        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
+      });
   }
 
   /**
    * Synchronizes the membership group.
    */
   private CompletableFuture<Void> sync() {
-    return client.submit(new GroupCommands.Listen()).thenAccept(status-> {
+    return client.submit(new GroupCommands.Listen()).thenAccept(status -> {
       for (GroupMemberInfo info : status.members()) {
         AbstractGroupMember member = this.members.get(info.memberId());
         if (member == null) {

groups/src/test/java/io/atomix/group/DistributedGroupTest.java

@@ -23,7 +23,6 @@
 import io.atomix.testing.AbstractCopycatTest;
 import org.testng.annotations.Test;
 
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import static org.testng.Assert.assertEquals;
@@ -599,56 +598,129 @@ public void testGroupMessageFailOnLeave() throws Throwable {
     await(10000, 2);
   }
 
+  /**
+   * Tests that a member already exists when a join event is received.
+   */
+  public void testMemberExistsOnJoinEvent() throws Throwable {
+    createServers(3);
+
+    final CopycatClient client1 = createCopycatClient();
+    final CopycatClient client2 = createCopycatClient();
+
+    final DistributedGroup group1 = createResource(client1, new DistributedGroup.Options());
+    final DistributedGroup group2 = createResource(client2, new DistributedGroup.Options());
+
+    group1.onJoin(m -> {
+      threadAssertEquals(1, group1.members().size());
+      resume();
+    });
+    group2.onJoin(m -> {
+      threadAssertEquals(1, group2.members().size());
+      resume();
+    });
+
+    group1.join().thenRun(this::resume);
+
+    await(5000, 3);
+  }
+
+  /**
+   * Tests that the local onLeave handler is called when a member leaves the group.
+   */
+  public void testLocalOnLeave() throws Throwable {
+    createServers(3);
+
+    final CopycatClient client1 = createCopycatClient();
+    final CopycatClient client2 = createCopycatClient();
+
+    final DistributedGroup group1 = createResource(client1, new DistributedGroup.Options());
+    final DistributedGroup group2 = createResource(client2, new DistributedGroup.Options());
+
+    // Group join handlers
+    group1.onJoin(m -> resume());
+    group2.onJoin(m -> resume());
+
+    // Join both members to the group
+    final LocalMember member1 = group1.join().get(5, TimeUnit.SECONDS);
+    final LocalMember member2 = group2.join().get(5, TimeUnit.SECONDS);
+
+    await(5000, 4);
+
+    group1.onLeave(m -> {
+      resume();
+    });
+    group2.onLeave(m -> {
+      resume();
+    });
+
+    member1.leave().thenRun(this::resume);
+
+    await(5000, 3);
+  }
+
+  /**
+   * Tests the recovery of a group resource/member in a group.
+   */
   public void testRecovery() throws Throwable {
     createServers(3);
+
     final CopycatClient client1 = createCopycatClient();
     final CopycatClient client2 = createCopycatClient();
+
     final DistributedGroup group1 = createResource(client1, new DistributedGroup.Options());
     final DistributedGroup group2 = createResource(client2, new DistributedGroup.Options());
 
+    // Group1 on join handler
     group1.onJoin(m -> {
+      System.out.println(group1.members().size());
       if (group1.members().size() == 2) {
         resume();
       }
     });
+
+    // Group2 on join handler
     group2.onJoin(m -> {
+      System.out.println(group2.members().size());
       if (group2.members().size() == 2) {
         resume();
       }
     });
 
-    final CountDownLatch leave = new CountDownLatch(1);
-    group1.onLeave(m -> leave.countDown());
+    // Join both members to the group
+    final LocalMember member1 = group1.join().get(5, TimeUnit.SECONDS);
+    final LocalMember member2 = group2.join().get(5, TimeUnit.SECONDS);
 
-    group1.join().get(10, TimeUnit.SECONDS);
-    final LocalMember member2 = group2.join().get(10, TimeUnit.SECONDS);
+    // Wait for both group instances to receive join events for both members
     await(5000, 2);
 
+    // Set a recovery handler for group1
     group1.onRecovery(attempt -> {
       threadAssertEquals(1, attempt);
       resume();
     });
 
-    final long startExpire = System.currentTimeMillis();
+    // Expire client 1's session, which should cause group1 to expire
     ((ClientSession) client1.session()).expire().whenComplete((v, e) -> {
       threadAssertNull(e);
       resume();
     });
 
-    await(5000, 2);
+    // Wait for the client's session to expire and be recovered
+    await(5000, 4);
 
-    // wait for the old session to expire in the state machine
-    final long remainingExpire = 5000 - (System.currentTimeMillis() - startExpire);
-    Thread.sleep(Math.max(0, remainingExpire));
+    // Ensure one member remains once a node is removed
+    group1.onLeave(m -> {
+      threadAssertEquals(1, group1.members().size());
+      resume();
+    });
+    group2.onLeave(m -> {
+      threadAssertEquals(1, group2.members().size());
+      resume();
+    });
 
+    // Remove member2 from the group
     member2.leave().thenRun(this::resume);
-    await(5000, 1);
-
-    // recovered client should receive events via new session
-    threadAssertTrue(leave.await(5, TimeUnit.SECONDS));
-    threadAssertEquals(1, group1.members().size());
-
-    // recovered client should rejoin the cluster
-    threadAssertEquals(1, group2.members().size());
+    await(5000, 3);
   }
+
 }

groups/src/test/resources/logback.xml

@@ -24,4 +24,4 @@
   <root level="INFO">
     <appender-ref ref="STDOUT" />
   </root>
-</configuration>
+</configuration>