diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseManager.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseManager.java
index 010276a..363f330 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseManager.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseManager.java
@@ -34,6 +34,7 @@
 import net.kuujo.copycat.protocol.Protocol;
 import net.kuujo.copycat.util.concurrent.NamedThreadFactory;
 
+import org.apache.commons.lang.math.RandomUtils;
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
@@ -41,6 +42,7 @@
 import org.apache.felix.scr.annotations.ReferenceCardinality;
 import org.apache.felix.scr.annotations.Service;
 import org.onosproject.cluster.ClusterService;
+import org.onosproject.core.IdGenerator;
 import org.onosproject.store.cluster.impl.DistributedClusterStore;
 import org.onosproject.store.cluster.impl.NodeInfo;
 import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
@@ -53,11 +55,13 @@
 import org.onosproject.store.service.SetBuilder;
 import org.onosproject.store.service.StorageAdminService;
 import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.Transaction;
 import org.onosproject.store.service.TransactionContext;
 import org.slf4j.Logger;
 
 import java.io.File;
 import java.io.IOException;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -92,6 +96,9 @@
     private PartitionedDatabase partitionedDatabase;
     private Database inMemoryDatabase;
 
+    private TransactionManager transactionManager;
+    private final IdGenerator transactionIdGenerator = () -> RandomUtils.nextLong();
+
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected ClusterService clusterService;
 
@@ -188,6 +195,7 @@
             Thread.currentThread().interrupt();
             log.warn("Failed to complete database initialization.");
         }
+        transactionManager = new TransactionManager(partitionedDatabase);
         log.info("Started");
     }
 
@@ -218,7 +226,7 @@
 
     @Override
     public TransactionContext createTransactionContext() {
-        return new DefaultTransactionContext(partitionedDatabase);
+        return new DefaultTransactionContext(partitionedDatabase, transactionIdGenerator.getNewId());
     }
 
     @Override
@@ -331,6 +339,11 @@
             .collect(Collectors.toList());
     }
 
+    @Override
+    public Collection<Transaction> getTransactions() {
+        return complete(transactionManager.getTransactions());
+    }
+
     private static <T> T complete(CompletableFuture<T> future) {
         try {
             return future.get(DATABASE_OPERATION_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
@@ -343,4 +356,9 @@
             throw new ConsistentMapException(e.getCause());
         }
     }
+
+    @Override
+    public void redriveTransactions() {
+        getTransactions().stream().forEach(transactionManager::execute);
+    }
 }
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseProxy.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseProxy.java
index 3ea06fb..3578525 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseProxy.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseProxy.java
@@ -17,12 +17,11 @@
 package org.onosproject.store.consistent.impl;
 
 import java.util.Collection;
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 
-import org.onosproject.store.service.UpdateOperation;
+import org.onosproject.store.service.Transaction;
 import org.onosproject.store.service.Versioned;
 
 /**
@@ -87,7 +86,7 @@
    * @param value The value to set.
    * @return A completable future to be completed with the result once complete.
    */
-  CompletableFuture<Versioned<V>> put(String tableName, K key, V value);
+  CompletableFuture<Result<Versioned<V>>> put(String tableName, K key, V value);
 
   /**
    * Removes a value from the table.
@@ -96,7 +95,7 @@
    * @param key The key to remove.
    * @return A completable future to be completed with the result once complete.
    */
-  CompletableFuture<Versioned<V>> remove(String tableName, K key);
+  CompletableFuture<Result<Versioned<V>>> remove(String tableName, K key);
 
   /**
    * Clears the table.
@@ -104,7 +103,7 @@
    * @param tableName table name
    * @return A completable future to be completed with the result once complete.
    */
-  CompletableFuture<Void> clear(String tableName);
+  CompletableFuture<Result<Void>> clear(String tableName);
 
   /**
    * Gets a set of keys in the table.
@@ -138,7 +137,7 @@
    * @param value The value to set if the given key does not exist.
    * @return A completable future to be completed with the result once complete.
    */
-  CompletableFuture<Versioned<V>> putIfAbsent(String tableName, K key, V value);
+  CompletableFuture<Result<Versioned<V>>> putIfAbsent(String tableName, K key, V value);
 
   /**
    * Removes a key and if the existing value for that key matches the specified value.
@@ -148,7 +147,7 @@
    * @param value The value to remove.
    * @return A completable future to be completed with the result once complete.
    */
-  CompletableFuture<Boolean> remove(String tableName, K key, V value);
+  CompletableFuture<Result<Boolean>> remove(String tableName, K key, V value);
 
   /**
    * Removes a key and if the existing version for that key matches the specified version.
@@ -158,7 +157,7 @@
    * @param version The expected version.
    * @return A completable future to be completed with the result once complete.
    */
-  CompletableFuture<Boolean> remove(String tableName, K key, long version);
+  CompletableFuture<Result<Boolean>> remove(String tableName, K key, long version);
 
   /**
    * Replaces the entry for the specified key only if currently mapped to the specified value.
@@ -169,7 +168,7 @@
    * @param newValue The value with which to replace the given key and value.
    * @return A completable future to be completed with the result once complete.
    */
-  CompletableFuture<Boolean> replace(String tableName, K key, V oldValue, V newValue);
+  CompletableFuture<Result<Boolean>> replace(String tableName, K key, V oldValue, V newValue);
 
   /**
    * Replaces the entry for the specified key only if currently mapped to the specified version.
@@ -180,14 +179,42 @@
    * @param newValue The value with which to replace the given key and version.
    * @return A completable future to be completed with the result once complete.
    */
-  CompletableFuture<Boolean> replace(String tableName, K key, long oldVersion, V newValue);
+  CompletableFuture<Result<Boolean>> replace(String tableName, K key, long oldVersion, V newValue);
 
   /**
-   * Perform a atomic batch update operation i.e. either all operations in batch succeed or
-   * none do and no state changes are made.
+   * Prepare and commit the specified transaction.
    *
-   * @param updates list of updates to apply atomically.
-   * @return A completable future to be completed with the result once complete.
+   * @param transaction transaction to commit (after preparation)
+   * @return A completable future to be completed with the result once complete
    */
-  CompletableFuture<Boolean> atomicBatchUpdate(List<UpdateOperation<K, V>> updates);
+  CompletableFuture<Boolean> prepareAndCommit(Transaction transaction);
+
+  /**
+   * Prepare the specified transaction for commit. A successful prepare implies
+   * all the affected resources are locked thus ensuring no concurrent updates can interfere.
+   *
+   * @param transaction transaction to prepare (for commit)
+   * @return A completable future to be completed with the result once complete. The future is completed
+   * with true if the transaction is successfully prepared i.e. all pre-conditions are met and
+   * applicable resources locked.
+   */
+  CompletableFuture<Boolean> prepare(Transaction transaction);
+
+  /**
+   * Commit the specified transaction. A successful commit implies
+   * all the updates are applied, are now durable and are now visible externally.
+   *
+   * @param transaction transaction to commit
+   * @return A completable future to be completed with the result once complete
+   */
+  CompletableFuture<Boolean> commit(Transaction transaction);
+
+  /**
+   * Rollback the specified transaction. A successful rollback implies
+   * all previously acquired locks for the affected resources are released.
+   *
+   * @param transaction transaction to rollback
+   * @return A completable future to be completed with the result once complete
+   */
+  CompletableFuture<Boolean> rollback(Transaction transaction);
 }
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseSerializer.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseSerializer.java
index 237c85b..f6fa6d1 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseSerializer.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseSerializer.java
@@ -23,6 +23,8 @@
 import org.onlab.util.KryoNamespace;
 import org.onosproject.store.serializers.KryoNamespaces;
 import org.onosproject.store.serializers.KryoSerializer;
+import org.onosproject.store.service.DatabaseUpdate;
+import org.onosproject.store.service.Transaction;
 import org.onosproject.store.service.Versioned;
 
 import net.kuujo.copycat.cluster.internal.MemberInfo;
@@ -63,8 +65,14 @@
     private static final KryoNamespace ONOS_STORE = KryoNamespace.newBuilder()
             .nextId(KryoNamespace.FLOATING_ID)
             .register(Versioned.class)
