diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/DistributedFlowRuleStore.java b/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/DistributedFlowRuleStore.java
index bde57c6..85f928a 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/DistributedFlowRuleStore.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/DistributedFlowRuleStore.java
@@ -5,10 +5,14 @@
 import static org.onlab.onos.store.flow.impl.FlowStoreMessageSubjects.*;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.List;
 
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
@@ -19,11 +23,17 @@
 import org.onlab.onos.ApplicationId;
 import org.onlab.onos.cluster.ClusterService;
 import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.flow.CompletedBatchOperation;
 import org.onlab.onos.net.flow.DefaultFlowEntry;
 import org.onlab.onos.net.flow.FlowEntry;
 import org.onlab.onos.net.flow.FlowEntry.FlowEntryState;
 import org.onlab.onos.net.flow.FlowRule;
+import org.onlab.onos.net.flow.FlowRuleBatchEntry;
+import org.onlab.onos.net.flow.FlowRuleBatchEvent;
+import org.onlab.onos.net.flow.FlowRuleBatchOperation;
+import org.onlab.onos.net.flow.FlowRuleBatchRequest;
 import org.onlab.onos.net.flow.FlowRuleEvent;
+import org.onlab.onos.net.flow.FlowRuleBatchEntry.FlowRuleOperation;
 import org.onlab.onos.net.flow.FlowRuleEvent.Type;
 import org.onlab.onos.net.flow.FlowRuleStore;
 import org.onlab.onos.net.flow.FlowRuleStoreDelegate;
@@ -43,6 +53,7 @@
 import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Multimap;
