diff --git a/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/ClusterManager.java b/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/ClusterManager.java
deleted file mode 100644
index f8202fc..0000000
--- a/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/ClusterManager.java
+++ /dev/null
@@ -1,346 +0,0 @@
-/*
- * Copyright 2015 Open Networking Laboratory
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.onosproject.store.cluster.impl;
-
-import static org.onlab.util.Tools.groupedThreads;
-import static org.slf4j.LoggerFactory.getLogger;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.NetworkInterface;
-import java.net.SocketException;
-import java.util.Enumeration;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-
-import org.apache.felix.scr.annotations.Activate;
-import org.apache.felix.scr.annotations.Component;
-import org.apache.felix.scr.annotations.Deactivate;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.ReferenceCardinality;
-import org.apache.felix.scr.annotations.Service;
-import org.onlab.netty.Endpoint;
-import org.onlab.netty.Message;
-import org.onlab.netty.MessageHandler;
-import org.onlab.netty.NettyMessagingService;
-import org.onlab.packet.IpAddress;
-import org.onlab.util.KryoNamespace;
-import org.onosproject.cluster.ClusterAdminService;
-import org.onosproject.cluster.ClusterEvent;
-import org.onosproject.cluster.ClusterEventListener;
-import org.onosproject.cluster.ClusterService;
-import org.onosproject.cluster.ControllerNode;
-import org.onosproject.cluster.ControllerNode.State;
-import org.onosproject.cluster.DefaultControllerNode;
-import org.onosproject.cluster.NodeId;
-import org.onosproject.event.AbstractListenerRegistry;
-import org.onosproject.event.EventDeliveryService;
-import org.onosproject.store.serializers.KryoNamespaces;
-import org.onosproject.store.serializers.KryoSerializer;
-import org.slf4j.Logger;
-
-import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Maps;
-import com.hazelcast.util.AddressUtil;
-
-import static com.google.common.base.Preconditions.checkNotNull;
-import static com.google.common.base.Preconditions.checkArgument;
-
-/**
- * ClusterService implementation that employs an accrual failure
- * detector to identify cluster member up/down status.
- */
-@Component(immediate = true, enabled = false)
-@Service
-public class ClusterManager implements ClusterService, ClusterAdminService {
-
-    private final Logger log = getLogger(getClass());
-
-    protected final AbstractListenerRegistry<ClusterEvent, ClusterEventListener>
-        listenerRegistry = new AbstractListenerRegistry<>();
-
-    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
-    protected EventDeliveryService eventDispatcher;
-
-    // TODO: make these configurable.
-    private static final int HEARTBEAT_FD_PORT = 2419;
-    private static final int HEARTBEAT_INTERVAL_MS = 100;
-    private static final int PHI_FAILURE_THRESHOLD = 10;
-
-    private static final String CONFIG_DIR = "../config";
-    private static final String CLUSTER_DEFINITION_FILE = "cluster.json";
-
-    private ClusterDefinition clusterDefinition;
-
-    private Set<ControllerNode> seedNodes;
-    private final Map<NodeId, ControllerNode> allNodes = Maps.newConcurrentMap();
-    private final Map<NodeId, State> nodeStates = Maps.newConcurrentMap();
-    private NettyMessagingService messagingService = new NettyMessagingService();
-    private ScheduledExecutorService heartBeatSender = Executors.newSingleThreadScheduledExecutor(
-            groupedThreads("onos/cluster/membership", "heartbeat-sender"));
-    private ExecutorService heartBeatMessageHandler = Executors.newSingleThreadExecutor(
-            groupedThreads("onos/cluster/membership", "heartbeat-receiver"));
-
-    private static final String HEARTBEAT_MESSAGE = "onos-cluster-heartbeat";
-
-
-    private PhiAccrualFailureDetector failureDetector;
-
-    private ControllerNode localNode;
-
-    private static final KryoSerializer SERIALIZER = new KryoSerializer() {
-        @Override
-        protected void setupKryoPool() {
-            serializerPool = KryoNamespace.newBuilder()
-                .register(KryoNamespaces.API)
-                .register(HeartbeatMessage.class)
-                .build()
-                .populate(1);
-        }
-    };
-
-    private static final String INSTANCE_ID_NULL = "Instance ID cannot be null";
-
-    @Activate
-    public void activate() {
-
-        File clusterDefinitionFile = new File(CONFIG_DIR, CLUSTER_DEFINITION_FILE);
-
-        try {
-            clusterDefinition = new ClusterDefinitionStore(clusterDefinitionFile.getPath()).read();
-            seedNodes = ImmutableSet.copyOf(clusterDefinition.getNodes())
-                            .stream()
-                            .map(nodeInfo -> new DefaultControllerNode(
-                                        new NodeId(nodeInfo.getId()),
-                                        IpAddress.valueOf(nodeInfo.getIp()),
-                                        nodeInfo.getTcpPort()))
-                            .collect(Collectors.toSet());
-        } catch (IOException e) {
-            throw new IllegalStateException("Failed to read cluster definition.", e);
-        }
-
-        seedNodes.forEach(node -> {
-            allNodes.put(node.id(), node);
-            nodeStates.put(node.id(), State.INACTIVE);
-        });
-
-        establishSelfIdentity();
-
-        messagingService = new NettyMessagingService(HEARTBEAT_FD_PORT);
-
-        try {
-            messagingService.activate();
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            throw new IllegalStateException("Failed to cleanly initialize membership and"
-                    + " failure detector communication channel.", e);
-        }
-        messagingService.registerHandler(
-                HEARTBEAT_MESSAGE,
-                new HeartbeatMessageHandler(),
-                heartBeatMessageHandler);
-
-        eventDispatcher.addSink(ClusterEvent.class, listenerRegistry);
-        failureDetector = new PhiAccrualFailureDetector();
-
-        heartBeatSender.scheduleWithFixedDelay(
-                this::heartbeat,
-                0,
-                HEARTBEAT_INTERVAL_MS,
-                TimeUnit.MILLISECONDS);
-
-        log.info("Started");
-    }
-
-    @Deactivate
-    public void deactivate() {
-        try {
-            messagingService.deactivate();
-        } catch (Exception e) {
-            log.trace("Failed to cleanly shutdown cluster membership messaging", e);
-        }
-
-        heartBeatSender.shutdownNow();
-        heartBeatMessageHandler.shutdownNow();
-        eventDispatcher.removeSink(ClusterEvent.class);
-
-        log.info("Stopped");
-    }
-
-    @Override
-    public ControllerNode getLocalNode() {
-        return localNode;
-    }
-
-    @Override
-    public Set<ControllerNode> getNodes() {
-        return ImmutableSet.copyOf(allNodes.values());
-    }
-
-    @Override
-    public ControllerNode getNode(NodeId nodeId) {
-        checkNotNull(nodeId, INSTANCE_ID_NULL);
-        return allNodes.get(nodeId);
-    }
-
-    @Override
-    public State getState(NodeId nodeId) {
-        checkNotNull(nodeId, INSTANCE_ID_NULL);
-        return nodeStates.get(nodeId);
-    }
-
-    @Override
-    public void addListener(ClusterEventListener listener) {
-        checkNotNull(listener, "Listener must not be null");
-        listenerRegistry.addListener(listener);
-    }
-
-    @Override
-    public void removeListener(ClusterEventListener listener) {
-        checkNotNull(listener, "Listener must not be null");
-        listenerRegistry.removeListener(listener);
-    }
-
-    @Override
-    public ControllerNode addNode(NodeId nodeId, IpAddress ip, int tcpPort) {
-        checkNotNull(nodeId, INSTANCE_ID_NULL);
-        checkNotNull(ip, "IP address must not be null");
-        checkArgument(tcpPort > 5000, "Tcp port must be greater than 5000");
-        ControllerNode node = new DefaultControllerNode(nodeId, ip, tcpPort);
-        allNodes.put(node.id(), node);
-        nodeStates.put(nodeId, State.INACTIVE);
-        eventDispatcher.post(new ClusterEvent(ClusterEvent.Type.INSTANCE_ADDED, node));
-        return node;
-    }
-
-    @Override
-    public void removeNode(NodeId nodeId) {
-        checkNotNull(nodeId, INSTANCE_ID_NULL);
-        ControllerNode node = allNodes.remove(nodeId);
-        if (node != null) {
-            nodeStates.remove(nodeId);
-            eventDispatcher.post(new ClusterEvent(ClusterEvent.Type.INSTANCE_REMOVED, node));
-        }
-    }
-
-    private void establishSelfIdentity() {
-        try {
-            IpAddress ip = findLocalIp();
-            localNode = new DefaultControllerNode(new NodeId(ip.toString()), ip);
-            allNodes.put(localNode.id(), localNode);
-            nodeStates.put(localNode.id(), State.ACTIVE);
-            log.info("Local Node: {}", localNode);
-        } catch (SocketException e) {
-            throw new IllegalStateException("Cannot determine local IP", e);
-        }
-    }
-
-    private void heartbeat() {
-        try {
-            Set<ControllerNode> peers = allNodes.values()
-                    .stream()
-                    .filter(node -> !(node.id().equals(localNode.id())))
-                    .collect(Collectors.toSet());
-            byte[] hbMessagePayload = SERIALIZER.encode(new HeartbeatMessage(localNode, peers));
-            peers.forEach((node) -> {
-                heartbeatToPeer(hbMessagePayload, node);
-                State currentState = nodeStates.get(node.id());
-                double phi = failureDetector.phi(node.id());
-                if (phi >= PHI_FAILURE_THRESHOLD) {
-                    if (currentState == State.ACTIVE) {
-                        nodeStates.put(node.id(), State.INACTIVE);
-                        notifyStateChange(node.id(), State.ACTIVE, State.INACTIVE);
-                    }
-                } else {
-                    if (currentState == State.INACTIVE) {
-                        nodeStates.put(node.id(), State.ACTIVE);
-                        notifyStateChange(node.id(), State.INACTIVE, State.ACTIVE);
-                    }
-                }
-            });
-        } catch (Exception e) {
-            log.debug("Failed to send heartbeat", e);
-        }
-    }
-
-    private void notifyStateChange(NodeId nodeId, State oldState, State newState) {
-        if (newState == State.ACTIVE) {
-            eventDispatcher.post(new ClusterEvent(ClusterEvent.Type.INSTANCE_ACTIVATED, allNodes.get(nodeId)));
-        } else {
-            eventDispatcher.post(new ClusterEvent(ClusterEvent.Type.INSTANCE_DEACTIVATED, allNodes.get(nodeId)));
-        }
-    }
-
-    private void heartbeatToPeer(byte[] messagePayload, ControllerNode peer) {
-        Endpoint remoteEp = new Endpoint(peer.ip(), HEARTBEAT_FD_PORT);
-        try {
-            messagingService.sendAsync(remoteEp, HEARTBEAT_MESSAGE, messagePayload);
-        } catch (IOException e) {
-            log.debug("Sending heartbeat to {} failed", remoteEp, e);
-        }
-    }
-
-    private class HeartbeatMessageHandler implements MessageHandler {
-        @Override
-        public void handle(Message message) throws IOException {
-            HeartbeatMessage hb = SERIALIZER.decode(message.payload());
-            failureDetector.report(hb.source().id());
-            hb.knownPeers().forEach(node -> {
-                allNodes.put(node.id(), node);
-            });
-        }
-    }
-
-    private static class HeartbeatMessage {
-        private ControllerNode source;
-        private Set<ControllerNode> knownPeers;
-
-        public HeartbeatMessage(ControllerNode source, Set<ControllerNode> members) {
-            this.source = source;
-            this.knownPeers = ImmutableSet.copyOf(members);
-        }
-
-        public ControllerNode source() {
-            return source;
-        }
-
-        public Set<ControllerNode> knownPeers() {
-            return knownPeers;
-        }
-    }
-
-    private IpAddress findLocalIp() throws SocketException {
-        Enumeration<NetworkInterface> interfaces =
-                NetworkInterface.getNetworkInterfaces();
-        while (interfaces.hasMoreElements()) {
-            NetworkInterface iface = interfaces.nextElement();
-            Enumeration<InetAddress> inetAddresses =  iface.getInetAddresses();
-            while (inetAddresses.hasMoreElements()) {
-                IpAddress ip = IpAddress.valueOf(inetAddresses.nextElement());
-                if (AddressUtil.matchInterface(ip.toString(), clusterDefinition.getIpPrefix())) {
-                    return ip;
-                }
-            }
-        }
-        throw new IllegalStateException("Unable to determine local ip");
-    }
-}
diff --git a/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/DistributedClusterStore.java b/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/DistributedClusterStore.java
index 4da1b46..213d0f4 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/DistributedClusterStore.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/DistributedClusterStore.java
@@ -15,167 +15,323 @@
  */
 package org.onosproject.store.cluster.impl;
 
