diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/BatchOperationResult.java b/core/api/src/main/java/org/onlab/onos/net/flow/BatchOperationResult.java
index 43fd694..33f1845 100644
--- a/core/api/src/main/java/org/onlab/onos/net/flow/BatchOperationResult.java
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/BatchOperationResult.java
@@ -1,6 +1,6 @@
 package org.onlab.onos.net.flow;
 
-import java.util.List;
+import java.util.Set;
 
 /**
  * Interface capturing the result of a batch operation.
@@ -15,9 +15,9 @@
     boolean isSuccess();
 
     /**
-     * Obtains a list of items which failed.
-     * @return a list of failures
+     * Obtains a set of items which failed.
+     * @return a set of failures
      */
-    List<T> failedItems();
+    Set<T> failedItems();
 
 }
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/CompletedBatchOperation.java b/core/api/src/main/java/org/onlab/onos/net/flow/CompletedBatchOperation.java
index e9889cd..4e671e3 100644
--- a/core/api/src/main/java/org/onlab/onos/net/flow/CompletedBatchOperation.java
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/CompletedBatchOperation.java
@@ -1,18 +1,18 @@
 package org.onlab.onos.net.flow;
 
-import java.util.List;
+import java.util.Set;
 
-import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
 
 public class CompletedBatchOperation implements BatchOperationResult<FlowEntry> {
 
 
     private final boolean success;
-    private final List<FlowEntry> failures;
+    private final Set<FlowEntry> failures;
 
-    public CompletedBatchOperation(boolean success, List<FlowEntry> failures) {
+    public CompletedBatchOperation(boolean success, Set<FlowEntry> failures) {
         this.success = success;
-        this.failures = ImmutableList.copyOf(failures);
+        this.failures = ImmutableSet.copyOf(failures);
     }
 
     @Override
@@ -21,7 +21,7 @@
     }
 
     @Override
-    public List<FlowEntry> failedItems() {
+    public Set<FlowEntry> failedItems() {
         return failures;
     }
 
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleBatchEvent.java b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleBatchEvent.java
new file mode 100644
index 0000000..4ba3366
--- /dev/null
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleBatchEvent.java
@@ -0,0 +1,67 @@
+package org.onlab.onos.net.flow;
+
+import org.onlab.onos.event.AbstractEvent;
+
+/**
+ * Describes flow rule batch event.
+ */
+public final class FlowRuleBatchEvent extends AbstractEvent<FlowRuleBatchEvent.Type, FlowRuleBatchRequest> {
+
+    /**
+     * Type of flow rule events.
+     */
+    public enum Type {
+
+        /**
+         * Signifies that a batch operation has been initiated.
+         */
+        BATCH_OPERATION_REQUESTED,
+
+        /**
+         * Signifies that a batch operation has completed.
+         */
+        BATCH_OPERATION_COMPLETED,
+    }
+
+    private final CompletedBatchOperation result;
+
+    /**
+     * Constructs a new FlowRuleBatchEvent.
+     * @param request batch operation request.
+     * @return event.
+     */
+    public static FlowRuleBatchEvent create(FlowRuleBatchRequest request) {
+        FlowRuleBatchEvent event = new FlowRuleBatchEvent(Type.BATCH_OPERATION_REQUESTED, request, null);
+        return event;
+    }
+
+    /**
+     * Constructs a new FlowRuleBatchEvent.
+     * @param request batch operation request.
+     * @param result completed batch operation result.
+     * @return event.
+     */
+    public static FlowRuleBatchEvent create(FlowRuleBatchRequest request, CompletedBatchOperation result) {
+        FlowRuleBatchEvent event = new FlowRuleBatchEvent(Type.BATCH_OPERATION_COMPLETED, request, result);
+        return event;
+    }
+
+    /**
+     * Returns the result of this batch operation.
+     * @return batch operation result.
+     */
+    public CompletedBatchOperation result() {
+        return result;
+    }
+
+    /**
+     * Creates an event of a given type and for the specified flow rule batch.
+     *
+     * @param type    flow rule batch event type
+     * @param batch    event flow rule batch subject
+     */
+    private FlowRuleBatchEvent(Type type, FlowRuleBatchRequest request, CompletedBatchOperation result) {
+        super(type, request);
+        this.result = result;
+    }
+}
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleBatchRequest.java b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleBatchRequest.java
new file mode 100644
index 0000000..0414fcb
--- /dev/null
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleBatchRequest.java
@@ -0,0 +1,38 @@
+package org.onlab.onos.net.flow;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.onlab.onos.net.flow.FlowRuleBatchEntry.FlowRuleOperation;
+
+import com.google.common.collect.Lists;
+
+public class FlowRuleBatchRequest {
+
+    private final List<FlowEntry> toAdd;
+    private final List<FlowEntry> toRemove;
+
+    public FlowRuleBatchRequest(List<FlowEntry> toAdd, List<FlowEntry> toRemove) {
+        this.toAdd = Collections.unmodifiableList(toAdd);
+        this.toRemove = Collections.unmodifiableList(toRemove);
+    }
+
+    public List<FlowEntry> toAdd() {
+        return toAdd;
+    }
+
+    public List<FlowEntry> toRemove() {
+        return toRemove;
+    }
+
+    public FlowRuleBatchOperation asBatchOperation() {
+        List<FlowRuleBatchEntry> entries = Lists.newArrayList();
+        for (FlowEntry e : toAdd) {
+            entries.add(new FlowRuleBatchEntry(FlowRuleOperation.ADD, e));
+        }
+        for (FlowEntry e : toRemove) {
+            entries.add(new FlowRuleBatchEntry(FlowRuleOperation.REMOVE, e));
+        }
+        return new FlowRuleBatchOperation(entries);
+    }
+}
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleProvider.java b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleProvider.java
index 3592e39..5a57b88 100644
--- a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleProvider.java
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleProvider.java
@@ -1,11 +1,11 @@
 package org.onlab.onos.net.flow;
 
-import java.util.concurrent.Future;
-
 import org.onlab.onos.ApplicationId;
 import org.onlab.onos.net.intent.BatchOperation;
 import org.onlab.onos.net.provider.Provider;
 
+import com.google.common.util.concurrent.ListenableFuture;
+
 /**
  * Abstraction of a flow rule provider.
  */
@@ -43,6 +43,6 @@
      * @param batch a batch of flow rules
      * @return a future indicating the status of this execution
      */
-    Future<CompletedBatchOperation> executeBatch(BatchOperation<FlowRuleBatchEntry> batch);
+    ListenableFuture<CompletedBatchOperation> executeBatch(BatchOperation<FlowRuleBatchEntry> batch);
 
 }
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStore.java b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStore.java
index abb9a10..c53a32d 100644
--- a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStore.java
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStore.java
@@ -1,5 +1,7 @@
 package org.onlab.onos.net.flow;
 
+import java.util.concurrent.Future;
+
 import org.onlab.onos.ApplicationId;
 import org.onlab.onos.net.DeviceId;
 import org.onlab.onos.store.Store;
@@ -7,7 +9,7 @@
 /**
  * Manages inventory of flow rules; not intended for direct use.
  */
-public interface FlowRuleStore extends Store<FlowRuleEvent, FlowRuleStoreDelegate> {
+public interface FlowRuleStore extends Store<FlowRuleBatchEvent, FlowRuleStoreDelegate> {
 
     /**
      * Returns the number of flow rule in the store.
@@ -41,12 +43,26 @@
     Iterable<FlowRule> getFlowRulesByAppId(ApplicationId appId);
 
     /**
+     // TODO: Better description of method behavior.
      * Stores a new flow rule without generating events.
      *
      * @param rule the flow rule to add
-     * @return true if the rule should be handled locally
      */
-    boolean storeFlowRule(FlowRule rule);
+    void storeFlowRule(FlowRule rule);
+
+    /**
+     * Stores a batch of flow rules.
+     * @param batchOperation batch of flow rules.
+     * @return Future response indicating success/failure of the batch operation
+     *     all the way down to the device.
+     */
+    Future<CompletedBatchOperation> storeBatch(FlowRuleBatchOperation batchOperation);
+
+    /**
+     * Invoked on the completion of a storeBatch operation.
+     * @param result
+     */
+    void batchOperationComplete(FlowRuleBatchEvent event);
 
     /**
      * Marks a flow rule for deletion. Actual deletion will occur
@@ -55,7 +71,7 @@
      * @param rule the flow rule to delete
      * @return true if the rule should be handled locally
      */
-    boolean deleteFlowRule(FlowRule rule);
+    void deleteFlowRule(FlowRule rule);
 
     /**
      * Stores a new flow rule, or updates an existing entry.
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStoreDelegate.java b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStoreDelegate.java
index 119712b..66973dd 100644
--- a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStoreDelegate.java
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStoreDelegate.java
@@ -5,5 +5,5 @@
 /**
  * Flow rule store delegate abstraction.
  */
-public interface FlowRuleStoreDelegate extends StoreDelegate<FlowRuleEvent> {
+public interface FlowRuleStoreDelegate extends StoreDelegate<FlowRuleBatchEvent> {
 }
diff --git a/core/net/src/main/java/org/onlab/onos/net/flow/impl/FlowRuleManager.java b/core/net/src/main/java/org/onlab/onos/net/flow/impl/FlowRuleManager.java
index 67e0867..ba37d22 100644
--- a/core/net/src/main/java/org/onlab/onos/net/flow/impl/FlowRuleManager.java
+++ b/core/net/src/main/java/org/onlab/onos/net/flow/impl/FlowRuleManager.java
@@ -5,8 +5,10 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -30,7 +32,9 @@
 import org.onlab.onos.net.flow.FlowRule;
 import org.onlab.onos.net.flow.FlowRuleBatchEntry;
 import org.onlab.onos.net.flow.FlowRuleBatchEntry.FlowRuleOperation;
+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.FlowRuleListener;
 import org.onlab.onos.net.flow.FlowRuleProvider;
@@ -47,6 +51,9 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Multimap;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
 
 /**
  * Provides implementation of the flow NB &amp; SB APIs.
@@ -104,11 +111,7 @@
     public void applyFlowRules(FlowRule... flowRules) {
         for (int i = 0; i < flowRules.length; i++) {
             FlowRule f = flowRules[i];
-            boolean local = store.storeFlowRule(f);
-            if (local) {
-                // TODO: aggregate all local rules and push down once?
-                applyFlowRulesToProviders(f);
-            }
+            store.storeFlowRule(f);
         }
     }
 
@@ -132,11 +135,7 @@
         FlowRule f;
         for (int i = 0; i < flowRules.length; i++) {
             f = flowRules[i];
-            boolean local = store.deleteFlowRule(f);
-            if (local) {
-                // TODO: aggregate all local rules and push down once?
-                removeFlowRulesFromProviders(f);
-            }
+            store.deleteFlowRule(f);
         }
     }
 
@@ -180,33 +179,21 @@
     @Override
     public Future<CompletedBatchOperation> applyBatch(
             FlowRuleBatchOperation batch) {
-        Multimap<FlowRuleProvider, FlowRuleBatchEntry> batches =
+        Multimap<DeviceId, FlowRuleBatchEntry> perDeviceBatches =
                 ArrayListMultimap.create();
         List<Future<CompletedBatchOperation>> futures = Lists.newArrayList();
         for (FlowRuleBatchEntry fbe : batch.getOperations()) {
             final FlowRule f = fbe.getTarget();
-            final Device device = deviceService.getDevice(f.deviceId());
-            final FlowRuleProvider frp = getProvider(device.providerId());
-            batches.put(frp, fbe);
-            switch (fbe.getOperator()) {
-                case ADD:
-                    store.storeFlowRule(f);
-                    break;
-                case REMOVE:
-                    store.deleteFlowRule(f);
-                    break;
-                case MODIFY:
-                default:
-                    log.error("Batch operation type {} unsupported.", fbe.getOperator());
-            }
+            perDeviceBatches.put(f.deviceId(), fbe);
         }
-        for (FlowRuleProvider provider : batches.keySet()) {
+
+        for (DeviceId deviceId : perDeviceBatches.keySet()) {
             FlowRuleBatchOperation b =
-                    new FlowRuleBatchOperation(batches.get(provider));
-            Future<CompletedBatchOperation> future = provider.executeBatch(b);
+                    new FlowRuleBatchOperation(perDeviceBatches.get(deviceId));
+            Future<CompletedBatchOperation> future = store.storeBatch(b);
             futures.add(future);
         }
-        return new FlowRuleBatchFuture(futures, batches);
+        return new FlowRuleBatchFuture(futures, perDeviceBatches);
     }
 
     @Override
@@ -318,6 +305,7 @@
                     post(event);
                 }
             } else {
+                log.info("Removing flow rules....");
                 removeFlowRules(flowEntry);
             }
 
@@ -385,21 +373,47 @@
 
     // Store delegate to re-post events emitted from the store.
     private class InternalStoreDelegate implements FlowRuleStoreDelegate {
+        // TODO: Right now we only dispatch events at individual flowEntry level.
+        // It may be more efficient for also dispatch events as a batch.
         @Override
-        public void notify(FlowRuleEvent event) {
+        public void notify(FlowRuleBatchEvent event) {
+            final FlowRuleBatchRequest request = event.subject();
             switch (event.type()) {
-            case RULE_ADD_REQUESTED:
-                applyFlowRulesToProviders(event.subject());
-                break;
-            case RULE_REMOVE_REQUESTED:
-                removeFlowRulesFromProviders(event.subject());
-                break;
+            case BATCH_OPERATION_REQUESTED:
+//                for (FlowEntry entry : request.toAdd()) {
+//                    //eventDispatcher.post(new FlowRuleEvent(FlowRuleEvent.Type.RULE_ADD_REQUESTED, entry));
+//                }
+//                for (FlowEntry entry : request.toRemove()) {
+//                    //eventDispatcher.post(new FlowRuleEvent(FlowRuleEvent.Type.RULE_REMOVE_REQUESTED, entry));
+//                }
+//                // FIXME: what about op.equals(FlowRuleOperation.MODIFY) ?
+//
+                FlowRuleBatchOperation batchOperation = request.asBatchOperation();
 
-            case RULE_ADDED:
-            case RULE_REMOVED:
-            case RULE_UPDATED:
-                // only dispatch events related to switch
-                eventDispatcher.post(event);
+                FlowRuleProvider flowRuleProvider =
+                        getProvider(batchOperation.getOperations().get(0).getTarget().deviceId());
+                final ListenableFuture<CompletedBatchOperation> result =
+                        flowRuleProvider.executeBatch(batchOperation);
+                result.addListener(new Runnable() {
+                    @Override
+                    public void run() {
+                        store.batchOperationComplete(FlowRuleBatchEvent.create(request, Futures.getUnchecked(result)));
+                    }
+                }, Executors.newCachedThreadPool());
+
+                break;
+            case BATCH_OPERATION_COMPLETED:
+                Set<FlowEntry> failedItems = event.result().failedItems();
+                for (FlowEntry entry : request.toAdd()) {
+                    if (!failedItems.contains(entry)) {
+                        eventDispatcher.post(new FlowRuleEvent(FlowRuleEvent.Type.RULE_ADDED, entry));
+                    }
+                }
+                for (FlowEntry entry : request.toRemove()) {
+                    if (!failedItems.contains(entry)) {
+                            eventDispatcher.post(new FlowRuleEvent(FlowRuleEvent.Type.RULE_REMOVED, entry));
+                    }
+                }
                 break;
             default:
                 break;
@@ -407,18 +421,15 @@
         }
     }
 
-    private class FlowRuleBatchFuture
-        implements Future<CompletedBatchOperation> {
+    private class FlowRuleBatchFuture implements Future<CompletedBatchOperation> {
 
         private final List<Future<CompletedBatchOperation>> futures;
-        private final Multimap<FlowRuleProvider, FlowRuleBatchEntry> batches;
+        private final Multimap<DeviceId, FlowRuleBatchEntry> batches;
         private final AtomicReference<BatchState> state;
         private CompletedBatchOperation overall;
 
-
-
         public FlowRuleBatchFuture(List<Future<CompletedBatchOperation>> futures,
-                Multimap<FlowRuleProvider, FlowRuleBatchEntry> batches) {
+                Multimap<DeviceId, FlowRuleBatchEntry> batches) {
             this.futures = futures;
             this.batches = batches;
             state = new AtomicReference<FlowRuleManager.BatchState>();
@@ -460,7 +471,7 @@
             }
 
             boolean success = true;
-            List<FlowEntry> failed = Lists.newLinkedList();
+            Set<FlowEntry> failed = Sets.newHashSet();
             CompletedBatchOperation completed;
             for (Future<CompletedBatchOperation> future : futures) {
                 completed = future.get();
@@ -480,7 +491,7 @@
                 return overall;
             }
             boolean success = true;
-            List<FlowEntry> failed = Lists.newLinkedList();
+            Set<FlowEntry> failed = Sets.newHashSet();
             CompletedBatchOperation completed;
             long start = System.nanoTime();
             long end = start + unit.toNanos(timeout);
@@ -494,7 +505,7 @@
             return finalizeBatchOperation(success, failed);
         }
 
-        private boolean validateBatchOperation(List<FlowEntry> failed,
+        private boolean validateBatchOperation(Set<FlowEntry> failed,
                 CompletedBatchOperation completed) {
 
             if (isCancelled()) {
@@ -516,7 +527,7 @@
         }
 
         private CompletedBatchOperation finalizeBatchOperation(boolean success,
-                List<FlowEntry> failed) {
+                Set<FlowEntry> failed) {
             synchronized (this) {
                 if (!state.compareAndSet(BatchState.STARTED, BatchState.FINISHED)) {
                     if (state.get() == BatchState.FINISHED) {
@@ -539,11 +550,6 @@
                     store.storeFlowRule(fbe.getTarget());
                 }
             }
-
         }
     }
-
-
-
-
 }
diff --git a/core/net/src/test/java/org/onlab/onos/net/flow/impl/FlowRuleManagerTest.java b/core/net/src/test/java/org/onlab/onos/net/flow/impl/FlowRuleManagerTest.java
index ca7cc07..59b2963 100644
--- a/core/net/src/test/java/org/onlab/onos/net/flow/impl/FlowRuleManagerTest.java
+++ b/core/net/src/test/java/org/onlab/onos/net/flow/impl/FlowRuleManagerTest.java
@@ -1,6 +1,5 @@
 package org.onlab.onos.net.flow.impl;
 
-import static java.util.Collections.EMPTY_LIST;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -17,6 +16,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -64,6 +64,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.ListenableFuture;
 
 /**
  * Test codifying the flow rule service & flow rule provider service contracts.
@@ -515,13 +516,13 @@
         }
 
         @Override
-        public Future<CompletedBatchOperation> executeBatch(
+        public ListenableFuture<CompletedBatchOperation> executeBatch(
                 BatchOperation<FlowRuleBatchEntry> batch) {
             return new TestInstallationFuture();
         }
 
         private class TestInstallationFuture
-                implements Future<CompletedBatchOperation> {
+                implements ListenableFuture<CompletedBatchOperation> {
 
             @Override
             public boolean cancel(boolean mayInterruptIfRunning) {
@@ -541,7 +542,7 @@
             @Override
             public CompletedBatchOperation get()
                     throws InterruptedException, ExecutionException {
-                return new CompletedBatchOperation(true, EMPTY_LIST);
+                return new CompletedBatchOperation(true, Collections.<FlowEntry>emptySet());
             }
 
             @Override
@@ -550,6 +551,11 @@
                     ExecutionException, TimeoutException {
                 return null;
             }
+
+            @Override
+            public void addListener(Runnable task, Executor executor) {
+                // TODO: add stuff.
+            }
         }
 
     }
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);
+    }
 }
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/FlowStoreMessageSubjects.java b/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/FlowStoreMessageSubjects.java
index a43dad6..ca833b8 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/FlowStoreMessageSubjects.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/FlowStoreMessageSubjects.java
@@ -12,4 +12,5 @@
     public static final MessageSubject ADD_OR_UPDATE_FLOW_RULE =
         new MessageSubject("peer-forward-add-or-update-flow-rule");
     public static final MessageSubject REMOVE_FLOW_RULE = new MessageSubject("peer-forward-remove-flow-rule");
+    public static final MessageSubject GET_FLOW_ENTRY = new MessageSubject("peer-forward-get-flow-entry");
 }
diff --git a/core/store/serializers/src/main/java/org/onlab/onos/store/serializers/KryoNamespaces.java b/core/store/serializers/src/main/java/org/onlab/onos/store/serializers/KryoNamespaces.java
index 9de6d8c..e6259d8 100644
--- a/core/store/serializers/src/main/java/org/onlab/onos/store/serializers/KryoNamespaces.java
+++ b/core/store/serializers/src/main/java/org/onlab/onos/store/serializers/KryoNamespaces.java
@@ -26,10 +26,12 @@
 import org.onlab.onos.net.PortNumber;
 import org.onlab.onos.net.device.DefaultDeviceDescription;
 import org.onlab.onos.net.device.DefaultPortDescription;
+import org.onlab.onos.net.flow.DefaultFlowEntry;
 import org.onlab.onos.net.flow.DefaultFlowRule;
 import org.onlab.onos.net.flow.DefaultTrafficSelector;
 import org.onlab.onos.net.flow.DefaultTrafficTreatment;
 import org.onlab.onos.net.flow.FlowId;
+import org.onlab.onos.net.flow.StoredFlowEntry;
 import org.onlab.onos.net.flow.criteria.Criteria;
 import org.onlab.onos.net.flow.criteria.Criterion;
 import org.onlab.onos.net.flow.instructions.Instructions;
@@ -93,6 +95,8 @@
                     HostId.class,
                     HostDescription.class,
                     DefaultHostDescription.class,
+                    DefaultFlowEntry.class,
+                    StoredFlowEntry.class,
                     DefaultFlowRule.class,
                     FlowId.class,
                     DefaultTrafficSelector.class,
diff --git a/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleFlowRuleStore.java b/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleFlowRuleStore.java
index a96cacb..3d10d3d 100644
--- a/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleFlowRuleStore.java
+++ b/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleFlowRuleStore.java
@@ -3,6 +3,8 @@
 import static org.onlab.onos.net.flow.FlowRuleEvent.Type.RULE_REMOVED;
 import static org.slf4j.LoggerFactory.getLogger;
 import static org.apache.commons.lang3.concurrent.ConcurrentUtils.createIfAbsentUnchecked;
+
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
@@ -10,6 +12,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Future;
 
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
@@ -17,11 +20,17 @@
 import org.apache.felix.scr.annotations.Service;
 import org.onlab.onos.ApplicationId;
 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.FlowId;
 import org.onlab.onos.net.flow.FlowRule;
+import org.onlab.onos.net.flow.FlowRuleBatchEntry;
+import org.onlab.onos.net.flow.FlowRuleBatchEntry.FlowRuleOperation;
+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.FlowRuleEvent.Type;
 import org.onlab.onos.net.flow.FlowRuleStore;
@@ -33,6 +42,7 @@
 
 import com.google.common.base.Function;
 import com.google.common.collect.FluentIterable;
+import com.google.common.util.concurrent.Futures;
 
 /**
  * Manages inventory of flow rules using trivial in-memory implementation.
@@ -40,7 +50,7 @@
 @Component(immediate = true)
 @Service
 public class SimpleFlowRuleStore
-        extends AbstractStore<FlowRuleEvent, FlowRuleStoreDelegate>
+        extends AbstractStore<FlowRuleBatchEvent, FlowRuleStoreDelegate>
         implements FlowRuleStore {
 
     private final Logger log = getLogger(getClass());
@@ -148,12 +158,11 @@
     }
 
     @Override
-    public boolean storeFlowRule(FlowRule rule) {
-        final boolean added = storeFlowRuleInternal(rule);
-        return added;
+    public void storeFlowRule(FlowRule rule) {
+        storeFlowRuleInternal(rule);
     }
 
-    private boolean storeFlowRuleInternal(FlowRule rule) {
+    private void storeFlowRuleInternal(FlowRule rule) {
         StoredFlowEntry f = new DefaultFlowEntry(rule);
         final DeviceId did = f.deviceId();
         final FlowId fid = f.id();
@@ -162,19 +171,20 @@
             for (StoredFlowEntry fe : existing) {
                 if (fe.equals(rule)) {
                     // was already there? ignore
-                    return false;
+                    return;
                 }
             }
             // new flow rule added
             existing.add(f);
-            // TODO: Should we notify only if it's "remote" event?
-            //notifyDelegate(new FlowRuleEvent(Type.RULE_ADD_REQUESTED, rule));
-            return true;
+            notifyDelegate(FlowRuleBatchEvent.create(
+                    new FlowRuleBatchRequest(
+                            Arrays.<FlowEntry>asList(f),
+                            Collections.<FlowEntry>emptyList())));
         }
     }
 
     @Override
-    public boolean deleteFlowRule(FlowRule rule) {
+    public void deleteFlowRule(FlowRule rule) {
 
         List<StoredFlowEntry> entries = getFlowEntries(rule.deviceId(), rule.id());
         synchronized (entries) {
@@ -184,13 +194,11 @@
                         entry.setState(FlowEntryState.PENDING_REMOVE);
                         // TODO: Should we notify only if it's "remote" event?
                         //notifyDelegate(new FlowRuleEvent(Type.RULE_REMOVE_REQUESTED, rule));
-                        return true;
                     }
                 }
             }
         }
         //log.warn("Cannot find rule {}", rule);
-        return false;
     }
 
     @Override
@@ -236,4 +244,24 @@
         }
         return null;
     }
+
+    @Override
+    public Future<CompletedBatchOperation> storeBatch(
+            FlowRuleBatchOperation batchOperation) {
+        for (FlowRuleBatchEntry entry : batchOperation.getOperations()) {
+            if (entry.getOperator().equals(FlowRuleOperation.ADD)) {
+                storeFlowRule(entry.getTarget());
+            } else if (entry.getOperator().equals(FlowRuleOperation.REMOVE)) {
+                deleteFlowRule(entry.getTarget());
+            } else {
+                throw new UnsupportedOperationException("Unsupported operation type");
+            }
+        }
+        return Futures.immediateFuture(new CompletedBatchOperation(true, Collections.<FlowEntry>emptySet()));
+    }
+
+    @Override
+    public void batchOperationComplete(FlowRuleBatchEvent event) {
+        notifyDelegate(event);
+    }
 }
diff --git a/providers/openflow/flow/src/main/java/org/onlab/onos/provider/of/flow/impl/OpenFlowRuleProvider.java b/providers/openflow/flow/src/main/java/org/onlab/onos/provider/of/flow/impl/OpenFlowRuleProvider.java
index 54265ba..b7d84f0 100644
--- a/providers/openflow/flow/src/main/java/org/onlab/onos/provider/of/flow/impl/OpenFlowRuleProvider.java
+++ b/providers/openflow/flow/src/main/java/org/onlab/onos/provider/of/flow/impl/OpenFlowRuleProvider.java
@@ -10,7 +10,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -69,9 +69,11 @@
 import org.slf4j.Logger;
 
 import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Multimap;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.ExecutionList;
+import com.google.common.util.concurrent.ListenableFuture;
 
 /**
  * Provider which uses an OpenFlow controller to detect network
@@ -97,6 +99,8 @@
 
     private final InternalFlowProvider listener = new InternalFlowProvider();
 
+    // FIXME: This should be an expiring map to ensure futures that don't have
+    // a future eventually get garbage collected.
     private final Map<Long, InstallationFuture> pendingFutures =
             new ConcurrentHashMap<Long, InstallationFuture>();
 
@@ -159,7 +163,7 @@
     }
 
     @Override
-    public Future<CompletedBatchOperation> executeBatch(BatchOperation<FlowRuleBatchEntry> batch) {
+    public ListenableFuture<CompletedBatchOperation> executeBatch(BatchOperation<FlowRuleBatchEntry> batch) {
         final Set<Dpid> sws =
                 Collections.newSetFromMap(new ConcurrentHashMap<Dpid, Boolean>());
         final Map<Long, FlowRuleBatchEntry> fmXids = new HashMap<Long, FlowRuleBatchEntry>();
@@ -315,18 +319,20 @@
         }
     }
 
-    private class InstallationFuture implements Future<CompletedBatchOperation> {
+    private class InstallationFuture implements ListenableFuture<CompletedBatchOperation> {
 
         private final Set<Dpid> sws;
         private final AtomicBoolean ok = new AtomicBoolean(true);
         private final Map<Long, FlowRuleBatchEntry> fms;
 
-        private final List<FlowEntry> offendingFlowMods = Lists.newLinkedList();
+        private final Set<FlowEntry> offendingFlowMods = Sets.newHashSet();
 
         private final CountDownLatch countDownLatch;
         private Long pendingXid;
         private BatchState state;
 
+        private final ExecutionList executionList = new ExecutionList();
+
         public InstallationFuture(Set<Dpid> sws, Map<Long, FlowRuleBatchEntry> fmXids) {
             this.state = BatchState.STARTED;
             this.sws = sws;
@@ -335,6 +341,7 @@
         }
 
         public void fail(OFErrorMsg msg, Dpid dpid) {
+
             ok.set(false);
             removeRequirement(dpid);
             FlowEntry fe = null;
@@ -407,6 +414,9 @@
 
         @Override
         public boolean cancel(boolean mayInterruptIfRunning) {
+            if (isDone()) {
+                return false;
+            }
             ok.set(false);
             this.state = BatchState.CANCELLED;
             cleanUp();
@@ -419,7 +429,8 @@
                 }
 
             }
-            return isCancelled();
+            invokeCallbacks();
+            return true;
         }
 
         @Override
@@ -429,14 +440,15 @@
 
         @Override
         public boolean isDone() {
-            return this.state == BatchState.FINISHED;
+            return this.state == BatchState.FINISHED || isCancelled();
         }
 
         @Override
         public CompletedBatchOperation get() throws InterruptedException, ExecutionException {
             countDownLatch.await();
             this.state = BatchState.FINISHED;
-            return new CompletedBatchOperation(ok.get(), offendingFlowMods);
+            CompletedBatchOperation result = new CompletedBatchOperation(ok.get(), offendingFlowMods);
+            return result;
         }
 
         @Override
@@ -445,7 +457,8 @@
                 TimeoutException {
             if (countDownLatch.await(timeout, unit)) {
                 this.state = BatchState.FINISHED;
-                return new CompletedBatchOperation(ok.get(), offendingFlowMods);
+                CompletedBatchOperation result = new CompletedBatchOperation(ok.get(), offendingFlowMods);
+                return result;
             }
             throw new TimeoutException();
         }
@@ -463,10 +476,21 @@
 
         private void removeRequirement(Dpid dpid) {
             countDownLatch.countDown();
+            if (countDownLatch.getCount() == 0) {
+                invokeCallbacks();
+            }
             sws.remove(dpid);
             cleanUp();
         }
 
+        @Override
+        public void addListener(Runnable runnable, Executor executor) {
+            executionList.add(runnable, executor);
+        }
+
+        private void invokeCallbacks() {
+            executionList.execute();
+        }
     }
 
 }