diff --git a/api/pom.xml b/api/pom.xml
index cb10179..b4d81dd 100644
--- a/api/pom.xml
+++ b/api/pom.xml
@@ -38,6 +38,7 @@
             <groupId>org.opencord</groupId>
             <artifactId>sadis-api</artifactId>
             <version>${sadis.api.version}</version>
+            <scope>provided</scope>
         </dependency>
     </dependencies>
 
diff --git a/app/pom.xml b/app/pom.xml
index 02f912d..287ab4d 100644
--- a/app/pom.xml
+++ b/app/pom.xml
@@ -72,6 +72,14 @@
 
         <dependency>
             <groupId>org.onosproject</groupId>
+            <artifactId>onos-api</artifactId>
+            <version>${onos.version}</version>
+            <classifier>tests</classifier>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
             <artifactId>onos-core-serializers</artifactId>
             <version>${onos.version}</version>
             <scope>provided</scope>
diff --git a/app/src/main/java/org/opencord/olt/impl/ConsistentHasher.java b/app/src/main/java/org/opencord/olt/impl/ConsistentHasher.java
new file mode 100644
index 0000000..52e9b96
--- /dev/null
+++ b/app/src/main/java/org/opencord/olt/impl/ConsistentHasher.java
@@ -0,0 +1,130 @@
+/*
+ * Copyright 2020-present Open Networking Foundation
+ *
+ * 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.opencord.olt.impl;
+
+import com.google.common.hash.Hashing;
+import org.onosproject.cluster.NodeId;
+
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Returns a server hosting a given key based on consistent hashing.
+ */
+public class ConsistentHasher {
+
+    private static class Entry implements Comparable<Entry> {
+        private final NodeId server;
+        private final int hash;
+
+        public Entry(NodeId server, int hash) {
+            this.server = server;
+            this.hash = hash;
+        }
+
+        public Entry(int hash) {
+            server = null;
+            this.hash = hash;
+        }
+
+        @Override
+        public int compareTo(Entry o) {
+            if (this.hash > o.hash) {
+                return 1;
+            } else if (this.hash < o.hash) {
+                return -1;
+            } // else
+            return 0;
+        }
+    }
+
+    private final int weight;
+
+    private List<Entry> table;
+
+    /**
+     * Creates a new hasher with the given list of servers.
+     *
+     * @param servers list of servers
+     * @param weight weight
+     */
+    public ConsistentHasher(List<NodeId> servers, int weight) {
+        this.weight = weight;
+
+        this.table = new ArrayList<>();
+
+        servers.forEach(this::addServer);
+    }
+
+    /**
+     * Adds a new server to the list of servers.
+     *
+     * @param server server ID
+     */
+    public synchronized void addServer(NodeId server) {
+        // Add weight number of buckets for each server
+        for (int i = 0; i < weight; i++) {
+            String label = server.toString() + i;
+            int hash = getHash(label);
+            Entry e = new Entry(server, hash);
+            table.add(e);
+        }
+
+        Collections.sort(table);
+    }
+
+    /**
+     * Removes a server from the list of servers.
+     *
+     * @param server server ID
+     */
+    public synchronized void removeServer(NodeId server) {
+        table.removeIf(e -> e.server.equals(server));
+    }
+
+    /**
+     * Hashes a given input string and finds that server that should
+     * handle the given ID.
+     *
+     * @param s input
+     * @return server ID
+     */
+    public synchronized NodeId hash(String s) {
+        Entry temp = new Entry(getHash(s));
+
+        int pos = Collections.binarySearch(this.table, temp);
+
+        // translate a negative not-found result into the closest following match
+        if (pos < 0) {
+            pos = Math.abs(pos + 1);
+        }
+
+        // wraparound if the hash was after the last entry in the table
+        if (pos == this.table.size()) {
+            pos = 0;
+        }
+
+        return table.get(pos).server;
+    }
+
+    private int getHash(String s) {
+        // stable, uniformly-distributed hash
+        return Hashing.murmur3_128().hashString(s, Charset.defaultCharset()).asInt();
+    }
+}
diff --git a/app/src/main/java/org/opencord/olt/impl/Olt.java b/app/src/main/java/org/opencord/olt/impl/Olt.java
index 50a8fd6..000d5e6 100644
--- a/app/src/main/java/org/opencord/olt/impl/Olt.java
+++ b/app/src/main/java/org/opencord/olt/impl/Olt.java
@@ -16,14 +16,18 @@
 package org.opencord.olt.impl;
 
 import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 import org.onlab.packet.VlanId;
+import org.onlab.util.KryoNamespace;
 import org.onosproject.cfg.ComponentConfigService;