-import com.google.common.base.Optional;
-import com.google.common.cache.LoadingCache;
-import com.google.common.collect.ImmutableSet;
-import com.hazelcast.core.IMap;
-import com.hazelcast.core.Member;
-import com.hazelcast.core.MemberAttributeEvent;
-import com.hazelcast.core.MembershipEvent;
-import com.hazelcast.core.MembershipListener;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.onlab.util.Tools.groupedThreads;
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.net.SocketException;
+import java.util.Enumeration;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
 import org.apache.felix.scr.annotations.Service;
+import org.onlab.netty.Endpoint;
+import org.onlab.netty.Message;
+import org.onlab.netty.MessageHandler;
+import org.onlab.netty.NettyMessagingService;
+import org.onlab.packet.IpAddress;
+import org.onlab.util.KryoNamespace;
 import org.onosproject.cluster.ClusterEvent;
 import org.onosproject.cluster.ClusterStore;
 import org.onosproject.cluster.ClusterStoreDelegate;
 import org.onosproject.cluster.ControllerNode;
 import org.onosproject.cluster.DefaultControllerNode;
+import org.onosproject.cluster.ControllerNode.State;
 import org.onosproject.cluster.NodeId;
-import org.onosproject.store.hz.AbsentInvalidatingLoadingCache;
-import org.onosproject.store.hz.AbstractHazelcastStore;
-import org.onosproject.store.hz.OptionalCacheLoader;
-import org.onlab.packet.IpAddress;
+import org.onosproject.store.AbstractStore;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.serializers.KryoSerializer;
+import org.slf4j.Logger;
 
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Maps;
+import com.hazelcast.util.AddressUtil;
 