+            .register(DatabaseUpdate.class)
+            .register(DatabaseUpdate.Type.class)
             .register(Pair.class)
             .register(ImmutablePair.class)
+            .register(Result.class)
+            .register(Result.Status.class)
+            .register(DefaultTransaction.class)
+            .register(Transaction.State.class)
             .build();
 
     private static final KryoSerializer SERIALIZER = new KryoSerializer() {
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseState.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseState.java
index 097261d..6685dde 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseState.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseState.java
@@ -17,11 +17,10 @@
 package org.onosproject.store.consistent.impl;
 
 import java.util.Collection;
-import java.util.List;
 import java.util.Map.Entry;
 import java.util.Set;
 
-import org.onosproject.store.service.UpdateOperation;
+import org.onosproject.store.service.Transaction;
 import org.onosproject.store.service.Versioned;
 
 import net.kuujo.copycat.state.Command;
@@ -62,13 +61,13 @@
   Versioned<V> get(String tableName, K key);
 
   @Command
-  Versioned<V> put(String tableName, K key, V value);
+  Result<Versioned<V>> put(String tableName, K key, V value);
 
   @Command
-  Versioned<V> remove(String tableName, K key);
+  Result<Versioned<V>> remove(String tableName, K key);
 
   @Command
-  void clear(String tableName);
+  Result<Void> clear(String tableName);
 
   @Query
   Set<K> keySet(String tableName);
@@ -80,20 +79,29 @@
   Set<Entry<K, Versioned<V>>> entrySet(String tableName);
 
   @Command
-  Versioned<V> putIfAbsent(String tableName, K key, V value);
+  Result<Versioned<V>> putIfAbsent(String tableName, K key, V value);
 
   @Command
-  boolean remove(String tableName, K key, V value);
+  Result<Boolean> remove(String tableName, K key, V value);
 
   @Command
-  boolean remove(String tableName, K key, long version);
+  Result<Boolean> remove(String tableName, K key, long version);
 
   @Command
-  boolean replace(String tableName, K key, V oldValue, V newValue);
+  Result<Boolean> replace(String tableName, K key, V oldValue, V newValue);
 
   @Command
-  boolean replace(String tableName, K key, long oldVersion, V newValue);
+  Result<Boolean> replace(String tableName, K key, long oldVersion, V newValue);
 
   @Command
-  boolean batchUpdate(List<UpdateOperation<K, V>> updates);
+  boolean prepareAndCommit(Transaction transaction);
+
+  @Command
+  boolean prepare(Transaction transaction);
+
+  @Command
+  boolean commit(Transaction transaction);
+
+  @Command
+  boolean rollback(Transaction transaction);
 }
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAsyncConsistentMap.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAsyncConsistentMap.java
index c86a6ea..09f612e 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAsyncConsistentMap.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAsyncConsistentMap.java
@@ -28,6 +28,7 @@
 import org.apache.commons.lang3.tuple.Pair;
 import org.onlab.util.HexString;
 import org.onosproject.store.service.AsyncConsistentMap;
+import org.onosproject.store.service.ConsistentMapException;
 import org.onosproject.store.service.Serializer;
 import org.onosproject.store.service.Versioned;
 
@@ -108,6 +109,7 @@
         checkNotNull(key, ERROR_NULL_KEY);
         checkNotNull(value, ERROR_NULL_VALUE);
         return database.put(name, keyCache.getUnchecked(key), serializer.encode(value))
+                .thenApply(this::unwrapResult)
                 .thenApply(v -> v != null
                 ? new Versioned<>(serializer.decode(v.value()), v.version(), v.creationTime()) : null);
     }
@@ -116,13 +118,14 @@
     public CompletableFuture<Versioned<V>> remove(K key) {
         checkNotNull(key, ERROR_NULL_KEY);
         return database.remove(name, keyCache.getUnchecked(key))
+                .thenApply(this::unwrapResult)
                 .thenApply(v -> v != null
                 ? new Versioned<>(serializer.decode(v.value()), v.version(), v.creationTime()) : null);
     }
 
     @Override
     public CompletableFuture<Void> clear() {
-        return database.clear(name);
+        return database.clear(name).thenApply(this::unwrapResult);
     }
 
     @Override
@@ -154,23 +157,27 @@
     public CompletableFuture<Versioned<V>> putIfAbsent(K key, V value) {
         checkNotNull(key, ERROR_NULL_KEY);
         checkNotNull(value, ERROR_NULL_VALUE);
-        return database.putIfAbsent(
-                name, keyCache.getUnchecked(key), serializer.encode(value)).thenApply(v ->
-                v != null ?
-                new Versioned<>(serializer.decode(v.value()), v.version(), v.creationTime()) : null);
+        return database.putIfAbsent(name,
+                                    keyCache.getUnchecked(key),
+                                    serializer.encode(value))
+               .thenApply(this::unwrapResult)
+               .thenApply(v -> v != null ?
+                       new Versioned<>(serializer.decode(v.value()), v.version(), v.creationTime()) : null);
     }
 
     @Override
     public CompletableFuture<Boolean> remove(K key, V value) {
         checkNotNull(key, ERROR_NULL_KEY);
         checkNotNull(value, ERROR_NULL_VALUE);
-        return database.remove(name, keyCache.getUnchecked(key), serializer.encode(value));
+        return database.remove(name, keyCache.getUnchecked(key), serializer.encode(value))
+                .thenApply(this::unwrapResult);
     }
 
     @Override
     public CompletableFuture<Boolean> remove(K key, long version) {
         checkNotNull(key, ERROR_NULL_KEY);
-        return database.remove(name, keyCache.getUnchecked(key), version);
+        return database.remove(name, keyCache.getUnchecked(key), version)
+                .thenApply(this::unwrapResult);
 
     }
 
@@ -179,14 +186,16 @@
         checkNotNull(key, ERROR_NULL_KEY);
         checkNotNull(newValue, ERROR_NULL_VALUE);
         byte[] existing = oldValue != null ? serializer.encode(oldValue) : null;
-        return database.replace(name, keyCache.getUnchecked(key), existing, serializer.encode(newValue));
+        return database.replace(name, keyCache.getUnchecked(key), existing, serializer.encode(newValue))
+                .thenApply(this::unwrapResult);
     }
 
     @Override
     public CompletableFuture<Boolean> replace(K key, long oldVersion, V newValue) {
         checkNotNull(key, ERROR_NULL_KEY);
         checkNotNull(newValue, ERROR_NULL_VALUE);
-        return database.replace(name, keyCache.getUnchecked(key), oldVersion, serializer.encode(newValue));
+        return database.replace(name, keyCache.getUnchecked(key), oldVersion, serializer.encode(newValue))
+                .thenApply(this::unwrapResult);
     }
 
     private Map.Entry<K, Versioned<V>> fromRawEntry(Map.Entry<String, Versioned<byte[]>> e) {
@@ -197,4 +206,14 @@
                         e.getValue().version(),
                         e.getValue().creationTime()));
     }
+
+    private <T> T unwrapResult(Result<T> result) {
+        if (result.status() == Result.Status.LOCKED) {
+            throw new ConsistentMapException.ConcurrentModification();
+        } else if (result.success()) {
+            return result.value();
+        } else {
+            throw new IllegalStateException("Must not be here");
+        }
+    }
 }
\ No newline at end of file
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabase.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabase.java
index c14e57c..8ed7670 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabase.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabase.java
@@ -23,13 +23,12 @@
 import net.kuujo.copycat.util.concurrent.Futures;
 
 import java.util.Collection;
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Supplier;
 