+import com.google.common.util.concurrent.Futures;
 
 /**
  * Manages inventory of flow rules using a distributed state management protocol.
@@ -50,7 +61,7 @@
 @Component(immediate = true)
 @Service
 public class DistributedFlowRuleStore
-        extends AbstractStore<FlowRuleEvent, FlowRuleStoreDelegate>
+        extends AbstractStore<FlowRuleBatchEvent, FlowRuleStoreDelegate>
         implements FlowRuleStore {
 
     private final Logger log = getLogger(getClass());
@@ -92,7 +103,7 @@
             public void handle(ClusterMessage message) {
                 FlowRule rule = SERIALIZER.decode(message.payload());
                 log.info("received add request for {}", rule);
-                storeFlowEntryInternal(rule);
+                storeFlowRule(rule);
                 // FIXME what to respond.
                 try {
                     message.respond(SERIALIZER.encode("ACK"));
@@ -108,7 +119,7 @@
             public void handle(ClusterMessage message) {
                 FlowRule rule = SERIALIZER.decode(message.payload());
                 log.info("received delete request for {}", rule);
-                deleteFlowRuleInternal(rule);
+                deleteFlowRule(rule);
                 // FIXME what to respond.
                 try {
                     message.respond(SERIALIZER.encode("ACK"));
@@ -118,6 +129,22 @@
 
             }
         });
+
+        clusterCommunicator.addSubscriber(GET_FLOW_ENTRY, new ClusterMessageHandler() {
+
+            @Override
+            public void handle(ClusterMessage message) {
+                FlowRule rule = SERIALIZER.decode(message.payload());
+                log.info("received get flow entry request for {}", rule);
+                FlowEntry flowEntry = getFlowEntryInternal(rule);
+                try {
+                    message.respond(SERIALIZER.encode(flowEntry));
+                } catch (IOException e) {
+                    log.error("Failed to respond back", e);
+                }
+            }
+        });
+
         log.info("Started");
     }
 
@@ -127,6 +154,9 @@
     }
 
 
+    // TODO: This is not a efficient operation on a distributed sharded
+    // flow store. We need to revisit the need for this operation or at least
+    // make it device specific.
     @Override
     public int getFlowRuleCount() {
         return flowEntries.size();
@@ -134,7 +164,26 @@
 
     @Override
     public synchronized FlowEntry getFlowEntry(FlowRule rule) {
-        return getFlowEntryInternal(rule);
+        ReplicaInfo replicaInfo = replicaInfoManager.getReplicaInfoFor(rule.deviceId());
+        if (replicaInfo.master().get().equals(clusterService.getLocalNode().id())) {
+            return getFlowEntryInternal(rule);
+        }
+
+        log.info("Forwarding getFlowEntry to {}, which is the primary (master) for device {}",
+                replicaInfo.master().orNull(), rule.deviceId());
+
+        ClusterMessage message = new ClusterMessage(
+                clusterService.getLocalNode().id(),
+                FlowStoreMessageSubjects.GET_FLOW_ENTRY,
+                SERIALIZER.encode(rule));
+
+        try {
+            ClusterMessageResponse response = clusterCommunicator.sendAndReceive(message, replicaInfo.master().get());
+            return SERIALIZER.decode(response.get(FLOW_RULE_STORE_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS));
+        } catch (IOException | TimeoutException e) {
+            // FIXME: throw a FlowStoreException
+            throw new RuntimeException(e);
+        }
     }
 
     private synchronized StoredFlowEntry getFlowEntryInternal(FlowRule rule) {
@@ -165,19 +214,30 @@
     }
 
     @Override
-    public boolean storeFlowRule(FlowRule rule) {
-        ReplicaInfo replicaInfo = replicaInfoManager.getReplicaInfoFor(rule.deviceId());
-        if (replicaInfo.master().get().equals(clusterService.getLocalNode().id())) {
-            return storeFlowEntryInternal(rule);
+    public void storeFlowRule(FlowRule rule) {
+        storeBatch(new FlowRuleBatchOperation(Arrays.asList(new FlowRuleBatchEntry(FlowRuleOperation.ADD, rule))));
+    }
+
+    public Future<CompletedBatchOperation> storeBatch(FlowRuleBatchOperation operation) {
+        if (operation.getOperations().isEmpty()) {
+            return Futures.immediateFuture(new CompletedBatchOperation(true, Collections.<FlowEntry>emptySet()));
         }
 
-        log.info("Forwarding storeFlowRule to {}, which is the primary (master) for device {}",
-                replicaInfo.master().orNull(), rule.deviceId());
+        DeviceId deviceId = operation.getOperations().get(0).getTarget().deviceId();
+
+        ReplicaInfo replicaInfo = replicaInfoManager.getReplicaInfoFor(deviceId);
+
+        if (replicaInfo.master().get().equals(clusterService.getLocalNode().id())) {
+            return storeBatchInternal(operation);
+        }
+
+        log.info("Forwarding storeBatch to {}, which is the primary (master) for device {}",
+                replicaInfo.master().orNull(), deviceId);
 
         ClusterMessage message = new ClusterMessage(
                 clusterService.getLocalNode().id(),
                 FlowStoreMessageSubjects.STORE_FLOW_RULE,
-                SERIALIZER.encode(rule));
+                SERIALIZER.encode(operation));
 
         try {
             ClusterMessageResponse response = clusterCommunicator.sendAndReceive(message, replicaInfo.master().get());
@@ -186,58 +246,44 @@
             // FIXME: throw a FlowStoreException
             throw new RuntimeException(e);
         }
-        return false;
+
+        return null;
     }
 
-    private synchronized boolean storeFlowEntryInternal(FlowRule flowRule) {
-        StoredFlowEntry flowEntry = new DefaultFlowEntry(flowRule);
-        DeviceId deviceId = flowRule.deviceId();
-        // write to local copy.
-        if (!flowEntries.containsEntry(deviceId, flowEntry)) {
-            flowEntries.put(deviceId, flowEntry);
-            flowEntriesById.put(flowRule.appId(), flowEntry);
-            notifyDelegate(new FlowRuleEvent(Type.RULE_ADD_REQUESTED, flowRule));
-            return true;
+    private Future<CompletedBatchOperation> storeBatchInternal(FlowRuleBatchOperation operation) {
+        List<FlowEntry> toRemove = new ArrayList<>();
+        List<FlowEntry> toAdd = new ArrayList<>();
+        // TODO: backup changes to hazelcast map
+        for (FlowRuleBatchEntry batchEntry : operation.getOperations()) {
+            FlowRule flowRule = batchEntry.getTarget();
+            FlowRuleOperation op = batchEntry.getOperator();
+            if (op.equals(FlowRuleOperation.REMOVE)) {
+                StoredFlowEntry entry = getFlowEntryInternal(flowRule);
+                if (entry != null) {
+                    entry.setState(FlowEntryState.PENDING_REMOVE);
+                }
+                toRemove.add(entry);
+            } else if (op.equals(FlowRuleOperation.ADD)) {
+                StoredFlowEntry flowEntry = new DefaultFlowEntry(flowRule);
+                DeviceId deviceId = flowRule.deviceId();
+                if (!flowEntries.containsEntry(deviceId, flowEntry)) {
+                    flowEntries.put(deviceId, flowEntry);
+                    flowEntriesById.put(flowRule.appId(), flowEntry);
+                    toAdd.add(flowEntry);
+                }
+            }
         }
-        // write to backup.
-        // TODO: write to a hazelcast map.
-        return false;
+        if (toAdd.isEmpty() && toRemove.isEmpty()) {
+            return Futures.immediateFuture(new CompletedBatchOperation(true, Collections.<FlowEntry>emptySet()));
+        }
+        notifyDelegate(FlowRuleBatchEvent.create(new FlowRuleBatchRequest(toAdd, toRemove)));
+        // TODO: imlpement this.
+        return Futures.immediateFailedFuture(new RuntimeException("Implement this."));
     }
 
     @Override
-    public synchronized boolean deleteFlowRule(FlowRule rule) {
-        ReplicaInfo replicaInfo = replicaInfoManager.getReplicaInfoFor(rule.deviceId());
-        if (replicaInfo.master().get().equals(clusterService.getLocalNode().id())) {
-            return deleteFlowRuleInternal(rule);
-        }
-
-        ClusterMessage message = new ClusterMessage(
-                clusterService.getLocalNode().id(),
-                FlowStoreMessageSubjects.DELETE_FLOW_RULE,
-                SERIALIZER.encode(rule));
-
-        try {
-            ClusterMessageResponse response = clusterCommunicator.sendAndReceive(message, replicaInfo.master().get());
-            response.get(FLOW_RULE_STORE_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
-        } catch (IOException | TimeoutException e) {
-            // FIXME: throw a FlowStoreException
-            throw new RuntimeException(e);
-        }
-        return false;
-    }
-
-    private synchronized boolean deleteFlowRuleInternal(FlowRule flowRule) {
-        StoredFlowEntry entry = getFlowEntryInternal(flowRule);
-        if (entry == null) {
-            return false;
-        }
-        entry.setState(FlowEntryState.PENDING_REMOVE);
-
-        // TODO: also update backup.
-
-        notifyDelegate(new FlowRuleEvent(Type.RULE_REMOVE_REQUESTED, flowRule));
-
-        return true;
+    public void deleteFlowRule(FlowRule rule) {
+        storeBatch(new FlowRuleBatchOperation(Arrays.asList(new FlowRuleBatchEntry(FlowRuleOperation.REMOVE, rule))));
     }
 
     @Override
@@ -315,4 +361,9 @@
         }
         // TODO: also update backup.
     }
+
+    @Override
+    public void batchOperationComplete(FlowRuleBatchEvent event) {
+        notifyDelegate(event);
+    }
 }