-import static com.google.common.cache.CacheBuilder.newBuilder;
-import static org.onosproject.cluster.ClusterEvent.Type.INSTANCE_ACTIVATED;
-import static org.onosproject.cluster.ClusterEvent.Type.INSTANCE_DEACTIVATED;
-import static org.onosproject.cluster.ControllerNode.State;
-
-/**
- * Distributed implementation of the cluster nodes store.
- */
 @Component(immediate = true)
 @Service
+/**
+ * Distributed cluster nodes store that employs an accrual failure
+ * detector to identify cluster member up/down status.
+ */
 public class DistributedClusterStore
-        extends AbstractHazelcastStore<ClusterEvent, ClusterStoreDelegate>
+        extends AbstractStore<ClusterEvent, ClusterStoreDelegate>
         implements ClusterStore {
 
-    private IMap<byte[], byte[]> rawNodes;
-    private LoadingCache<NodeId, Optional<DefaultControllerNode>> nodes;
+    private final Logger log = getLogger(DistributedClusterStore.class);
 
-    private String listenerId;
-    private final MembershipListener listener = new InternalMembershipListener();
-    private final Map<NodeId, State> states = new ConcurrentHashMap<>();
+    // TODO: make these configurable.
+    private static final int HEARTBEAT_FD_PORT = 2419;
+    private static final int HEARTBEAT_INTERVAL_MS = 100;
+    private static final int PHI_FAILURE_THRESHOLD = 10;
 
-    private String nodesListenerId;
+    private static final String CONFIG_DIR = "../config";
+    private static final String CLUSTER_DEFINITION_FILE = "cluster.json";
+    private static final String HEARTBEAT_MESSAGE = "onos-cluster-heartbeat";
 
-    @Override
+    private static final KryoSerializer SERIALIZER = new KryoSerializer() {
+        @Override
+        protected void setupKryoPool() {
+            serializerPool = KryoNamespace.newBuilder()
+                .register(KryoNamespaces.API)
+                .register(HeartbeatMessage.class)
+                .build()
+                .populate(1);
+        }
+    };
+
+    private static final String INSTANCE_ID_NULL = "Instance ID cannot be null";
+
+    private ClusterDefinition clusterDefinition;
+
+    private Set<ControllerNode> seedNodes;
+    private final Map<NodeId, ControllerNode> allNodes = Maps.newConcurrentMap();
+    private final Map<NodeId, State> nodeStates = Maps.newConcurrentMap();
+    private NettyMessagingService messagingService = new NettyMessagingService();
+    private ScheduledExecutorService heartBeatSender = Executors.newSingleThreadScheduledExecutor(
+            groupedThreads("onos/cluster/membership", "heartbeat-sender"));
+    private ExecutorService heartBeatMessageHandler = Executors.newSingleThreadExecutor(
+            groupedThreads("onos/cluster/membership", "heartbeat-receiver"));
+
+    private PhiAccrualFailureDetector failureDetector;
+
+    private ControllerNode localNode;
+
     @Activate
     public void activate() {
-        super.activate();
-        listenerId = theInstance.getCluster().addMembershipListener(listener);
+        File clusterDefinitionFile = new File(CONFIG_DIR,
+                CLUSTER_DEFINITION_FILE);
 
-        rawNodes = theInstance.getMap("nodes");
-        OptionalCacheLoader<NodeId, DefaultControllerNode> nodeLoader
-                = new OptionalCacheLoader<>(serializer, rawNodes);
-        nodes = new AbsentInvalidatingLoadingCache<>(newBuilder().build(nodeLoader));
-        nodesListenerId = rawNodes.addEntryListener(new RemoteCacheEventHandler<>(nodes), true);
+        try {
+            clusterDefinition = new ClusterDefinitionStore(
+                    clusterDefinitionFile.getPath()).read();
+            seedNodes = ImmutableSet
+                    .copyOf(clusterDefinition.getNodes())
+                    .stream()
+                    .map(nodeInfo -> new DefaultControllerNode(new NodeId(
+                            nodeInfo.getId()), IpAddress.valueOf(nodeInfo
+                            .getIp()), nodeInfo.getTcpPort()))
+                    .collect(Collectors.toSet());
+        } catch (IOException e) {
+            throw new IllegalStateException(
+                    "Failed to read cluster definition.", e);
+        }
 
-        loadClusterNodes();
+        seedNodes.forEach(node -> {
+            allNodes.put(node.id(), node);
+            nodeStates.put(node.id(), State.INACTIVE);
+        });
+
+        establishSelfIdentity();
+
+        messagingService = new NettyMessagingService(HEARTBEAT_FD_PORT);
+
+        try {
+            messagingService.activate();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IllegalStateException(
+                    "Failed to cleanly initialize membership and"
+                            + " failure detector communication channel.", e);
+        }
+        messagingService.registerHandler(HEARTBEAT_MESSAGE,
+                new HeartbeatMessageHandler(), heartBeatMessageHandler);
+
+        failureDetector = new PhiAccrualFailureDetector();
+
+        heartBeatSender.scheduleWithFixedDelay(this::heartbeat, 0,
+                HEARTBEAT_INTERVAL_MS, TimeUnit.MILLISECONDS);
 
         log.info("Started");
     }
 
-    // Loads the initial set of cluster nodes
-    private void loadClusterNodes() {
-        for (Member member : theInstance.getCluster().getMembers()) {
-            addNode(node(member));
-        }
-    }
-
     @Deactivate
     public void deactivate() {
-        rawNodes.removeEntryListener(nodesListenerId);
-        theInstance.getCluster().removeMembershipListener(listenerId);
+        try {
+            messagingService.deactivate();
+        } catch (Exception e) {
+            log.trace("Failed to cleanly shutdown cluster membership messaging", e);
+        }
+
+        heartBeatSender.shutdownNow();
+        heartBeatMessageHandler.shutdownNow();
+
         log.info("Stopped");
     }
 
     @Override
+    public void setDelegate(ClusterStoreDelegate delegate) {
+        checkNotNull(delegate, "Delegate cannot be null");
+        this.delegate = delegate;
+    }
+
+    @Override
+    public void unsetDelegate(ClusterStoreDelegate delegate) {
+        this.delegate = null;
+    }
+
+    @Override
+    public boolean hasDelegate() {
+        return this.delegate != null;
+    }
+
+    @Override
     public ControllerNode getLocalNode() {
-        return node(theInstance.getCluster().getLocalMember());
+        return localNode;
     }
 
     @Override
     public Set<ControllerNode> getNodes() {
-        ImmutableSet.Builder<ControllerNode> builder = ImmutableSet.builder();
-        for (Optional<DefaultControllerNode> optional : nodes.asMap().values()) {
-            builder.add(optional.get());
-        }
-        return builder.build();
+        return ImmutableSet.copyOf(allNodes.values());
     }
 
     @Override
     public ControllerNode getNode(NodeId nodeId) {
-        return nodes.getUnchecked(nodeId).orNull();
+        checkNotNull(nodeId, INSTANCE_ID_NULL);
+        return allNodes.get(nodeId);
     }
 
     @Override
     public State getState(NodeId nodeId) {
-        State state = states.get(nodeId);
-        return state == null ? State.INACTIVE : state;
+        checkNotNull(nodeId, INSTANCE_ID_NULL);
+        return nodeStates.get(nodeId);
     }
 
     @Override
     public ControllerNode addNode(NodeId nodeId, IpAddress ip, int tcpPort) {
-        return addNode(new DefaultControllerNode(nodeId, ip, tcpPort));
+        checkNotNull(nodeId, INSTANCE_ID_NULL);
+        checkNotNull(ip, "IP address must not be null");
+        checkArgument(tcpPort > 5000, "Tcp port must be greater than 5000");
+        ControllerNode node = new DefaultControllerNode(nodeId, ip, tcpPort);
+        allNodes.put(node.id(), node);
+        nodeStates.put(nodeId, State.INACTIVE);
+        delegate.notify(new ClusterEvent(ClusterEvent.Type.INSTANCE_ADDED, node));
+        return node;
     }
 
     @Override
     public void removeNode(NodeId nodeId) {
-        synchronized (this) {
-            rawNodes.remove(serialize(nodeId));
-            nodes.invalidate(nodeId);
+        checkNotNull(nodeId, INSTANCE_ID_NULL);
+        ControllerNode node = allNodes.remove(nodeId);
+        if (node != null) {
+            nodeStates.remove(nodeId);
+            delegate.notify(new ClusterEvent(ClusterEvent.Type.INSTANCE_REMOVED, node));
         }
     }
 
-    // Adds a new node based on the specified member
-    private synchronized ControllerNode addNode(DefaultControllerNode node) {
-        rawNodes.put(serialize(node.id()), serialize(node));
-        nodes.put(node.id(), Optional.of(node));
-        states.put(node.id(), State.ACTIVE);
-        return node;
+    private void establishSelfIdentity() {
+        try {
+            IpAddress ip = findLocalIp();
+            localNode = new DefaultControllerNode(new NodeId(ip.toString()), ip);
+            allNodes.put(localNode.id(), localNode);
+            nodeStates.put(localNode.id(), State.ACTIVE);
+            log.info("Local Node: {}", localNode);
+        } catch (SocketException e) {
+            throw new IllegalStateException("Cannot determine local IP", e);
+        }
     }
 
-    // Creates a controller node descriptor from the Hazelcast member.
-    private DefaultControllerNode node(Member member) {
-        IpAddress ip = memberAddress(member);
-        return new DefaultControllerNode(new NodeId(ip.toString()), ip);
+    private void heartbeat() {
+        try {
+            Set<ControllerNode> peers = allNodes.values()
+                    .stream()
+                    .filter(node -> !(node.id().equals(localNode.id())))
+                    .collect(Collectors.toSet());
+            byte[] hbMessagePayload = SERIALIZER.encode(new HeartbeatMessage(localNode, peers));
+            peers.forEach((node) -> {
+                heartbeatToPeer(hbMessagePayload, node);
+                State currentState = nodeStates.get(node.id());
+                double phi = failureDetector.phi(node.id());
+                if (phi >= PHI_FAILURE_THRESHOLD) {
+                    if (currentState == State.ACTIVE) {
+                        nodeStates.put(node.id(), State.INACTIVE);
+                        notifyStateChange(node.id(), State.ACTIVE, State.INACTIVE);
+                    }
+                } else {
+                    if (currentState == State.INACTIVE) {
+                        nodeStates.put(node.id(), State.ACTIVE);
+                        notifyStateChange(node.id(), State.INACTIVE, State.ACTIVE);
+                    }
+                }
+            });
+        } catch (Exception e) {
+            log.debug("Failed to send heartbeat", e);
+        }
     }
 
-    private IpAddress memberAddress(Member member) {
-        return IpAddress.valueOf(member.getSocketAddress().getAddress());
+    private void notifyStateChange(NodeId nodeId, State oldState, State newState) {
+        ControllerNode node = allNodes.get(nodeId);
+        if (newState == State.ACTIVE) {
+            delegate.notify(new ClusterEvent(ClusterEvent.Type.INSTANCE_ACTIVATED, node));
+        } else {
+            delegate.notify(new ClusterEvent(ClusterEvent.Type.INSTANCE_DEACTIVATED, node));
+        }
     }
 
-    // Interceptor for membership events.
-    private class InternalMembershipListener implements MembershipListener {
+    private void heartbeatToPeer(byte[] messagePayload, ControllerNode peer) {
+        Endpoint remoteEp = new Endpoint(peer.ip(), HEARTBEAT_FD_PORT);
+        try {
+            messagingService.sendAsync(remoteEp, HEARTBEAT_MESSAGE, messagePayload);
+        } catch (IOException e) {
+            log.debug("Sending heartbeat to {} failed", remoteEp, e);
+        }
+    }
+
+    private IpAddress findLocalIp() throws SocketException {
+        Enumeration<NetworkInterface> interfaces =
+                NetworkInterface.getNetworkInterfaces();
+        while (interfaces.hasMoreElements()) {
+            NetworkInterface iface = interfaces.nextElement();
+            Enumeration<InetAddress> inetAddresses =  iface.getInetAddresses();
+            while (inetAddresses.hasMoreElements()) {
+                IpAddress ip = IpAddress.valueOf(inetAddresses.nextElement());
+                if (AddressUtil.matchInterface(ip.toString(), clusterDefinition.getIpPrefix())) {
+                    return ip;
+                }
+            }
+        }
+        throw new IllegalStateException("Unable to determine local ip");
+    }
+
+    private class HeartbeatMessageHandler implements MessageHandler {
         @Override
-        public void memberAdded(MembershipEvent membershipEvent) {
-            log.info("Member {} added", membershipEvent.getMember());
-            ControllerNode node = addNode(node(membershipEvent.getMember()));
-            notifyDelegate(new ClusterEvent(INSTANCE_ACTIVATED, node));
-        }
-
-        @Override
-        public void memberRemoved(MembershipEvent membershipEvent) {
-            log.info("Member {} removed", membershipEvent.getMember());
-            NodeId nodeId = new NodeId(memberAddress(membershipEvent.getMember()).toString());
-            states.put(nodeId, State.INACTIVE);
-            notifyDelegate(new ClusterEvent(INSTANCE_DEACTIVATED, getNode(nodeId)));
-        }
-
-        @Override
-        public void memberAttributeChanged(MemberAttributeEvent memberAttributeEvent) {
-            log.info("Member {} attribute {} changed to {}",
-                     memberAttributeEvent.getMember(),
-                     memberAttributeEvent.getKey(),
-                     memberAttributeEvent.getValue());
+        public void handle(Message message) throws IOException {
+            HeartbeatMessage hb = SERIALIZER.decode(message.payload());
+            failureDetector.report(hb.source().id());
+            hb.knownPeers().forEach(node -> {
+                allNodes.put(node.id(), node);
+            });
         }
     }
+
+    private static class HeartbeatMessage {
+        private ControllerNode source;
+        private Set<ControllerNode> knownPeers;
+
+        public HeartbeatMessage(ControllerNode source, Set<ControllerNode> members) {
+            this.source = source;
+            this.knownPeers = ImmutableSet.copyOf(members);
+        }
+
+        public ControllerNode source() {
+            return source;
+        }
+
+        public Set<ControllerNode> knownPeers() {
+            return knownPeers;
+        }
+    }
+
 }
diff --git a/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/HazelcastClusterStore.java b/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/HazelcastClusterStore.java
new file mode 100644
index 0000000..d32f7a8
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/HazelcastClusterStore.java
@@ -0,0 +1,181 @@
+/*
+ * Copyright 2014 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.store.cluster.impl;
+
+import com.google.common.base.Optional;
+import com.google.common.cache.LoadingCache;
+import com.google.common.collect.ImmutableSet;
+import com.hazelcast.core.IMap;
+import com.hazelcast.core.Member;
+import com.hazelcast.core.MemberAttributeEvent;
+import com.hazelcast.core.MembershipEvent;
+import com.hazelcast.core.MembershipListener;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Service;
+import org.onosproject.cluster.ClusterEvent;
+import org.onosproject.cluster.ClusterStore;
+import org.onosproject.cluster.ClusterStoreDelegate;
+import org.onosproject.cluster.ControllerNode;
+import org.onosproject.cluster.DefaultControllerNode;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.store.hz.AbsentInvalidatingLoadingCache;
+import org.onosproject.store.hz.AbstractHazelcastStore;
+import org.onosproject.store.hz.OptionalCacheLoader;
+import org.onlab.packet.IpAddress;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static com.google.common.cache.CacheBuilder.newBuilder;
+import static org.onosproject.cluster.ClusterEvent.Type.INSTANCE_ACTIVATED;
+import static org.onosproject.cluster.ClusterEvent.Type.INSTANCE_DEACTIVATED;
+import static org.onosproject.cluster.ControllerNode.State;
+
+/**
+ * Distributed, Hazelcast-based implementation of the cluster nodes store.
+ */
+@Component(immediate = true, enabled = false)
+@Service
+public class HazelcastClusterStore
+        extends AbstractHazelcastStore<ClusterEvent, ClusterStoreDelegate>
+        implements ClusterStore {
+
+    private IMap<byte[], byte[]> rawNodes;
+    private LoadingCache<NodeId, Optional<DefaultControllerNode>> nodes;
+
+    private String listenerId;
+    private final MembershipListener listener = new InternalMembershipListener();
+    private final Map<NodeId, State> states = new ConcurrentHashMap<>();
+
+    private String nodesListenerId;
+
+    @Override
+    @Activate
+    public void activate() {
+        super.activate();
+        listenerId = theInstance.getCluster().addMembershipListener(listener);
+
+        rawNodes = theInstance.getMap("nodes");
+        OptionalCacheLoader<NodeId, DefaultControllerNode> nodeLoader
+                = new OptionalCacheLoader<>(serializer, rawNodes);
+        nodes = new AbsentInvalidatingLoadingCache<>(newBuilder().build(nodeLoader));
+        nodesListenerId = rawNodes.addEntryListener(new RemoteCacheEventHandler<>(nodes), true);
+
+        loadClusterNodes();
+
+        log.info("Started");
+    }
+
+    // Loads the initial set of cluster nodes
+    private void loadClusterNodes() {
+        for (Member member : theInstance.getCluster().getMembers()) {
+            addNode(node(member));
+        }
+    }
+
+    @Deactivate
+    public void deactivate() {
+        rawNodes.removeEntryListener(nodesListenerId);
+        theInstance.getCluster().removeMembershipListener(listenerId);
+        log.info("Stopped");
+    }
+
+    @Override
+    public ControllerNode getLocalNode() {
+        return node(theInstance.getCluster().getLocalMember());
+    }
+
+    @Override
+    public Set<ControllerNode> getNodes() {
+        ImmutableSet.Builder<ControllerNode> builder = ImmutableSet.builder();
+        for (Optional<DefaultControllerNode> optional : nodes.asMap().values()) {
+            builder.add(optional.get());
+        }
+        return builder.build();
+    }
+
+    @Override
+    public ControllerNode getNode(NodeId nodeId) {
+        return nodes.getUnchecked(nodeId).orNull();
+    }
+
+    @Override
+    public State getState(NodeId nodeId) {
+        State state = states.get(nodeId);
+        return state == null ? State.INACTIVE : state;
+    }
+
+    @Override
+    public ControllerNode addNode(NodeId nodeId, IpAddress ip, int tcpPort) {
+        return addNode(new DefaultControllerNode(nodeId, ip, tcpPort));
+    }
+
+    @Override
+    public void removeNode(NodeId nodeId) {
+        synchronized (this) {
+            rawNodes.remove(serialize(nodeId));
+            nodes.invalidate(nodeId);
+        }
+    }
+
+    // Adds a new node based on the specified member
+    private synchronized ControllerNode addNode(DefaultControllerNode node) {
+        rawNodes.put(serialize(node.id()), serialize(node));
+        nodes.put(node.id(), Optional.of(node));
+        states.put(node.id(), State.ACTIVE);
+        return node;
+    }
+
+    // Creates a controller node descriptor from the Hazelcast member.
+    private DefaultControllerNode node(Member member) {
+        IpAddress ip = memberAddress(member);
+        return new DefaultControllerNode(new NodeId(ip.toString()), ip);
+    }
+
+    private IpAddress memberAddress(Member member) {
+        return IpAddress.valueOf(member.getSocketAddress().getAddress());
+    }
+
+    // Interceptor for membership events.
+    private class InternalMembershipListener implements MembershipListener {
+        @Override
+        public void memberAdded(MembershipEvent membershipEvent) {
+            log.info("Member {} added", membershipEvent.getMember());
+            ControllerNode node = addNode(node(membershipEvent.getMember()));
+            notifyDelegate(new ClusterEvent(INSTANCE_ACTIVATED, node));
+        }
+
+        @Override
+        public void memberRemoved(MembershipEvent membershipEvent) {
+            log.info("Member {} removed", membershipEvent.getMember());
+            NodeId nodeId = new NodeId(memberAddress(membershipEvent.getMember()).toString());
+            states.put(nodeId, State.INACTIVE);
+            notifyDelegate(new ClusterEvent(INSTANCE_DEACTIVATED, getNode(nodeId)));
+        }
+
+        @Override
+        public void memberAttributeChanged(MemberAttributeEvent memberAttributeEvent) {
+            log.info("Member {} attribute {} changed to {}",
+                     memberAttributeEvent.getMember(),
+                     memberAttributeEvent.getKey(),
+                     memberAttributeEvent.getValue());
+        }
+    }
+}