-import org.onosproject.store.service.UpdateOperation;
+import org.onosproject.store.service.Transaction;
 import org.onosproject.store.service.Versioned;
 
 /**
@@ -39,7 +38,7 @@
     private final StateMachine<DatabaseState<String, byte[]>> stateMachine;
     private DatabaseProxy<String, byte[]> proxy;
 
-    @SuppressWarnings("unchecked")
+    @SuppressWarnings({ "unchecked", "rawtypes" })
     public DefaultDatabase(ResourceContext context) {
         super(context);
         this.stateMachine = new DefaultStateMachine(context, DatabaseState.class, DefaultDatabaseState.class);
@@ -91,17 +90,17 @@
     }
 
     @Override
-    public CompletableFuture<Versioned<byte[]>> put(String tableName, String key, byte[] value) {
+    public CompletableFuture<Result<Versioned<byte[]>>> put(String tableName, String key, byte[] value) {
         return checkOpen(() -> proxy.put(tableName, key, value));
     }
 
     @Override
-    public CompletableFuture<Versioned<byte[]>> remove(String tableName, String key) {
+    public CompletableFuture<Result<Versioned<byte[]>>> remove(String tableName, String key) {
         return checkOpen(() -> proxy.remove(tableName, key));
     }
 
     @Override
-    public CompletableFuture<Void> clear(String tableName) {
+    public CompletableFuture<Result<Void>> clear(String tableName) {
         return checkOpen(() -> proxy.clear(tableName));
     }
 
@@ -121,33 +120,48 @@
     }
 
     @Override
-    public CompletableFuture<Versioned<byte[]>> putIfAbsent(String tableName, String key, byte[] value) {
+    public CompletableFuture<Result<Versioned<byte[]>>> putIfAbsent(String tableName, String key, byte[] value) {
         return checkOpen(() -> proxy.putIfAbsent(tableName, key, value));
     }
 
     @Override
-    public CompletableFuture<Boolean> remove(String tableName, String key, byte[] value) {
+    public CompletableFuture<Result<Boolean>> remove(String tableName, String key, byte[] value) {
         return checkOpen(() -> proxy.remove(tableName, key, value));
     }
 
     @Override
-    public CompletableFuture<Boolean> remove(String tableName, String key, long version) {
+    public CompletableFuture<Result<Boolean>> remove(String tableName, String key, long version) {
         return checkOpen(() -> proxy.remove(tableName, key, version));
     }
 
     @Override
-    public CompletableFuture<Boolean> replace(String tableName, String key, byte[] oldValue, byte[] newValue) {
+    public CompletableFuture<Result<Boolean>> replace(String tableName, String key, byte[] oldValue, byte[] newValue) {
         return checkOpen(() -> proxy.replace(tableName, key, oldValue, newValue));
     }
 
     @Override
-    public CompletableFuture<Boolean> replace(String tableName, String key, long oldVersion, byte[] newValue) {
+    public CompletableFuture<Result<Boolean>> replace(String tableName, String key, long oldVersion, byte[] newValue) {
         return checkOpen(() -> proxy.replace(tableName, key, oldVersion, newValue));
     }
 
     @Override
-    public CompletableFuture<Boolean> atomicBatchUpdate(List<UpdateOperation<String, byte[]>> updates) {
-        return checkOpen(() -> proxy.atomicBatchUpdate(updates));
+    public CompletableFuture<Boolean> prepareAndCommit(Transaction transaction) {
+        return checkOpen(() -> proxy.prepareAndCommit(transaction));
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepare(Transaction transaction) {
+        return checkOpen(() -> proxy.prepare(transaction));
+    }
+
+    @Override
+    public CompletableFuture<Boolean> commit(Transaction transaction) {
+        return checkOpen(() -> proxy.commit(transaction));
+    }
+
+    @Override
+    public CompletableFuture<Boolean> rollback(Transaction transaction) {
+        return checkOpen(() -> proxy.rollback(transaction));
     }
 
     @Override
@@ -180,4 +194,4 @@
         }
         return false;
     }
-}
\ No newline at end of file
+}
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabaseState.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabaseState.java
index e63a3d8..c190a28 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabaseState.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabaseState.java
@@ -18,43 +18,58 @@
 
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.HashSet;
-import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.stream.Collectors;
 import java.util.Set;
 
 import org.apache.commons.lang3.tuple.Pair;
-import org.onosproject.store.service.UpdateOperation;
+import org.onosproject.store.service.DatabaseUpdate;
+import org.onosproject.store.service.Transaction;
 import org.onosproject.store.service.Versioned;
+import org.onosproject.store.service.DatabaseUpdate.Type;
 
+import com.google.common.base.Objects;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Maps;
 
 import net.kuujo.copycat.state.Initializer;
 import net.kuujo.copycat.state.StateContext;
 
 /**
  * Default database state.
- *
- * @param <K> key type
- * @param <V> value type
  */