+import org.onosproject.cluster.ClusterEvent;
+import org.onosproject.cluster.ClusterEventListener;
+import org.onosproject.cluster.ClusterService;
+import org.onosproject.cluster.ControllerNode;
+import org.onosproject.cluster.NodeId;
 import org.onosproject.core.ApplicationId;
 import org.onosproject.core.CoreService;
 import org.onosproject.event.AbstractListenerManager;
-import org.onosproject.mastership.MastershipService;
 import org.onosproject.net.AnnotationKeys;
 import org.onosproject.net.ConnectPoint;
 import org.onosproject.net.Device;
@@ -39,6 +43,10 @@
 import org.onosproject.net.flowobjective.ObjectiveContext;
 import org.onosproject.net.flowobjective.ObjectiveError;
 import org.onosproject.net.meter.MeterId;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.ConsistentMultimap;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.StorageService;
 import org.opencord.olt.AccessDeviceEvent;
 import org.opencord.olt.AccessDeviceListener;
 import org.opencord.olt.AccessDeviceService;
@@ -60,23 +68,30 @@
 import org.slf4j.Logger;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Dictionary;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.stream.Collectors;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
+import static java.util.stream.Collectors.collectingAndThen;
+import static java.util.stream.Collectors.groupingBy;
+import static java.util.stream.Collectors.mapping;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
 import static org.onlab.util.Tools.get;
 import static org.onlab.util.Tools.groupedThreads;
-import static org.opencord.olt.impl.OsgiPropertyConstants.*;
+import static org.opencord.olt.impl.OsgiPropertyConstants.DEFAULT_BP_ID;
+import static org.opencord.olt.impl.OsgiPropertyConstants.DEFAULT_BP_ID_DEFAULT;
+import static org.opencord.olt.impl.OsgiPropertyConstants.DEFAULT_MCAST_SERVICE_NAME;
+import static org.opencord.olt.impl.OsgiPropertyConstants.DEFAULT_MCAST_SERVICE_NAME_DEFAULT;
 import static org.slf4j.LoggerFactory.getLogger;
 
 /**
@@ -95,6 +110,8 @@
     private static final short EAPOL_DEFAULT_VLAN = 4091;
     private static final String NO_UPLINK_PORT = "No uplink port found for OLT device {}";
 
+    public static final int HASH_WEIGHT = 10;
+
     private final Logger log = getLogger(getClass());
 
     private static final String NNI = "nni-";
@@ -103,9 +120,6 @@
     protected FlowObjectiveService flowObjectiveService;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY)
-    protected MastershipService mastershipService;
-
-    @Reference(cardinality = ReferenceCardinality.MANDATORY)
     protected DeviceService deviceService;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY)
@@ -123,6 +137,12 @@
     @Reference(cardinality = ReferenceCardinality.MANDATORY)
     protected AccessDeviceMeterService oltMeterService;
 
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected StorageService storageService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected ClusterService clusterService;
+
     /**
      * Default bandwidth profile id that is used for authentication trap flows.
      **/
@@ -134,6 +154,9 @@
     protected String multicastServiceName = DEFAULT_MCAST_SERVICE_NAME;
 
     private final DeviceListener deviceListener = new InternalDeviceListener();
+    private final ClusterEventListener clusterListener = new InternalClusterListener();
+
+    private ConsistentHasher hasher;
 
     protected BaseInformationService<SubscriberAndDeviceInformation> subsService;
     private BaseInformationService<BandwidthProfileInformation> bpService;
@@ -144,7 +167,7 @@
 
     protected ExecutorService eventExecutor;
 
