Fix StoragePartition to return a furture for opening partition client + Fixes i…

…n AtomixLeaderElector

Change-Id: I6adf91e84cc17aec8acc895884dc8fbe75037978

@@ -50,6 +50,7 @@ public class StoragePartition extends DefaultPartition implements Managed<Storag
     private final MessagingService messagingService;
     private final MessagingService messagingService;
     private final ClusterService clusterService;
     private final ClusterService clusterService;
     private final File logFolder;
     private final File logFolder;
+    private CompletableFuture<StoragePartitionServer> serverOpenFuture;
     private static final Collection<ResourceType> RESOURCE_TYPES = ImmutableSet.of(
     private static final Collection<ResourceType> RESOURCE_TYPES = ImmutableSet.of(
                                                         new ResourceType(DistributedLong.class),
                                                         new ResourceType(DistributedLong.class),
                                                         new ResourceType(AtomixLeaderElector.class),
                                                         new ResourceType(AtomixLeaderElector.class),
@@ -90,9 +91,9 @@ public class StoragePartition extends DefaultPartition implements Managed<Storag
 
 
     @Override
     @Override
     public CompletableFuture<Void> open() {
     public CompletableFuture<Void> open() {
-        return openServer().thenAccept(s -> server = Optional.ofNullable(s))
-                           .thenCompose(v-> openClient())
-                           .thenAccept(v -> isOpened.set(true))
+        serverOpenFuture = openServer();
+        serverOpenFuture.thenAccept(s -> server = Optional.ofNullable(s));
+        return openClient().thenAccept(v -> isOpened.set(true))
                            .thenApply(v -> null);
                            .thenApply(v -> null);
     }
     }
 
 

@@ -44,7 +44,7 @@ public class AtomixLeaderElector extends Resource<AtomixLeaderElector>
     private final Set<Consumer<Change<Leadership>>> leadershipChangeListeners =
     private final Set<Consumer<Change<Leadership>>> leadershipChangeListeners =
             Sets.newConcurrentHashSet();
             Sets.newConcurrentHashSet();
 
 
-    public static final String CHANGE_SUBJECT = "changeEvents";
+    public static final String CHANGE_SUBJECT = "leadershipChangeEvents";
     private Listener<Change<Leadership>> listener;
     private Listener<Change<Leadership>> listener;
 
 
     public AtomixLeaderElector(CopycatClient client, Resource.Options options) {
     public AtomixLeaderElector(CopycatClient client, Resource.Options options) {

@@ -298,6 +298,11 @@ public final class AtomixLeaderElectorCommands {
         private String topic;
         private String topic;
         private NodeId nodeId;
         private NodeId nodeId;
 
 
+        ElectionChangeCommand() {
+            topic = null;
+            nodeId = null;
+        }
+
         public ElectionChangeCommand(String topic, NodeId nodeId) {
         public ElectionChangeCommand(String topic, NodeId nodeId) {
             this.topic = topic;
             this.topic = topic;
             this.nodeId = nodeId;
             this.nodeId = nodeId;
@@ -347,6 +352,10 @@ public final class AtomixLeaderElectorCommands {
      */
      */
     @SuppressWarnings("serial")
     @SuppressWarnings("serial")
     public static class Anoint extends ElectionChangeCommand<Boolean> {
     public static class Anoint extends ElectionChangeCommand<Boolean> {
+
+        private Anoint() {
+        }
+
         public Anoint(String topic, NodeId nodeId) {
         public Anoint(String topic, NodeId nodeId) {
             super(topic, nodeId);
             super(topic, nodeId);
         }
         }
@@ -357,6 +366,10 @@ public final class AtomixLeaderElectorCommands {
      */
      */
     @SuppressWarnings("serial")
     @SuppressWarnings("serial")
     public static class Promote extends ElectionChangeCommand<Boolean> {
     public static class Promote extends ElectionChangeCommand<Boolean> {
+
+        private Promote() {
+        }
+
         public Promote(String topic, NodeId nodeId) {
         public Promote(String topic, NodeId nodeId) {
             super(topic, nodeId);
             super(topic, nodeId);
         }
         }

@@ -202,9 +202,10 @@ public class AtomixLeaderElectorState extends ResourceStateMachine
     public boolean anoint(Commit<? extends Anoint> commit) {
     public boolean anoint(Commit<? extends Anoint> commit) {
         try {
         try {
             String topic = commit.operation().topic();
             String topic = commit.operation().topic();
+            NodeId nodeId = commit.operation().nodeId();
             Leadership oldLeadership = leadership(topic);
             Leadership oldLeadership = leadership(topic);
             ElectionState electionState = elections.computeIfPresent(topic,
             ElectionState electionState = elections.computeIfPresent(topic,
-                    (k, v) -> new ElectionState(v).transferLeadership(commit.operation().nodeId(), termCounter(topic)));
+                    (k, v) -> v.transferLeadership(nodeId, termCounter(topic)));
             Leadership newLeadership = leadership(topic);
             Leadership newLeadership = leadership(topic);
             if (!Objects.equal(oldLeadership, newLeadership)) {
             if (!Objects.equal(oldLeadership, newLeadership)) {
                 notifyLeadershipChange(oldLeadership, newLeadership);
                 notifyLeadershipChange(oldLeadership, newLeadership);
@@ -230,7 +231,7 @@ public class AtomixLeaderElectorState extends ResourceStateMachine
             if (oldLeadership == null || !oldLeadership.candidates().contains(nodeId)) {
             if (oldLeadership == null || !oldLeadership.candidates().contains(nodeId)) {
                 return false;
                 return false;
             }
             }
-            elections.computeIfPresent(topic, (k, v) -> new ElectionState(v).promote(commit.operation().nodeId()));
+            elections.computeIfPresent(topic, (k, v) -> v.promote(nodeId));
             Leadership newLeadership = leadership(topic);
             Leadership newLeadership = leadership(topic);
             if (!Objects.equal(oldLeadership, newLeadership)) {
             if (!Objects.equal(oldLeadership, newLeadership)) {
                 notifyLeadershipChange(oldLeadership, newLeadership);
                 notifyLeadershipChange(oldLeadership, newLeadership);
@@ -498,7 +499,7 @@ public class AtomixLeaderElectorState extends ResourceStateMachine
                                                   .filter(r -> r.nodeId().equals(nodeId))
                                                   .filter(r -> r.nodeId().equals(nodeId))
                                                   .findFirst()
                                                   .findFirst()
                                                   .orElse(null);
                                                   .orElse(null);
-            List<Registration> updatedRegistrations = Lists.newLinkedList();
+            List<Registration> updatedRegistrations = Lists.newArrayList();
             updatedRegistrations.add(registration);
             updatedRegistrations.add(registration);
             registrations.stream()
             registrations.stream()
                          .filter(r -> !r.nodeId().equals(nodeId))
                          .filter(r -> !r.nodeId().equals(nodeId))