-public class DefaultDatabaseState<K, V> implements DatabaseState<K, V> {
-
+public class DefaultDatabaseState implements DatabaseState<String, byte[]> {
     private Long nextVersion;
-    private Map<String, Map<K, Versioned<V>>> tables;
+    private Map<String, Map<String, Versioned<byte[]>>> tables;
+
+    /**
+     * This locks map has a structure similar to the "tables" map above and
+     * holds all the provisional updates made during a transaction's prepare phase.
+     * The entry value is represented as the tuple: (transactionId, newValue)
+     * If newValue == null that signifies this update is attempting to
+     * delete the existing value.
+     * This map also serves as a lock on the entries that are being updated.
+     * The presence of a entry in this map indicates that element is
+     * participating in a transaction and is currently locked for updates.
+     */
+    private Map<String, Map<String, Pair<Long, byte[]>>> locks;
 
     @Initializer
     @Override
-    public void init(StateContext<DatabaseState<K, V>> context) {
+    public void init(StateContext<DatabaseState<String, byte[]>> context) {
         tables = context.get("tables");
         if (tables == null) {
-            tables = new HashMap<>();
+            tables = Maps.newConcurrentMap();
             context.put("tables", tables);
         }
+        locks = context.get("locks");
+        if (locks == null) {
+            locks = Maps.newConcurrentMap();
+            context.put("locks", locks);
+        }
         nextVersion = context.get("nextVersion");
         if (nextVersion == null) {
             nextVersion = new Long(0);
@@ -62,15 +77,6 @@
         }
     }
 
-    private Map<K, Versioned<V>> getTableMap(String tableName) {
-        Map<K, Versioned<V>> table = tables.get(tableName);
-        if (table == null) {
-            table = new HashMap<>();
-            tables.put(tableName, table);
-        }
-        return table;
-    }
-
     @Override
     public Set<String> tableNames() {
         return new HashSet<>(tables.keySet());
@@ -87,47 +93,55 @@
     }
 
     @Override
-    public boolean containsKey(String tableName, K key) {
+    public boolean containsKey(String tableName, String key) {
         return getTableMap(tableName).containsKey(key);
     }
 
     @Override
-    public boolean containsValue(String tableName, V value) {
-        return getTableMap(tableName).values().stream().anyMatch(v -> checkEquality(v.value(), value));
+    public boolean containsValue(String tableName, byte[] value) {
+        return getTableMap(tableName).values().stream().anyMatch(v -> Arrays.equals(v.value(), value));
     }
 
     @Override
-    public Versioned<V> get(String tableName, K key) {
+    public Versioned<byte[]> get(String tableName, String key) {
         return getTableMap(tableName).get(key);
     }
 
     @Override
-    public Versioned<V> put(String tableName, K key, V value) {
-        return getTableMap(tableName).put(key, new Versioned<>(value, ++nextVersion));
+    public Result<Versioned<byte[]>> put(String tableName, String key, byte[] value) {
+        return isLockedForUpdates(tableName, key)
+                ? Result.locked()
+                : Result.ok(getTableMap(tableName).put(key, new Versioned<>(value, ++nextVersion)));
     }
 
     @Override
-    public Versioned<V> remove(String tableName, K key) {
-        return getTableMap(tableName).remove(key);
+    public Result<Versioned<byte[]>> remove(String tableName, String key) {
+        return isLockedForUpdates(tableName, key)
+                ? Result.locked()
+                : Result.ok(getTableMap(tableName).remove(key));
     }
 
     @Override
-    public void clear(String tableName) {
+    public Result<Void> clear(String tableName) {
+        if (areTransactionsInProgress(tableName)) {
+            return Result.locked();
+        }
         getTableMap(tableName).clear();
+        return Result.ok(null);
     }
 
     @Override
-    public Set<K> keySet(String tableName) {
+    public Set<String> keySet(String tableName) {
         return ImmutableSet.copyOf(getTableMap(tableName).keySet());
     }
 
     @Override
-    public Collection<Versioned<V>> values(String tableName) {
+    public Collection<Versioned<byte[]>> values(String tableName) {
         return ImmutableList.copyOf(getTableMap(tableName).values());
     }
 
     @Override
-    public Set<Entry<K, Versioned<V>>> entrySet(String tableName) {
+    public Set<Entry<String, Versioned<byte[]>>> entrySet(String tableName) {
         return ImmutableSet.copyOf(getTableMap(tableName)
                 .entrySet()
                 .stream()
@@ -136,93 +150,113 @@
     }
 
     @Override
-    public Versioned<V> putIfAbsent(String tableName, K key, V value) {
-        Versioned<V> existingValue = getTableMap(tableName).get(key);
-        return existingValue != null ? existingValue : put(tableName, key, value);
-    }
-
-    @Override
-    public boolean remove(String tableName, K key, V value) {
-        Versioned<V> existing = getTableMap(tableName).get(key);
-        if (existing != null && checkEquality(existing.value(), value)) {
-            getTableMap(tableName).remove(key);
-            return true;
+    public Result<Versioned<byte[]>> putIfAbsent(String tableName, String key, byte[] value) {
+        if (isLockedForUpdates(tableName, key)) {
+            return Result.locked();
         }
-        return false;
+        Versioned<byte[]> existingValue = get(tableName, key);
+        Versioned<byte[]> currentValue = existingValue != null ? existingValue : put(tableName, key, value).value();
+        return Result.ok(currentValue);
     }
 
     @Override
-    public boolean remove(String tableName, K key, long version) {
-        Versioned<V> existing = getTableMap(tableName).get(key);
+    public Result<Boolean> remove(String tableName, String key, byte[] value) {
+        if (isLockedForUpdates(tableName, key)) {
+            return Result.locked();
+        }
+        Versioned<byte[]> existing = get(tableName, key);
+        if (existing != null && Arrays.equals(existing.value(), value)) {
+            getTableMap(tableName).remove(key);
+            return Result.ok(true);
+        }
+        return Result.ok(false);
+    }
+
+    @Override
+    public Result<Boolean> remove(String tableName, String key, long version) {
+        if (isLockedForUpdates(tableName, key)) {
+            return Result.locked();
+        }
+        Versioned<byte[]> existing = get(tableName, key);
         if (existing != null && existing.version() == version) {
             remove(tableName, key);
-            return true;
+            return Result.ok(true);
         }
-        return false;
+        return Result.ok(false);
     }
 
     @Override
-    public boolean replace(String tableName, K key, V oldValue, V newValue) {
-        Versioned<V> existing = getTableMap(tableName).get(key);
-        if (existing != null && checkEquality(existing.value(), oldValue)) {
+    public Result<Boolean> replace(String tableName, String key, byte[] oldValue, byte[] newValue) {
+        if (isLockedForUpdates(tableName, key)) {
+            return Result.locked();
+        }
+        Versioned<byte[]> existing = get(tableName, key);
+        if (existing != null && Arrays.equals(existing.value(), oldValue)) {
             put(tableName, key, newValue);
-            return true;
+            return Result.ok(true);
         }
-        return false;
+        return Result.ok(false);
     }
 
     @Override
-    public boolean replace(String tableName, K key, long oldVersion, V newValue) {
-        Versioned<V> existing = getTableMap(tableName).get(key);
+    public Result<Boolean> replace(String tableName, String key, long oldVersion, byte[] newValue) {
+        if (isLockedForUpdates(tableName, key)) {
+            return Result.locked();
+        }
+        Versioned<byte[]> existing = get(tableName, key);
         if (existing != null && existing.version() == oldVersion) {
             put(tableName, key, newValue);
+            return Result.ok(true);
+        }
+        return Result.ok(false);
+    }
+
+    @Override
+    public boolean prepareAndCommit(Transaction transaction) {
+        if (prepare(transaction)) {
+            return commit(transaction);
+        }
+        return false;
+    }
+
+    @Override
+    public boolean prepare(Transaction transaction) {
+        if (transaction.updates().stream().anyMatch(update ->
+                    isLockedByAnotherTransaction(update.tableName(),
+                                                 update.key(),
+                                                 transaction.id()))) {
+            return false;
+        }
+
+        if (transaction.updates().stream().allMatch(this::isUpdatePossible)) {
+            transaction.updates().forEach(update -> doProvisionalUpdate(update, transaction.id()));
             return true;
         }
         return false;
     }
 
     @Override
-    public boolean batchUpdate(List<UpdateOperation<K, V>> updates) {
-        if (updates.stream().anyMatch(update -> !checkIfUpdateIsPossible(update))) {
-            return false;
-        } else {
-            updates.stream().forEach(this::doUpdate);
-            return true;
-        }
+    public boolean commit(Transaction transaction) {
+        transaction.updates().forEach(update -> commitProvisionalUpdate(update, transaction.id()));
+        return true;
     }
 
-    private void doUpdate(UpdateOperation<K, V> update) {
-        String tableName = update.tableName();
-        K key = update.key();
-        switch (update.type()) {
-        case PUT:
-            put(tableName, key, update.value());
-            return;
-        case REMOVE:
-            remove(tableName, key);
-            return;
-        case PUT_IF_ABSENT:
-            putIfAbsent(tableName, key, update.value());
-            return;
-        case PUT_IF_VERSION_MATCH:
-            replace(tableName, key, update.currentValue(), update.value());
-            return;
-        case PUT_IF_VALUE_MATCH:
-            replace(tableName, key, update.currentVersion(), update.value());
-            return;
-        case REMOVE_IF_VERSION_MATCH:
-            remove(tableName, key, update.currentVersion());
-            return;
-        case REMOVE_IF_VALUE_MATCH:
-            remove(tableName, key, update.currentValue());
-            return;
-        default:
-            throw new IllegalStateException("Unsupported type: " + update.type());
-        }
+    @Override
+    public boolean rollback(Transaction transaction) {
+        transaction.updates().forEach(update -> undoProvisionalUpdate(update, transaction.id()));
+        return true;
     }
 
-    private boolean checkIfUpdateIsPossible(UpdateOperation<K, V> update) {
-        Versioned<V> existingEntry = get(update.tableName(), update.key());
+    private Map<String, Versioned<byte[]>> getTableMap(String tableName) {
+        return tables.computeIfAbsent(tableName, name -> Maps.newConcurrentMap());
+    }
+
+    private Map<String, Pair<Long, byte[]>> getLockMap(String tableName) {
+        return locks.computeIfAbsent(tableName, name -> Maps.newConcurrentMap());
+    }
+
+    private boolean isUpdatePossible(DatabaseUpdate update) {
+        Versioned<byte[]> existingEntry = get(update.tableName(), update.key());
         switch (update.type()) {
         case PUT:
         case REMOVE:
@@ -232,20 +266,85 @@
         case PUT_IF_VERSION_MATCH:
             return existingEntry != null && existingEntry.version() == update.currentVersion();
         case PUT_IF_VALUE_MATCH:
-            return existingEntry != null && checkEquality(existingEntry.value(), update.currentValue());
+            return existingEntry != null && Arrays.equals(existingEntry.value(), update.currentValue());
         case REMOVE_IF_VERSION_MATCH:
             return existingEntry == null || existingEntry.version() == update.currentVersion();
         case REMOVE_IF_VALUE_MATCH:
-            return existingEntry == null || checkEquality(existingEntry.value(), update.currentValue());
+            return existingEntry == null || Arrays.equals(existingEntry.value(), update.currentValue());
         default:
             throw new IllegalStateException("Unsupported type: " + update.type());
         }
     }
 
-    private boolean checkEquality(V value1, V value2) {
-        if (value1 instanceof byte[]) {
-            return Arrays.equals((byte[]) value1, (byte[]) value2);
+    private void doProvisionalUpdate(DatabaseUpdate update, long transactionId) {
+        Map<String, Pair<Long, byte[]>> lockMap = getLockMap(update.tableName());
+        switch (update.type()) {
+        case PUT:
+        case PUT_IF_ABSENT:
+        case PUT_IF_VERSION_MATCH:
+        case PUT_IF_VALUE_MATCH:
+            lockMap.put(update.key(), Pair.of(transactionId, update.value()));
+            break;
+        case REMOVE:
+        case REMOVE_IF_VERSION_MATCH:
+        case REMOVE_IF_VALUE_MATCH:
+            lockMap.put(update.key(), null);
+            break;
+        default:
+            throw new IllegalStateException("Unsupported type: " + update.type());
         }
-        return value1.equals(value2);
+    }
+
+    private void commitProvisionalUpdate(DatabaseUpdate update, long transactionId) {
+        String tableName = update.tableName();
+        String key = update.key();
+        Type type = update.type();
+        Pair<Long, byte[]> provisionalUpdate = getLockMap(tableName).get(key);
+        if (Objects.equal(transactionId, provisionalUpdate.getLeft()))  {
+            getLockMap(tableName).remove(key);
+        } else {
+            return;
+        }
+
+        switch (type) {
+        case PUT:
+        case PUT_IF_ABSENT:
+        case PUT_IF_VERSION_MATCH:
+        case PUT_IF_VALUE_MATCH:
+            put(tableName, key, provisionalUpdate.getRight());
+            break;
+        case REMOVE:
+        case REMOVE_IF_VERSION_MATCH:
+        case REMOVE_IF_VALUE_MATCH:
+            remove(tableName, key);
+            break;
+        default:
+            break;
+        }
+    }
+
+    private void undoProvisionalUpdate(DatabaseUpdate update, long transactionId) {
+        String tableName = update.tableName();
+        String key = update.key();
+        Pair<Long, byte[]> provisionalUpdate = getLockMap(tableName).get(key);
+        if (provisionalUpdate == null) {
+            return;
+        }
+        if (Objects.equal(transactionId, provisionalUpdate.getLeft()))  {
+            getLockMap(tableName).remove(key);
+        }
+    }
+
+    private boolean isLockedByAnotherTransaction(String tableName, String key, long transactionId) {
+        Pair<Long, byte[]> update = getLockMap(tableName).get(key);
+        return update != null && !Objects.equal(transactionId, update.getLeft());
+    }
+
+    private boolean isLockedForUpdates(String tableName, String key) {
+        return getLockMap(tableName).containsKey(key);
+    }
+
+    private boolean areTransactionsInProgress(String tableName) {
+        return !getLockMap(tableName).isEmpty();
     }
 }
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDistributedSet.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDistributedSet.java
index e6eac62..97d5ef3 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDistributedSet.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDistributedSet.java
@@ -18,6 +18,7 @@
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.Set;
+
 import org.onosproject.store.service.ConsistentMap;
 import org.onosproject.store.service.Serializer;
 
@@ -46,6 +47,7 @@
         return backingMap.isEmpty();
     }
 
+    @SuppressWarnings("unchecked")
     @Override
     public boolean contains(Object o) {
         return backingMap.containsKey((E) o);
@@ -71,6 +73,7 @@
         return backingMap.putIfAbsent(e, true) == null;
     }
 
+    @SuppressWarnings("unchecked")
     @Override
     public boolean remove(Object o) {
         return backingMap.remove((E) o, true);
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultTransaction.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultTransaction.java
new file mode 100644
index 0000000..2ff7a2d
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultTransaction.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2015 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.store.consistent.impl;
+
+import java.util.List;
+
+import org.onosproject.store.service.DatabaseUpdate;
+import org.onosproject.store.service.Transaction;
+
+import com.google.common.collect.ImmutableList;
+
+/**
+ * A Default transaction implementation.
+ */
+public class DefaultTransaction implements Transaction {
+
+    private final long transactionId;
+    private final List<DatabaseUpdate> updates;
+    private final State state;
+    private final long lastUpdated;
+
+    public DefaultTransaction(long transactionId, List<DatabaseUpdate> updates) {
+        this(transactionId, updates, State.PREPARING, System.currentTimeMillis());
+    }
+
+    private DefaultTransaction(long transactionId, List<DatabaseUpdate> updates, State state, long lastUpdated) {
+        this.transactionId = transactionId;
+        this.updates = ImmutableList.copyOf(updates);
+        this.state = state;
+        this.lastUpdated = lastUpdated;
+    }
+
+    @Override
+    public long id() {
+        return transactionId;
+    }
+
+    @Override
+    public List<DatabaseUpdate> updates() {
+        return updates;
+    }
+
+    @Override
+    public State state() {
+        return state;
+    }
+
+    @Override
+    public Transaction transition(State newState) {
+        return new DefaultTransaction(transactionId, updates, newState, System.currentTimeMillis());
+    }
+
+    @Override
+    public long lastUpdated() {
+        return lastUpdated;
+    }
+}
\ No newline at end of file
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultTransactionContext.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultTransactionContext.java
index d21543a..9d13833 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultTransactionContext.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultTransactionContext.java
@@ -18,19 +18,14 @@
 
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 
 import static com.google.common.base.Preconditions.*;
 
-import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.DatabaseUpdate;
 import org.onosproject.store.service.Serializer;
 import org.onosproject.store.service.TransactionContext;
 import org.onosproject.store.service.TransactionException;
 import org.onosproject.store.service.TransactionalMap;
-import org.onosproject.store.service.UpdateOperation;
 
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
@@ -40,80 +35,69 @@
  */
 public class DefaultTransactionContext implements TransactionContext {
 
-    private final Map<String, DefaultTransactionalMap> txMaps = Maps.newHashMap();
+    private static final String TX_NOT_OPEN_ERROR = "Transaction Context is not open";
+
+    @SuppressWarnings("rawtypes")
+    private final Map<String, DefaultTransactionalMap> txMaps = Maps.newConcurrentMap();
     private boolean isOpen = false;
     private final Database database;
-    private static final String TX_NOT_OPEN_ERROR = "Transaction is not open";
-    private static final int TRANSACTION_TIMEOUT_MILLIS = 2000;
+    private final long transactionId;
 
-    DefaultTransactionContext(Database database) {
-        this.database = checkNotNull(database, "Database must not be null");
+    public DefaultTransactionContext(Database database, long transactionId) {
+        this.database = checkNotNull(database);
+        this.transactionId = transactionId;
+    }
+
+    @Override
+    public long transactionId() {
+        return transactionId;
     }
 
     @Override
     public void begin() {
+        checkState(!isOpen, "Transaction Context is already open");
         isOpen = true;
     }
 
     @Override
+    public boolean isOpen() {
+        return isOpen;
+    }
+
+    @Override
     @SuppressWarnings("unchecked")
-    public <K, V> TransactionalMap<K, V> createTransactionalMap(String mapName,
+    public <K, V> TransactionalMap<K, V> getTransactionalMap(String mapName,
             Serializer serializer) {
-        checkNotNull(mapName, "map name is null");
-        checkNotNull(serializer, "serializer is null");
         checkState(isOpen, TX_NOT_OPEN_ERROR);
-        if (!txMaps.containsKey(mapName)) {
-            ConsistentMap<K, V> backingMap = new DefaultConsistentMap<>(mapName, database, serializer);
-            DefaultTransactionalMap<K, V> txMap = new DefaultTransactionalMap<>(mapName, backingMap, this, serializer);
-            txMaps.put(mapName, txMap);
-        }
-        return txMaps.get(mapName);
+        checkNotNull(mapName);
+        checkNotNull(serializer);
+        return txMaps.computeIfAbsent(mapName, name -> new DefaultTransactionalMap<>(
+                                name,
+                                new DefaultConsistentMap<>(name, database, serializer),
+                                this,
+                                serializer));
     }
 
     @SuppressWarnings("unchecked")
     @Override
     public void commit() {
         checkState(isOpen, TX_NOT_OPEN_ERROR);
-        List<UpdateOperation<String, byte[]>> allUpdates =
-                Lists.newLinkedList();
         try {
+            List<DatabaseUpdate> updates = Lists.newLinkedList();
             txMaps.values()
-                .stream()
-                .forEach(m -> {
-                    allUpdates.addAll(m.prepareDatabaseUpdates());
-                });
-
-            if (!complete(database.atomicBatchUpdate(allUpdates))) {
-                throw new TransactionException.OptimisticConcurrencyFailure();
-            }
+                  .forEach(m -> { updates.addAll(m.prepareDatabaseUpdates()); });
+            database.prepareAndCommit(new DefaultTransaction(transactionId, updates));
+        } catch (Exception e) {
+            abort();
+            throw new TransactionException(e);
         } finally {
             isOpen = false;
         }
     }
 
     @Override
-    public void rollback() {
+    public void abort() {
         checkState(isOpen, TX_NOT_OPEN_ERROR);
-        txMaps.values()
-        .stream()
-        .forEach(m -> m.rollback());
+        txMaps.values().forEach(m -> m.rollback());
     }
-
-    @Override
-    public boolean isOpen() {
-        return false;
-    }
-
-    private static <T> T complete(CompletableFuture<T> future) {
-        try {
-            return future.get(TRANSACTION_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            throw new TransactionException.Interrupted();
-        } catch (TimeoutException e) {
-            throw new TransactionException.Timeout();
-        } catch (ExecutionException e) {
-            throw new TransactionException(e.getCause());
-        }
-    }
-}
+}
\ No newline at end of file
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultTransactionalMap.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultTransactionalMap.java
index c98c336..91151b0 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultTransactionalMap.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultTransactionalMap.java
@@ -16,23 +16,24 @@
 
 package org.onosproject.store.consistent.impl;
 
-import java.util.Collection;
 import java.util.List;
 import java.util.Map;
-import java.util.Map.Entry;
-import java.util.stream.Collectors;
 import java.util.Set;
 
 import org.onlab.util.HexString;
 import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.DatabaseUpdate;
 import org.onosproject.store.service.Serializer;
 import org.onosproject.store.service.TransactionContext;
 import org.onosproject.store.service.TransactionalMap;
-import org.onosproject.store.service.UpdateOperation;
 import org.onosproject.store.service.Versioned;
 
 import static com.google.common.base.Preconditions.*;
 
+import com.google.common.base.Objects;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
@@ -55,6 +56,23 @@
     private final Map<K, V> writeCache = Maps.newConcurrentMap();
     private final Set<K> deleteSet = Sets.newConcurrentHashSet();
 
+    private static final String ERROR_NULL_VALUE = "Null values are not allowed";
+    private static final String ERROR_NULL_KEY = "Null key is not allowed";
+
+    private final LoadingCache<K, String> keyCache = CacheBuilder.newBuilder()
+            .softValues()
+            .build(new CacheLoader<K, String>() {
+
+                @Override
+                public String load(K key) {
+                    return HexString.toHexString(serializer.encode(key));
+                }
+            });
+
+    protected K dK(String key) {
+        return serializer.decode(HexString.fromHexString(key));
+    }
+
     public DefaultTransactionalMap(
             String name,
             ConsistentMap<K, V> backingMap,
@@ -69,15 +87,15 @@
     @Override
     public V get(K key) {
         checkState(txContext.isOpen(), TX_CLOSED_ERROR);
+        checkNotNull(key, ERROR_NULL_KEY);
         if (deleteSet.contains(key)) {
             return null;
-        } else if (writeCache.containsKey(key)) {
-            return writeCache.get(key);
+        }
+        V latest = writeCache.get(key);
+        if (latest != null) {
+            return latest;
         } else {
-            if (!readCache.containsKey(key)) {
-                readCache.put(key, backingMap.get(key));
-            }
-            Versioned<V> v = readCache.get(key);
+            Versioned<V> v = readCache.computeIfAbsent(key, k -> backingMap.get(k));
             return v != null ? v.value() : null;
         }
     }
@@ -85,25 +103,31 @@
     @Override
     public V put(K key, V value) {
         checkState(txContext.isOpen(), TX_CLOSED_ERROR);
-        Versioned<V> original = readCache.get(key);
-        V recentUpdate = writeCache.put(key, value);
+        checkNotNull(value, ERROR_NULL_VALUE);
+
+        V latest = get(key);
+        writeCache.put(key, value);
         deleteSet.remove(key);
-        return recentUpdate == null ? (original != null ? original.value() : null) : recentUpdate;
+        return latest;
     }
 
     @Override
     public V remove(K key) {
         checkState(txContext.isOpen(), TX_CLOSED_ERROR);
-        Versioned<V> original = readCache.get(key);
-        V recentUpdate = writeCache.remove(key);
-        deleteSet.add(key);
-        return recentUpdate == null ? (original != null ? original.value() : null) : recentUpdate;
+        V latest = get(key);
+        if (latest != null) {
+            writeCache.remove(key);
+            deleteSet.add(key);
+        }
+        return latest;
     }
 
     @Override
     public boolean remove(K key, V value) {
-        V currentValue = get(key);
-        if (value.equals(currentValue)) {
+        checkState(txContext.isOpen(), TX_CLOSED_ERROR);
+        checkNotNull(value, ERROR_NULL_VALUE);
+        V latest = get(key);
+        if (Objects.equal(value, latest)) {
             remove(key);
             return true;
         }
@@ -112,8 +136,11 @@
 
     @Override
     public boolean replace(K key, V oldValue, V newValue) {
-        V currentValue = get(key);
-        if (oldValue.equals(currentValue)) {
+        checkState(txContext.isOpen(), TX_CLOSED_ERROR);
+        checkNotNull(oldValue, ERROR_NULL_VALUE);
+        checkNotNull(newValue, ERROR_NULL_VALUE);
+        V latest = get(key);
+        if (Objects.equal(oldValue, latest)) {
             put(key, newValue);
             return true;
         }
@@ -121,70 +148,25 @@
     }
 
     @Override
-    public int size() {
-        // TODO
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public boolean isEmpty() {
-        return size() == 0;
-    }
-
-    @Override
-    public boolean containsKey(K key) {
-        return get(key) != null;
-    }
-
-    @Override
-    public boolean containsValue(V value) {
-        // TODO
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void clear() {
-        // TODO
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public Set<K> keySet() {
-        // TODO
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public Collection<V> values() {
-        // TODO
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public Set<Entry<K, V>> entrySet() {
-        // TODO
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
     public V putIfAbsent(K key, V value) {
-        V currentValue = get(key);
-        if (currentValue == null) {
+        checkState(txContext.isOpen(), TX_CLOSED_ERROR);
+        checkNotNull(value, ERROR_NULL_VALUE);
+        V latest = get(key);
+        if (latest == null) {
             put(key, value);
-            return null;
         }
-        return currentValue;
+        return latest;
     }
 
-    protected List<UpdateOperation<String, byte[]>> prepareDatabaseUpdates() {
-        List<UpdateOperation<K, V>> updates = Lists.newLinkedList();
+    protected List<DatabaseUpdate> prepareDatabaseUpdates() {
+        List<DatabaseUpdate> updates = Lists.newLinkedList();
         deleteSet.forEach(key -> {
             Versioned<V> original = readCache.get(key);
             if (original != null) {
-                updates.add(UpdateOperation.<K, V>newBuilder()
+                updates.add(DatabaseUpdate.newBuilder()
                         .withTableName(name)
-                        .withType(UpdateOperation.Type.REMOVE_IF_VERSION_MATCH)
-                        .withKey(key)
+                        .withType(DatabaseUpdate.Type.REMOVE_IF_VERSION_MATCH)
+                        .withKey(keyCache.getUnchecked(key))
                         .withCurrentVersion(original.version())
                         .build());
             }
@@ -192,44 +174,23 @@
         writeCache.forEach((key, value) -> {
             Versioned<V> original = readCache.get(key);
             if (original == null) {
-                updates.add(UpdateOperation.<K, V>newBuilder()
+                updates.add(DatabaseUpdate.newBuilder()
                         .withTableName(name)
-                        .withType(UpdateOperation.Type.PUT_IF_ABSENT)
-                        .withKey(key)
-                        .withValue(value)
+                        .withType(DatabaseUpdate.Type.PUT_IF_ABSENT)
+                        .withKey(keyCache.getUnchecked(key))
+                        .withValue(serializer.encode(value))
                         .build());
             } else {
-                updates.add(UpdateOperation.<K, V>newBuilder()
+                updates.add(DatabaseUpdate.newBuilder()
                         .withTableName(name)
-                        .withType(UpdateOperation.Type.PUT_IF_VERSION_MATCH)
-                        .withKey(key)
+                        .withType(DatabaseUpdate.Type.PUT_IF_VERSION_MATCH)
+                        .withKey(keyCache.getUnchecked(key))
                         .withCurrentVersion(original.version())
-                        .withValue(value)
+                        .withValue(serializer.encode(value))
                         .build());
             }
         });
-        return updates.stream().map(this::toRawUpdateOperation).collect(Collectors.toList());
-    }
-
-    private UpdateOperation<String, byte[]> toRawUpdateOperation(UpdateOperation<K, V> update) {
-
-        UpdateOperation.Builder<String, byte[]> rawUpdate = UpdateOperation.<String, byte[]>newBuilder();
-
-        rawUpdate = rawUpdate.withKey(HexString.toHexString(serializer.encode(update.key())))
-            .withCurrentVersion(update.currentVersion())
-            .withType(update.type());
-
-        rawUpdate = rawUpdate.withTableName(update.tableName());
-
-        if (update.value() != null) {
-            rawUpdate = rawUpdate.withValue(serializer.encode(update.value()));
-        }
-
-        if (update.currentValue() != null) {
-            rawUpdate = rawUpdate.withCurrentValue(serializer.encode(update.currentValue()));
-        }
-
-        return rawUpdate.build();
+        return updates;
     }
 
     /**
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/PartitionedDatabase.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/PartitionedDatabase.java
index ad049e6..178f49f 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/PartitionedDatabase.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/PartitionedDatabase.java
@@ -27,7 +27,8 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
-import org.onosproject.store.service.UpdateOperation;
+import org.onosproject.store.service.DatabaseUpdate;
+import org.onosproject.store.service.Transaction;
 import org.onosproject.store.service.Versioned;
 
 import com.google.common.collect.Lists;
@@ -129,24 +130,27 @@
     }
 
     @Override
-    public CompletableFuture<Versioned<byte[]>> put(String tableName, String key, byte[] value) {
+    public CompletableFuture<Result<Versioned<byte[]>>> put(String tableName, String key, byte[] value) {
         checkState(isOpen.get(), DB_NOT_OPEN);
         return partitioner.getPartition(tableName, key).put(tableName, key, value);
     }
 
     @Override
-    public CompletableFuture<Versioned<byte[]>> remove(String tableName, String key) {
+    public CompletableFuture<Result<Versioned<byte[]>>> remove(String tableName, String key) {
         checkState(isOpen.get(), DB_NOT_OPEN);
         return partitioner.getPartition(tableName, key).remove(tableName, key);
     }
 
     @Override
-    public CompletableFuture<Void> clear(String tableName) {
+    public CompletableFuture<Result<Void>> clear(String tableName) {
+        AtomicBoolean isLocked = new AtomicBoolean(false);
         checkState(isOpen.get(), DB_NOT_OPEN);
         return CompletableFuture.allOf(partitions
                     .stream()
-                    .map(p -> p.clear(tableName))
-                    .toArray(CompletableFuture[]::new));
+                    .map(p -> p.clear(tableName)
+                            .thenApply(v -> isLocked.compareAndSet(false, Result.Status.LOCKED == v.status())))
+                    .toArray(CompletableFuture[]::new))
+                .thenApply(v -> isLocked.get() ? Result.locked() : Result.ok(null));
     }
 
     @Override
@@ -183,59 +187,86 @@
     }
 
     @Override
-    public CompletableFuture<Versioned<byte[]>> putIfAbsent(String tableName, String key, byte[] value) {
+    public CompletableFuture<Result<Versioned<byte[]>>> putIfAbsent(String tableName, String key, byte[] value) {
         checkState(isOpen.get(), DB_NOT_OPEN);
         return partitioner.getPartition(tableName, key).putIfAbsent(tableName, key, value);
     }
 
     @Override
-    public CompletableFuture<Boolean> remove(String tableName, String key, byte[] value) {
+    public CompletableFuture<Result<Boolean>> remove(String tableName, String key, byte[] value) {
         checkState(isOpen.get(), DB_NOT_OPEN);
         return partitioner.getPartition(tableName, key).remove(tableName, key, value);
     }
 
     @Override
-    public CompletableFuture<Boolean> remove(String tableName, String key, long version) {
+    public CompletableFuture<Result<Boolean>> remove(String tableName, String key, long version) {
         checkState(isOpen.get(), DB_NOT_OPEN);
         return partitioner.getPartition(tableName, key).remove(tableName, key, version);
     }
 
     @Override
-    public CompletableFuture<Boolean> replace(String tableName, String key, byte[] oldValue, byte[] newValue) {
+    public CompletableFuture<Result<Boolean>> replace(
+            String tableName, String key, byte[] oldValue, byte[] newValue) {
         checkState(isOpen.get(), DB_NOT_OPEN);
         return partitioner.getPartition(tableName, key).replace(tableName, key, oldValue, newValue);
     }
 
     @Override
-    public CompletableFuture<Boolean> replace(String tableName, String key, long oldVersion, byte[] newValue) {
+    public CompletableFuture<Result<Boolean>> replace(
+            String tableName, String key, long oldVersion, byte[] newValue) {
         checkState(isOpen.get(), DB_NOT_OPEN);
         return partitioner.getPartition(tableName, key).replace(tableName, key, oldVersion, newValue);
     }
 
     @Override
-    public CompletableFuture<Boolean> atomicBatchUpdate(List<UpdateOperation<String, byte[]>> updates) {
-        checkState(isOpen.get(), DB_NOT_OPEN);
-        Map<Database, List<UpdateOperation<String, byte[]>>> perPartitionUpdates = Maps.newHashMap();
-        for (UpdateOperation<String, byte[]> update : updates) {
-            Database partition = partitioner.getPartition(update.tableName(), update.key());
-            List<UpdateOperation<String, byte[]>> partitionUpdates = perPartitionUpdates.get(partition);
-            if (partitionUpdates == null) {
-                partitionUpdates = Lists.newArrayList();
-                perPartitionUpdates.put(partition, partitionUpdates);
-            }
-            partitionUpdates.add(update);
-        }
-        if (perPartitionUpdates.size() > 1) {
-            // TODO
-            throw new UnsupportedOperationException("Cross partition transactional updates are not supported.");
+    public CompletableFuture<Boolean> prepareAndCommit(Transaction transaction) {
+        Map<Database, Transaction> subTransactions = createSubTransactions(transaction);
+        if (subTransactions.isEmpty()) {
+            return CompletableFuture.completedFuture(true);
+        } else if (subTransactions.size() == 1) {
+            Entry<Database, Transaction> entry =
+                    subTransactions.entrySet().iterator().next();
+            return entry.getKey().prepareAndCommit(entry.getValue());
         } else {
-            Entry<Database, List<UpdateOperation<String, byte[]>>> only =
-                    perPartitionUpdates.entrySet().iterator().next();
-            return only.getKey().atomicBatchUpdate(only.getValue());
+            return new TransactionManager(this).execute(transaction);
         }
     }
 
     @Override
+    public CompletableFuture<Boolean> prepare(Transaction transaction) {
+        Map<Database, Transaction> subTransactions = createSubTransactions(transaction);
+        AtomicBoolean status = new AtomicBoolean(true);
+        return CompletableFuture.allOf(subTransactions.entrySet()
+                .stream()
+                .map(entry -> entry
+                        .getKey()
+                        .prepare(entry.getValue())
+                        .thenApply(v -> status.compareAndSet(true, v)))
+                .toArray(CompletableFuture[]::new))
+            .thenApply(v -> status.get());
+    }
+
+    @Override
+    public CompletableFuture<Boolean> commit(Transaction transaction) {
+        Map<Database, Transaction> subTransactions = createSubTransactions(transaction);
+        return CompletableFuture.allOf(subTransactions.entrySet()
+                .stream()
+                .map(entry -> entry.getKey().commit(entry.getValue()))
+                .toArray(CompletableFuture[]::new))
+        .thenApply(v -> true);
+    }
+
+    @Override
+    public CompletableFuture<Boolean> rollback(Transaction transaction) {
+        Map<Database, Transaction> subTransactions = createSubTransactions(transaction);
+        return CompletableFuture.allOf(subTransactions.entrySet()
+                .stream()
+                .map(entry -> entry.getKey().rollback(entry.getValue()))
+                .toArray(CompletableFuture[]::new))
+            .thenApply(v -> true);
+    }
+
+    @Override
     public CompletableFuture<Database> open() {
         return CompletableFuture.allOf(partitions
                     .stream()
@@ -243,7 +274,8 @@
                     .toArray(CompletableFuture[]::new))
                 .thenApply(v -> {
                     isOpen.set(true);
-                    return this; });
+                    return this;
+                });
     }
 
     @Override
@@ -279,4 +311,19 @@
     public Database addShutdownTask(Task<CompletableFuture<Void>> task) {
         throw new UnsupportedOperationException();
     }
-}
\ No newline at end of file
+
+    private Map<Database, Transaction> createSubTransactions(
+            Transaction transaction) {
+        Map<Database, List<DatabaseUpdate>> perPartitionUpdates = Maps.newHashMap();
+        for (DatabaseUpdate update : transaction.updates()) {
+            Database partition = partitioner.getPartition(update.tableName(), update.key());
+            List<DatabaseUpdate> partitionUpdates =
+                    perPartitionUpdates.computeIfAbsent(partition, k -> Lists.newLinkedList());
+            partitionUpdates.add(update);
+        }
+        Map<Database, Transaction> subTransactions = Maps.newHashMap();
+        perPartitionUpdates.forEach((k, v) -> subTransactions.put(k, new DefaultTransaction(transaction.id(), v)));
+
+        return subTransactions;
+    }
+}
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/Result.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/Result.java
new file mode 100644
index 0000000..548174a
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/Result.java
@@ -0,0 +1,91 @@
+/*
+ * Copyright 2015 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.store.consistent.impl;
+
+/**
+ * Result of a database update operation.
+ *
+ * @param <V> return value type
+ */
+public final class Result<V> {
+
+    public enum Status {
+        /**
+         * Indicates a successful update.
+         */
+        OK,
+
+        /**
+         * Indicates a failure due to underlying state being locked by another transaction.
+         */
+        LOCKED
+    }
+
+    private final Status status;
+    private final V value;
+
+    /**
+     * Creates a new Result instance with the specified value with status set to Status.OK.
+     *
+     * @param <V> result value type
+     * @param value result value
+     * @return Result instance
+     */
+    public static <V> Result<V> ok(V value) {
+        return new Result<>(value, Status.OK);
+    }
+
+    /**
+     * Creates a new Result instance with status set to Status.LOCKED.
+     *
+     * @param <V> result value type
+     * @return Result instance
+     */
+    public static <V> Result<V> locked() {
+        return new Result<>(null, Status.LOCKED);
+    }
+
+    private Result(V value, Status status) {
+        this.value = value;
+        this.status = status;
+    }
+
+    /**
+     * Returns true if this result indicates a successful execution i.e status is Status.OK.
+     *
+     * @return true if successful, false otherwise
+     */
+    public boolean success() {
+        return status == Status.OK;
+    }
+
+    /**
+     * Returns the status of database update operation.
+     * @return database update status
+     */
+    public Status status() {
+        return status;
+    }
+
+    /**
+     * Returns the return value for the update.
+     * @return value returned by database update. If the status is another
+     * other than Status.OK, this returns a null
+     */
+    public V value() {
+        return value;
+    }
+}
\ No newline at end of file
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/SimpleTableHashPartitioner.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/SimpleTableHashPartitioner.java
index adc5477..e8daeff 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/SimpleTableHashPartitioner.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/SimpleTableHashPartitioner.java
@@ -36,4 +36,4 @@
     public Database getPartition(String tableName, String key) {
         return partitions.get(hash(tableName) % partitions.size());
     }
-}
\ No newline at end of file
+}
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/TransactionManager.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/TransactionManager.java
new file mode 100644
index 0000000..865934b
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/TransactionManager.java
@@ -0,0 +1,135 @@
+/*
+ * Copyright 2015 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.store.consistent.impl;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.onlab.util.KryoNamespace;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.AsyncConsistentMap;
+import org.onosproject.store.service.DatabaseUpdate;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.Transaction;
+import org.onosproject.store.service.Versioned;
+import org.onosproject.store.service.Transaction.State;
+
+/**
+ * Agent that runs the two phase commit protocol.
+ */
+public class TransactionManager {
+
+    private final Database database;
+    private final AsyncConsistentMap<Long, Transaction> transactions;
+
+    private final Serializer serializer = new Serializer() {
+
+        private KryoNamespace kryo = KryoNamespace.newBuilder()
+                .register(KryoNamespaces.BASIC)
+                .nextId(KryoNamespace.FLOATING_ID)
+                .register(Versioned.class)
+                .register(DatabaseUpdate.class)
+                .register(DatabaseUpdate.Type.class)
+                .register(DefaultTransaction.class)
+                .register(Transaction.State.class)
+                .register(Pair.class)
+                .register(ImmutablePair.class)
+                .build();
+
+        @Override
+        public <T> byte[] encode(T object) {
+            return kryo.serialize(object);
+        }
+
+        @Override
+        public <T> T decode(byte[] bytes) {
+            return kryo.deserialize(bytes);
+        }
+    };
+
+    /**
+     * Constructs a new TransactionManager for the specified database instance.
+     *
+     * @param database database
+     */
+    public TransactionManager(Database database) {
+        this.database = checkNotNull(database, "database cannot be null");
+        this.transactions = new DefaultAsyncConsistentMap<>("onos-transactions", this.database, serializer);
+    }
+
+    /**
+     * Executes the specified transaction by employing a two phase commit protocol.
+     *
+     * @param transaction transaction to commit
+     * @return transaction result. Result value true indicates a successful commit, false
+     * indicates abort
+     */
+    public CompletableFuture<Boolean> execute(Transaction transaction) {
+        // clean up if this transaction in already in a terminal state.
+        if (transaction.state() == Transaction.State.COMMITTED ||
+                transaction.state() == Transaction.State.ROLLEDBACK) {
+            return transactions.remove(transaction.id()).thenApply(v -> true);
+        } else if (transaction.state() == Transaction.State.COMMITTING) {
+            return commit(transaction);
+        } else if (transaction.state() == Transaction.State.ROLLINGBACK) {
+            return rollback(transaction);
+        } else {
+            return prepare(transaction).thenCompose(v -> v ? commit(transaction) : rollback(transaction));
+        }
+    }
+
+
+    /**
+     * Returns all transactions in the system.
+     *
+     * @return future for a collection of transactions
+     */
+    public CompletableFuture<Collection<Transaction>> getTransactions() {
+        return transactions.values().thenApply(c -> {
+            Collection<Transaction> txns = c.stream().map(v -> v.value()).collect(Collectors.toList());
+            return txns;
+        });
+    }
+
+    private CompletableFuture<Boolean> prepare(Transaction transaction) {
+        return transactions.put(transaction.id(), transaction)
+                .thenCompose(v -> database.prepare(transaction))
+                .thenCompose(status -> transactions.put(
+                            transaction.id(),
+                            transaction.transition(status ? State.COMMITTING : State.ROLLINGBACK))
+                .thenApply(v -> status));
+    }
+
+    private CompletableFuture<Boolean> commit(Transaction transaction) {
+        return database.commit(transaction)
+                .thenCompose(v -> transactions.put(
+                            transaction.id(),
+                            transaction.transition(Transaction.State.COMMITTED)))
+                .thenApply(v -> true);
+    }
+
+    private CompletableFuture<Boolean> rollback(Transaction transaction) {
+        return database.rollback(transaction)
+                .thenCompose(v -> transactions.put(
+                            transaction.id(),
+                            transaction.transition(Transaction.State.ROLLEDBACK)))
+                .thenApply(v -> true);
+    }
+}
\ No newline at end of file