-    private Map<ConnectPoint, Set<UniTagInformation>> programmedSubs;
+    private ConsistentMultimap<ConnectPoint, UniTagInformation> programmedSubs;
 
     @Activate
     public void activate(ComponentContext context) {
@@ -163,18 +186,37 @@
                 .preSetProperty("org.onosproject.net.meter.impl.MeterManager",
                                 "purgeOnDisconnection", "true");
         componentConfigService.registerProperties(getClass());
-        programmedSubs = Maps.newConcurrentMap();
+
+        KryoNamespace serializer = KryoNamespace.newBuilder()
+                .register(KryoNamespaces.API)
+                .register(UniTagInformation.class)
+                .build();
+
+        programmedSubs = storageService.<ConnectPoint, UniTagInformation>consistentMultimapBuilder()
+                .withName("volt-programmed-subs")
+                .withSerializer(Serializer.using(serializer))
+                .withApplicationId(appId)
+                .build();
 
         eventDispatcher.addSink(AccessDeviceEvent.class, listenerRegistry);
 
         subsService = sadisService.getSubscriberInfoService();
         bpService = sadisService.getBandwidthProfileService();
 
+        List<NodeId> readyNodes = clusterService.getNodes().stream()
+                .filter(c -> clusterService.getState(c.id()) == ControllerNode.State.READY)
+                .map(ControllerNode::id)
+                .collect(toList());
+        hasher = new ConsistentHasher(readyNodes, HASH_WEIGHT);
+        clusterService.addListener(clusterListener);
+
         // look for all provisioned devices in Sadis and create EAPOL flows for the
         // UNI ports
         Iterable<Device> devices = deviceService.getDevices();
         for (Device d : devices) {
-            checkAndCreateDeviceFlows(d);
+            if (isDeviceMine(d.id())) {
+                checkAndCreateDeviceFlows(d);
+            }
         }
 
         deviceService.addListener(deviceListener);
@@ -184,6 +226,7 @@
     @Deactivate
     public void deactivate() {
         componentConfigService.unregisterProperties(getClass(), false);
+        clusterService.removeListener(clusterListener);
         deviceService.removeListener(deviceListener);
         eventDispatcher.removeSink(AccessDeviceEvent.class);
         log.info("Stopped");
@@ -253,7 +296,7 @@
         DeviceId deviceId = connectPoint.deviceId();
         PortNumber subscriberPortNo = connectPoint.port();
 
-        Set<UniTagInformation> uniTagInformationSet = programmedSubs.get(connectPoint);
+        Collection<? extends UniTagInformation> uniTagInformationSet = programmedSubs.get(connectPoint).value();
         if (uniTagInformationSet == null || uniTagInformationSet.isEmpty()) {
             log.warn("Subscriber on connectionPoint {} was not previously programmed, " +
                              "no need to remove it", connectPoint);
@@ -373,7 +416,10 @@
 
     @Override
     public ImmutableMap<ConnectPoint, Set<UniTagInformation>> getProgSubs() {
-        return ImmutableMap.copyOf(programmedSubs);
+        return programmedSubs.stream()
+                .collect(collectingAndThen(
+                        groupingBy(Map.Entry::getKey, mapping(Map.Entry::getValue, toSet())),
+                        ImmutableMap::copyOf));
     }
 
     @Override
@@ -562,19 +608,11 @@
     }
 
     private void updateProgrammedSubscriber(ConnectPoint connectPoint, UniTagInformation tagInformation, Boolean add) {
-        programmedSubs.compute(connectPoint, (k, v) -> {
-            if (add) {
-                if (v == null) {
-                    v = Sets.newHashSet();
-                }
-                v.add(tagInformation);
-            } else {
-                if (v != null) {
-                    v.remove(tagInformation);
-                }
-            }
-            return v;
-        });
+        if (add) {
+            programmedSubs.put(connectPoint, tagInformation);
+        } else {
+            programmedSubs.remove(connectPoint, tagInformation);
+        }
     }
 
     /**
@@ -768,10 +806,6 @@
      * @param dev Device to look for
      */
     private void checkAndCreateDeviceFlows(Device dev) {
-        // we create only for the ones we are master of
-        if (!mastershipService.isLocalMaster(dev.id())) {
-            return;
-        }
         // check if this device is provisioned in Sadis
         SubscriberAndDeviceInformation deviceInfo = getOltInfo(dev);
         log.info("checkAndCreateDeviceFlows: deviceInfo {}", deviceInfo);
@@ -779,7 +813,7 @@
         if (deviceInfo != null) {
             // This is an OLT device as per Sadis, we create flows for UNI and NNI ports
             for (Port p : deviceService.getPorts(dev.id())) {
-                if (PortNumber.LOCAL.equals(p.number())) {
+                if (PortNumber.LOCAL.equals(p.number()) || !p.isEnabled()) {
                     continue;
                 }
                 if (isUniPort(dev, p)) {
@@ -871,6 +905,21 @@
         return subsService.get(devSerialNo);
     }
 
+    /**
+     * Determines if this instance should handle this device based on
+     * consistent hashing.
+     *
+     * @param id device ID
+     * @return true if this instance should handle the device, otherwise false
+     */
+    private boolean isDeviceMine(DeviceId id) {
+        NodeId nodeId = hasher.hash(id.toString());
+        if (log.isDebugEnabled()) {
+            log.debug("Node that will handle {} is {}", id, nodeId);
+        }
+        return nodeId.equals(clusterService.getLocalNode().id());
+    }
+
     private class InternalDeviceListener implements DeviceListener {
         private Set<DeviceId> programmedDevices = Sets.newConcurrentHashSet();
 
@@ -888,6 +937,11 @@
                     return;
                 }
 
+                // Only handle the event if the device belongs to us
+                if (!isDeviceMine(event.subject().id())) {
+                    return;
+                }
+
                 log.debug("OLT got {} event for {} {}", eventType, event.subject(), event.port());
 
                 if (getOltInfo(dev) == null) {
@@ -936,8 +990,8 @@
                         if (!isUniPort(dev, port)) {
                             break;
                         }
-                        Set<UniTagInformation> uniTagInformationSet = programmedSubs
-                                .get(new ConnectPoint(devId, port.number()));
+                        ConnectPoint cp = new ConnectPoint(devId, port.number());
+                        Collection<? extends UniTagInformation> uniTagInformationSet = programmedSubs.get(cp).value();
                         if (uniTagInformationSet == null || uniTagInformationSet.isEmpty()) {
                             if (port.isEnabled() && !port.number().equals(PortNumber.LOCAL)) {
                                 log.info("eapol will be processed for port updated {}", port);
@@ -988,6 +1042,7 @@
         private void handleDeviceDisconnection(Device device, boolean sendUniEvent) {
             programmedDevices.remove(device.id());
             removeAllSubscribers(device.id());
+            oltMeterService.clearMeters(device.id());
             post(new AccessDeviceEvent(
                     AccessDeviceEvent.Type.DEVICE_DISCONNECTED, device.id(),
                     null, null, null));
@@ -1008,12 +1063,25 @@
         }
 
         private void removeAllSubscribers(DeviceId deviceId) {
-            List<ConnectPoint> connectPoints = programmedSubs.keySet().stream()
-                    .filter(ks -> Objects.equals(ks.deviceId(), deviceId))
-                    .collect(Collectors.toList());
+            List<Map.Entry<ConnectPoint, UniTagInformation>> subs = programmedSubs.stream()
+                    .filter(e -> e.getKey().deviceId().equals(deviceId))
+                    .collect(toList());
 
-            connectPoints.forEach(cp -> programmedSubs.remove(cp));
+            subs.forEach(e -> programmedSubs.remove(e.getKey(), e.getValue()));
         }
 
     }
+
+    private class InternalClusterListener implements ClusterEventListener {
+
+        @Override
+        public void event(ClusterEvent event) {
+            if (event.type() == ClusterEvent.Type.INSTANCE_READY) {
+                hasher.addServer(event.subject().id());
+            }
+            if (event.type() == ClusterEvent.Type.INSTANCE_DEACTIVATED) {
+                hasher.removeServer(event.subject().id());
+            }
+        }
+    }
 }
diff --git a/app/src/main/java/org/opencord/olt/impl/OltFlowService.java b/app/src/main/java/org/opencord/olt/impl/OltFlowService.java
index bf49483..af8ba4a 100644
--- a/app/src/main/java/org/opencord/olt/impl/OltFlowService.java
+++ b/app/src/main/java/org/opencord/olt/impl/OltFlowService.java
@@ -380,14 +380,6 @@
             return;
         }
 
-        if (!mastershipService.isLocalMaster(devId)) {
-            log.warn("The master of the device {} is another instance", devId);
-            if (filterFuture != null) {
-                filterFuture.complete(ObjectiveError.DEVICEMISSING);
-            }
-            return;
-        }
-
         BandwidthProfileInformation bpInfo = getBandwidthProfileInformation(bpId);
         if (bpInfo == null) {
             log.warn("Bandwidth profile {} is not found. Authentication flow"
diff --git a/app/src/main/java/org/opencord/olt/impl/OltMeterService.java b/app/src/main/java/org/opencord/olt/impl/OltMeterService.java
index 825604c..7a3a3a3 100644
--- a/app/src/main/java/org/opencord/olt/impl/OltMeterService.java
+++ b/app/src/main/java/org/opencord/olt/impl/OltMeterService.java
@@ -15,13 +15,10 @@
  */
 package org.opencord.olt.impl;
 
-import com.google.common.collect.HashMultimap;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Maps;
-import com.google.common.collect.Multimaps;
-import com.google.common.collect.SetMultimap;
-import com.google.common.collect.Sets;
+import org.onlab.util.KryoNamespace;
 import org.onlab.util.Tools;
 import org.onosproject.cfg.ComponentConfigService;
 import org.onosproject.core.ApplicationId;
@@ -40,6 +37,10 @@
 import org.onosproject.net.meter.MeterListener;
 import org.onosproject.net.meter.MeterRequest;
 import org.onosproject.net.meter.MeterService;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.ConsistentMultimap;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.StorageService;
 import org.opencord.olt.internalapi.AccessDeviceMeterService;
 import org.opencord.sadis.BandwidthProfileInformation;
 import org.osgi.service.component.ComponentContext;
@@ -54,18 +55,21 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Dictionary;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
-import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
 
+import static java.util.stream.Collectors.collectingAndThen;
+import static java.util.stream.Collectors.groupingBy;
+import static java.util.stream.Collectors.mapping;
+import static java.util.stream.Collectors.toSet;
 import static org.onlab.util.Tools.groupedThreads;
 import static org.opencord.olt.impl.OsgiPropertyConstants.DELETE_METERS;
 import static org.opencord.olt.impl.OsgiPropertyConstants.DELETE_METERS_DEFAULT;
@@ -88,11 +92,13 @@
     @Reference(cardinality = ReferenceCardinality.MANDATORY)
     protected ComponentConfigService componentConfigService;
 
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected StorageService storageService;
+
     protected boolean deleteMeters = true;
 
-    protected SetMultimap<String, MeterKey> bpInfoToMeter =
-            Multimaps.synchronizedSetMultimap(HashMultimap.create());
-    protected Set<MeterKey> programmedMeters;
+    ConsistentMultimap<String, MeterKey> bpInfoToMeter;
+
     private ApplicationId appId;
     private static final String APP_NAME = "org.opencord.olt";
 
@@ -107,7 +113,19 @@
         eventExecutor = Executors.newFixedThreadPool(5, groupedThreads("onos/olt",
                 "events-%d", log));
         appId = coreService.registerApplication(APP_NAME);
-        programmedMeters = Sets.newConcurrentHashSet();
+        modified(context);
+
+        KryoNamespace serializer = KryoNamespace.newBuilder()
+                .register(KryoNamespaces.API)
+                .register(MeterKey.class)
+                .build();
+
+        bpInfoToMeter = storageService.<String, MeterKey>consistentMultimapBuilder()
+                .withName("volt-bp-info-to-meter")
+                .withSerializer(Serializer.using(serializer))
+                .withApplicationId(appId)
+                .build();
+
         meterService.addListener(meterListener);
         componentConfigService.registerProperties(getClass());
         log.info("Olt Meter service started");
@@ -131,23 +149,25 @@
 
     @Override
     public ImmutableMap<String, Collection<MeterKey>> getBpMeterMappings() {
-        return ImmutableMap.copyOf(bpInfoToMeter.asMap());
+        return bpInfoToMeter.stream()
+                .collect(collectingAndThen(
+                        groupingBy(Map.Entry::getKey, mapping(Map.Entry::getValue, toSet())),
+                        ImmutableMap::copyOf));
     }
 
-    @Override
-    public void addMeterIdToBpMapping(DeviceId deviceId, MeterId meterId, String bandwidthProfile) {
+    void addMeterIdToBpMapping(DeviceId deviceId, MeterId meterId, String bandwidthProfile) {
         bpInfoToMeter.put(bandwidthProfile, MeterKey.key(deviceId, meterId));
     }
 
     @Override
     public MeterId getMeterIdFromBpMapping(DeviceId deviceId, String bandwidthProfile) {
-        if (bpInfoToMeter.get(bandwidthProfile) == null) {
+        if (bpInfoToMeter.get(bandwidthProfile).value().isEmpty()) {
             log.warn("Bandwidth Profile '{}' is not currently mapped to a meter",
                     bandwidthProfile);
             return null;
         }
 
-        Optional<MeterKey> meterKeyForDevice = bpInfoToMeter.get(bandwidthProfile)
+        Optional<? extends MeterKey> meterKeyForDevice = bpInfoToMeter.get(bandwidthProfile).value()
                 .stream()
                 .filter(meterKey -> meterKey.deviceId().equals(deviceId))
                 .findFirst();
@@ -164,7 +184,9 @@
 
     @Override
     public ImmutableSet<MeterKey> getProgMeters() {
-        return ImmutableSet.copyOf(programmedMeters);
+        return bpInfoToMeter.stream()
+                .map(Map.Entry::getValue)
+                .collect(ImmutableSet.toImmutableSet());
     }
 
     @Override
@@ -210,11 +232,19 @@
         Meter meter = meterService.submit(meterRequest);
         meterIdRef.set(meter.id());
         addMeterIdToBpMapping(deviceId, meterIdRef.get(), bpInfo.id());
-        programmedMeters.add(MeterKey.key(deviceId, meter.id()));
         log.info("Meter is created. Meter Id {}", meter.id());
         return meter.id();
     }
 
+    @Override
+    public void clearMeters(DeviceId deviceId) {
+        List<Map.Entry<String, MeterKey>> meters = bpInfoToMeter.stream()
+                .filter(e -> e.getValue().deviceId().equals(deviceId))
+                .collect(Collectors.toList());
+
+        meters.forEach(e -> bpInfoToMeter.remove(e.getKey(), e.getValue()));
+    }
+
     private List<Band> createMeterBands(BandwidthProfileInformation bpInfo) {
         List<Band> meterBands = new ArrayList<>();
 
@@ -257,7 +287,6 @@
                 }
                 if (MeterEvent.Type.METER_REMOVED.equals(meterEvent.type())) {
                     log.info("Meter Removed Event is received for {}", meter.id());
-                    programmedMeters.remove(key);
                     pendingRemoveMeters.remove(key);
                     removeMeterFromBpMapping(key);
                 }
@@ -294,15 +323,11 @@
         }
 
         private void removeMeterFromBpMapping(MeterKey meterKey) {
-            Iterator<Map.Entry<String, MeterKey>> iterator = bpInfoToMeter.entries().iterator();
-            while (iterator.hasNext()) {
-                Map.Entry<String, MeterKey> entry = iterator.next();
-                if (entry.getValue().equals(meterKey)) {
-                    iterator.remove();
-                    log.info("Deleted meter for MeterKey {} - Last prog meters {}", meterKey, programmedMeters);
-                    break;
-                }
-            }
+            List<Map.Entry<String, MeterKey>> meters = bpInfoToMeter.stream()
+                    .filter(e -> e.getValue().equals(meterKey))
+                    .collect(Collectors.toList());
+
+            meters.forEach(e -> bpInfoToMeter.remove(e.getKey(), e.getValue()));
         }
     }
 }
diff --git a/app/src/main/java/org/opencord/olt/internalapi/AccessDeviceMeterService.java b/app/src/main/java/org/opencord/olt/internalapi/AccessDeviceMeterService.java
index 4f07131..48570cd 100644
--- a/app/src/main/java/org/opencord/olt/internalapi/AccessDeviceMeterService.java
+++ b/app/src/main/java/org/opencord/olt/internalapi/AccessDeviceMeterService.java
@@ -39,16 +39,6 @@
     ImmutableMap<String, Collection<MeterKey>> getBpMeterMappings();
 
     /**
-     * Adds a bandwidthProfile-meterKey (device / meter) mapping that have been programmed
-     * in the data plane.
-     *
-     * @param deviceId         the access device id
-     * @param meterId          the meter id that is mapped to the bandwidth profile
-     * @param bandwidthProfile the bandwidth profile id
-     */
-    void addMeterIdToBpMapping(DeviceId deviceId, MeterId meterId, String bandwidthProfile);
-
-    /**
      * Returns the meter id for a given bandwidth profile.
      *
      * @param deviceId         the access device id
@@ -77,5 +67,10 @@
     MeterId createMeter(DeviceId deviceId, BandwidthProfileInformation bpInfo,
                         CompletableFuture<Object> meterFuture);
 
-
+    /**
+     * Clears out bandwidth profile to meter mappings for the given device.
+     *
+     * @param deviceId device ID
+     */
+    void clearMeters(DeviceId deviceId);
 }
diff --git a/app/src/test/java/org/opencord/olt/impl/ConsistentHasherTest.java b/app/src/test/java/org/opencord/olt/impl/ConsistentHasherTest.java
new file mode 100644
index 0000000..1f682cb
--- /dev/null
+++ b/app/src/test/java/org/opencord/olt/impl/ConsistentHasherTest.java
@@ -0,0 +1,74 @@
+/*
+ * Copyright 2020-present Open Networking Foundation
+ *
+ * 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.opencord.olt.impl;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.net.DeviceId;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+public class ConsistentHasherTest {
+
+    private static final int WEIGHT = 10;
+
+    private static final NodeId N1 = new NodeId("10.0.0.1");
+    private static final NodeId N2 = new NodeId("10.0.0.2");
+    private static final NodeId N3 = new NodeId("10.0.0.3");
+
+    private ConsistentHasher hasher;
+
+    @Before
+    public void setUp() {
+        List<NodeId> servers = new ArrayList<>();
+        servers.add(N1);
+        servers.add(N2);
+
+        hasher = new ConsistentHasher(servers, WEIGHT);
+    }
+
+    @Test
+    public void testHasher() {
+        DeviceId deviceId = DeviceId.deviceId("foo");
+        NodeId server = hasher.hash(deviceId.toString());
+
+        assertThat(server, equalTo(N1));
+
+        deviceId = DeviceId.deviceId("bsaf");
+        server = hasher.hash(deviceId.toString());
+
+        assertThat(server, equalTo(N2));
+    }
+
+    @Test
+    public void testAddServer() {
+        DeviceId deviceId = DeviceId.deviceId("foo");
+        NodeId server = hasher.hash(deviceId.toString());
+
+        assertThat(server, equalTo(N1));
+
+        hasher.addServer(N3);
+
+        server = hasher.hash(deviceId.toString());
+
+        assertThat(server, equalTo(N3));
+    }
+}
diff --git a/app/src/test/java/org/opencord/olt/impl/OltFlowTest.java b/app/src/test/java/org/opencord/olt/impl/OltFlowTest.java
index 16b407b..e0bd109 100644
--- a/app/src/test/java/org/opencord/olt/impl/OltFlowTest.java
+++ b/app/src/test/java/org/opencord/olt/impl/OltFlowTest.java
@@ -188,11 +188,6 @@
         }
 
         @Override
-        public void addMeterIdToBpMapping(DeviceId deviceId, MeterId meterId, String bandwidthProfile) {
-
-        }
-
-        @Override
         public MeterId getMeterIdFromBpMapping(DeviceId deviceId, String bandwidthProfile) {
             return null;
         }
@@ -208,6 +203,10 @@
                                    CompletableFuture<Object> meterFuture) {
             return usMeterId;
         }
+
+        @Override
+        public void clearMeters(DeviceId deviceId) {
+        }
     }
 
     private class MockOltFlowObjectiveService implements org.onosproject.net.flowobjective.FlowObjectiveService {
diff --git a/app/src/test/java/org/opencord/olt/impl/OltMeterTest.java b/app/src/test/java/org/opencord/olt/impl/OltMeterTest.java
index da7e3e1..000ea35 100644
--- a/app/src/test/java/org/opencord/olt/impl/OltMeterTest.java
+++ b/app/src/test/java/org/opencord/olt/impl/OltMeterTest.java
@@ -15,12 +15,11 @@
  */
 package org.opencord.olt.impl;
 
-import com.google.common.collect.HashMultimap;
 import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Multimaps;
-import com.google.common.collect.Sets;
 import org.junit.Before;
 import org.junit.Test;
+import org.onosproject.cfg.ComponentConfigAdapter;
+import org.onosproject.core.CoreServiceAdapter;
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.meter.DefaultMeter;
 import org.onosproject.net.meter.Meter;
@@ -28,6 +27,7 @@
 import org.onosproject.net.meter.MeterKey;
 import org.onosproject.net.meter.MeterListener;
 import org.onosproject.net.meter.MeterRequest;
+import org.onosproject.store.service.TestStorageService;
 import org.opencord.sadis.BandwidthProfileInformation;
 
 import java.util.Collection;
@@ -41,9 +41,12 @@
     @Before
     public void setUp() {
         oltMeterService = new OltMeterService();
-        oltMeterService.bpInfoToMeter = Multimaps.synchronizedSetMultimap(HashMultimap.create());
-        oltMeterService.programmedMeters = Sets.newConcurrentHashSet();
+        oltMeterService.storageService = new TestStorageService();
         oltMeterService.meterService = new MockMeterService();
+        oltMeterService.coreService = new CoreServiceAdapter();
+        oltMeterService.componentConfigService = new ComponentConfigAdapter();
+        oltMeterService.activate(null);
+        oltMeterService.bpInfoToMeter = new MockConsistentMultimap<>();
     }
 
     @Test
diff --git a/app/src/test/java/org/opencord/olt/impl/TestBase.java b/app/src/test/java/org/opencord/olt/impl/TestBase.java
index b8447a9..bc5a4d6 100644
--- a/app/src/test/java/org/opencord/olt/impl/TestBase.java
+++ b/app/src/test/java/org/opencord/olt/impl/TestBase.java
@@ -15,18 +15,33 @@
  */
 package org.opencord.olt.impl;
 
+import com.google.common.collect.HashMultimap;
 import com.google.common.collect.Maps;
+import com.google.common.collect.Multiset;
 import org.onlab.packet.Ip4Address;
 import org.onlab.packet.MacAddress;
 import org.onosproject.core.DefaultApplicationId;
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.meter.MeterId;
+import org.onosproject.store.service.AsyncConsistentMultimap;
+import org.onosproject.store.service.ConsistentMultimap;
+import org.onosproject.store.service.ConsistentMultimapBuilder;
+import org.onosproject.store.service.MultimapEventListener;
+import org.onosproject.store.service.TestConsistentMultimap;
+import org.onosproject.store.service.Versioned;
 import org.opencord.sadis.BandwidthProfileInformation;
 import org.opencord.sadis.BaseInformationService;
 import org.opencord.sadis.SadisService;
 import org.opencord.sadis.SubscriberAndDeviceInformation;
 
+import java.util.AbstractMap;
+import java.util.Collection;
+import java.util.Iterator;
 import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
 
 public class TestBase {
 
@@ -124,4 +139,187 @@
             this.setCircuitId(circuitId);
         }
     }
+
+    class MockConsistentMultimap<K, V> implements ConsistentMultimap<K, V> {
+        private HashMultimap<K, Versioned<V>> innermap;
+        private AtomicLong counter = new AtomicLong();
+
+        public MockConsistentMultimap() {
+            this.innermap = HashMultimap.create();
+        }
+
+        private Versioned<V> version(V v) {
+            return new Versioned<>(v, counter.incrementAndGet(), System.currentTimeMillis());
+        }
+
+        private Versioned<Collection<? extends V>> versionCollection(Collection<? extends V> collection) {
+            return new Versioned<>(collection, counter.incrementAndGet(), System.currentTimeMillis());
+        }
+
+        @Override
+        public int size() {
+            return innermap.size();
+        }
+
+        @Override
+        public boolean isEmpty() {
+            return innermap.isEmpty();
+        }
+
+        @Override
+        public boolean containsKey(K key) {
+            return innermap.containsKey(key);
+        }
+
+        @Override
+        public boolean containsValue(V value) {
+            return innermap.containsValue(value);
+        }
+
+        @Override
+        public boolean containsEntry(K key, V value) {
+            return innermap.containsEntry(key, value);
+        }
+
+        @Override
+        public boolean put(K key, V value) {
+            return innermap.put(key, version(value));
+        }
+
+        @Override
+        public Versioned<Collection<? extends V>> putAndGet(K key, V value) {
+            innermap.put(key, version(value));
+            return (Versioned<Collection<? extends V>>) innermap.get(key);
+        }
+
+        @Override
+        public boolean remove(K key, V value) {
+            return innermap.remove(key, value);
+        }
+
+        @Override
+        public Versioned<Collection<? extends V>> removeAndGet(K key, V value) {
+            innermap.remove(key, value);
+            return (Versioned<Collection<? extends V>>) innermap.get(key);
+        }
+
+        @Override
+        public boolean removeAll(K key, Collection<? extends V> values) {
+            return false;
+        }
+
+        @Override
+        public Versioned<Collection<? extends V>> removeAll(K key) {
+            return null;
+        }
+
+        @Override
+        public boolean putAll(K key, Collection<? extends V> values) {
+            return false;
+        }
+
+        @Override
+        public Versioned<Collection<? extends V>> replaceValues(K key, Collection<V> values) {
+            return null;
+        }
+
+        @Override
+        public void clear() {
+            innermap.clear();
+        }
+
+        @Override
+        public Versioned<Collection<? extends V>> get(K key) {
+            Collection<? extends V> values = innermap.get(key).stream()
+                    .map(v -> v.value())
+                    .collect(Collectors.toList());
+            return versionCollection(values);
+        }
+
+        @Override
+        public Set<K> keySet() {
+            return innermap.keySet();
+        }
+
+        @Override
+        public Multiset<K> keys() {
+            return innermap.keys();
+        }
+
+        @Override
+        public Multiset<V> values() {
+            return null;
+        }
+
+        @Override
+        public Collection<Map.Entry<K, V>> entries() {
+            return null;
+        }
+
+        @Override
+        public Iterator<Map.Entry<K, V>> iterator() {
+            return new ConsistentMultimapIterator(innermap.entries().iterator());
+        }
+
+        @Override
+        public Map<K, Collection<V>> asMap() {
+            return null;
+        }
+
+        @Override
+        public void addListener(MultimapEventListener<K, V> listener, Executor executor) {
+        }
+
+        @Override
+        public void removeListener(MultimapEventListener<K, V> listener) {
+        }
+
+        @Override
+        public String name() {
+            return "mock multimap";
+        }
+
+        @Override
+        public Type primitiveType() {
+            return null;
+        }
+
+        private class ConsistentMultimapIterator implements Iterator<Map.Entry<K, V>> {
+
+            private final Iterator<Map.Entry<K, Versioned<V>>> it;
+
+            public ConsistentMultimapIterator(Iterator<Map.Entry<K, Versioned<V>>> it) {
+                this.it = it;
+            }
+
+            @Override
+            public boolean hasNext() {
+                return it.hasNext();
+            }
+
+            @Override
+            public Map.Entry<K, V> next() {
+                Map.Entry<K, Versioned<V>> e = it.next();
+                return new AbstractMap.SimpleEntry<>(e.getKey(), e.getValue().value());
+            }
+        }
+
+    }
+
+    public static TestConsistentMultimap.Builder builder() {
+        return new TestConsistentMultimap.Builder();
+    }
+
+    public static class Builder<K, V> extends ConsistentMultimapBuilder<K, V> {
+
+        @Override
+        public AsyncConsistentMultimap<K, V> buildMultimap() {
+            return null;
+        }
+
+        @Override
+        public ConsistentMultimap<K, V> build() {
+            return new TestConsistentMultimap<K, V>();
+        }
+    }
 }