Fixes some issues that were preventing intent tests from running

 - LeadershipStore to support serving getAllLeaderships from cache.
 - Removed a changed to KryoNamespaces that was causing serialization issues. Instead moved that type registration to McastStore.

Change-Id: I06acf1a397b6a982c0dfd0ebc0830b2161cf23a4

@@ -71,6 +71,8 @@ public class DistributedLeadershipStore
 
     protected NodeId localNodeId;
     protected ConsistentMap<String, InternalLeadership> leadershipMap;
+    protected Map<String, Versioned<InternalLeadership>> leadershipCache = Maps.newConcurrentMap();
+
     private final MapEventListener<String, InternalLeadership> leadershipChangeListener =
             event -> {
                 Leadership oldValue = InternalLeadership.toLeadership(Versioned.valueOrNull(event.oldValue()));
@@ -91,6 +93,12 @@ public class DistributedLeadershipStore
                 if (!leaderChanged && candidatesChanged) {
                     eventType = LeadershipEvent.Type.CANDIDATES_CHANGED;
                 }
+                leadershipCache.compute(event.key(), (k, v) -> {
+                    if (v == null || v.version() < event.newValue().version()) {
+                        return event.newValue();
+                    }
+                    return v;
+                });
                 notifyDelegate(new LeadershipEvent(eventType, newValue));
             };
 
@@ -103,6 +111,7 @@ public class DistributedLeadershipStore
                                       .withRelaxedReadConsistency()
                                       .withSerializer(Serializer.using(KryoNamespaces.API, InternalLeadership.class))
                                       .build();
+        leadershipMap.entrySet().forEach(e -> leadershipCache.put(e.getKey(), e.getValue()));
         leadershipMap.addListener(leadershipChangeListener);
         log.info("Started");
     }
@@ -210,16 +219,13 @@ public class DistributedLeadershipStore
 
     @Override
     public Leadership getLeadership(String topic) {
-        return InternalLeadership.toLeadership(Versioned.valueOrNull(leadershipMap.get(topic)));
+        InternalLeadership internalLeadership = Versioned.valueOrNull(leadershipCache.get(topic));
+        return internalLeadership == null ? null : internalLeadership.asLeadership();
     }
 
     @Override
     public Map<String, Leadership> getLeaderships() {
-        Map<String, Leadership> leaderships = Maps.newHashMap();
-        leadershipMap.entrySet().forEach(e -> {
-            leaderships.put(e.getKey(), e.getValue().value().asLeadership());
-        });
-        return ImmutableMap.copyOf(leaderships);
+        return ImmutableMap.copyOf(Maps.transformValues(leadershipCache, v -> v.value().asLeadership()));
     }
 
     private static class InternalLeadership {

@@ -225,7 +225,6 @@ import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
 
 public final class KryoNamespaces {
 
@@ -235,7 +234,6 @@ public final class KryoNamespaces {
             .register(AtomicBoolean.class)
             .register(AtomicInteger.class)
             .register(AtomicLong.class)
-            .register(AtomicReference.class)
             .register(new ImmutableListSerializer(),
                       ImmutableList.class,
                       ImmutableList.of(1).getClass(),

@@ -22,6 +22,7 @@ import org.slf4j.Logger;
 
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.slf4j.LoggerFactory.getLogger;
 
@@ -56,6 +57,7 @@ public class DistributedMcastStore extends AbstractStore<McastEvent, McastStoreD
                 .withSerializer(Serializer.using(KryoNamespace.newBuilder()
                                                          .register(KryoNamespaces.API)
                                                          .register(
+                                                                 AtomicReference.class,
                                                                  MulticastData.class,
                                                                  McastRoute.class,
                                                                  McastRoute.Type.class

@@ -57,6 +57,7 @@ public final class MulticastData {
     }
 
     public void setSource(ConnectPoint source) {
+        // FIXME: violates immutability
         isEmpty.set(false);
         this.source.set(source);
     }