diff --git a/core/api/src/main/java/org/onosproject/store/primitives/MapUpdate.java b/core/api/src/main/java/org/onosproject/store/primitives/MapUpdate.java
index 3b12021..e04f10e 100644
--- a/core/api/src/main/java/org/onosproject/store/primitives/MapUpdate.java
+++ b/core/api/src/main/java/org/onosproject/store/primitives/MapUpdate.java
@@ -19,6 +19,7 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 
+import java.util.Objects;
 import java.util.function.Function;
 
 import org.onlab.util.ByteArraySizeHashPrinter;
@@ -42,6 +43,7 @@
          * Insert/Update entry without any checks.
          */
         PUT,
+
         /**
          * Insert an entry iff there is no existing entry for that key.
          */
@@ -73,7 +75,6 @@
         REMOVE_IF_VALUE_MATCH,
     }
 
-    private String mapName;
     private Type type;
     private K key;
     private V value;
@@ -81,15 +82,6 @@
     private long currentVersion = -1;
 
     /**
-     * Returns the name of the map.
-     *
-     * @return map name
-     */
-    public String mapName() {
-        return mapName;
-    }
-
-    /**
      * Returns the type of update operation.
      * @return type of update.
      */
@@ -140,7 +132,6 @@
      */
     public <S, T> MapUpdate<S, T> map(Function<K, S> keyMapper, Function<V, T> valueMapper) {
         return MapUpdate.<S, T>newBuilder()
-                .withMapName(mapName)
                 .withType(type)
                 .withKey(keyMapper.apply(key))
                 .withValue(value == null ? null : valueMapper.apply(value))
@@ -150,9 +141,26 @@
     }
 
     @Override
+    public int hashCode() {
+        return Objects.hash(type, key, value, currentValue, currentVersion);
+    }
+
+    @Override
+    public boolean equals(Object object) {
+        if (object instanceof MapUpdate) {
+            MapUpdate that = (MapUpdate) object;
+            return this.type == that.type
+                    && Objects.equals(this.key, that.key)
+                    && Objects.equals(this.value, that.value)
+                    && Objects.equals(this.currentValue, that.currentValue)
+                    && Objects.equals(this.currentVersion, that.currentVersion);
+        }
+        return false;
+    }
+
+    @Override
     public String toString() {
         return MoreObjects.toStringHelper(this)
-            .add("mapName", mapName)
             .add("type", type)
             .add("key", key)
             .add("value", value instanceof byte[] ? new ByteArraySizeHashPrinter((byte[]) value) : value)
@@ -187,11 +195,6 @@
             return update;
         }
 
-        public Builder<K, V> withMapName(String name) {
-            update.mapName = checkNotNull(name, "name cannot be null");
-            return this;
-        }
-
         public Builder<K, V> withType(Type type) {
             update.type = checkNotNull(type, "type cannot be null");
             return this;
diff --git a/core/api/src/main/java/org/onosproject/store/service/AsyncConsistentMap.java b/core/api/src/main/java/org/onosproject/store/service/AsyncConsistentMap.java
index 8bfd952..0704853 100644
--- a/core/api/src/main/java/org/onosproject/store/service/AsyncConsistentMap.java
+++ b/core/api/src/main/java/org/onosproject/store/service/AsyncConsistentMap.java
@@ -17,8 +17,8 @@
 package org.onosproject.store.service;
 
 import java.util.Collection;
-import java.util.Objects;
 import java.util.Map.Entry;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
@@ -26,10 +26,9 @@
 import java.util.function.Function;
 import java.util.function.Predicate;
 
-import org.onosproject.store.primitives.DefaultConsistentMap;
-import org.onosproject.store.primitives.TransactionId;
-
 import com.google.common.util.concurrent.MoreExecutors;
+import org.onosproject.store.primitives.DefaultConsistentMap;
+import org.onosproject.store.primitives.MapUpdate;
 
 /**
  * A distributed, strongly consistent map whose methods are all executed asynchronously.
@@ -55,7 +54,7 @@
  * the returned future will be {@link CompletableFuture#complete completed} when the
  * operation finishes.
  */
-public interface AsyncConsistentMap<K, V> extends DistributedPrimitive {
+public interface AsyncConsistentMap<K, V> extends DistributedPrimitive, Transactional<MapUpdate<K, V>> {
 
     @Override
     default DistributedPrimitive.Type primitiveType() {
@@ -346,36 +345,6 @@
     CompletableFuture<Void> removeListener(MapEventListener<K, V> listener);
 
     /**
-     * Prepares a transaction for commitment.
-     * @param transaction transaction
-     * @return {@code true} if prepare is successful and transaction is ready to be committed;
-     * {@code false} otherwise
-     */
-    CompletableFuture<Boolean> prepare(MapTransaction<K, V> transaction);
-
-    /**
-     * Commits a previously prepared transaction.
-     * @param transactionId transaction identifier
-     * @return future that will be completed when the operation finishes
-     */
-    CompletableFuture<Void> commit(TransactionId transactionId);
-
-    /**
-     * Aborts a previously prepared transaction.
-     * @param transactionId transaction identifier
-     * @return future that will be completed when the operation finishes
-     */
-    CompletableFuture<Void> rollback(TransactionId transactionId);
-
-    /**
-     * Prepares a transaction and commits it in one go.
-     * @param transaction transaction
-     * @return {@code true} if operation is successful and updates are committed
-     * {@code false} otherwise
-     */
-    CompletableFuture<Boolean> prepareAndCommit(MapTransaction<K, V> transaction);
-
-    /**
      * Returns a new {@link ConsistentMap} that is backed by this instance.
      *
      * @return new {@code ConsistentMap} instance
diff --git a/core/api/src/main/java/org/onosproject/store/service/MapTransaction.java b/core/api/src/main/java/org/onosproject/store/service/MapTransaction.java
deleted file mode 100644
index acedf17..0000000
--- a/core/api/src/main/java/org/onosproject/store/service/MapTransaction.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * Copyright 2016-present 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.service;
-
-import java.util.List;
-import java.util.function.Function;
-
-import org.onosproject.store.primitives.MapUpdate;
-import org.onosproject.store.primitives.TransactionId;
-
-import com.google.common.base.MoreObjects;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Lists;
-
-/**
- * Collection of map updates to be committed atomically.
- *
- * @param <K> key type
- * @param <V> value type
- */
-public class MapTransaction<K, V> {
-
-    private final TransactionId transactionId;
-    private final List<MapUpdate<K, V>> updates;
-
-    public MapTransaction(TransactionId transactionId, List<MapUpdate<K, V>> updates) {
-        this.transactionId = transactionId;
-        this.updates = ImmutableList.copyOf(updates);
-    }
-
-    /**
-     * Returns the transaction identifier.
-     *
-     * @return transaction id
-     */
-    public TransactionId transactionId() {
-        return transactionId;
-    }
-
-    /**
-     * Returns the list of map updates.
-     *
-     * @return map updates
-     */
-    public List<MapUpdate<K, V>> updates() {
-        return updates;
-    }
-
-    @Override
-    public String toString() {
-        return MoreObjects.toStringHelper(getClass())
-                .add("transactionId", transactionId)
-                .add("updates", updates)
-                .toString();
-    }
-
-    /**
-     * Maps this instance to another {@code MapTransaction} with different key and value types.
-     *
-     * @param keyMapper function for mapping key types
-     * @param valueMapper function for mapping value types
-     * @return newly typed instance
-     *
-     * @param <S> key type of returned instance
-     * @param <T> value type of returned instance
-     */
-    public <S, T> MapTransaction<S, T> map(Function<K, S> keyMapper, Function<V, T> valueMapper) {
-        return new MapTransaction<>(transactionId, Lists.transform(updates, u -> u.map(keyMapper, valueMapper)));
-    }
-}
diff --git a/core/api/src/main/java/org/onosproject/store/service/TransactionLog.java b/core/api/src/main/java/org/onosproject/store/service/TransactionLog.java
new file mode 100644
index 0000000..c367f5d
--- /dev/null
+++ b/core/api/src/main/java/org/onosproject/store/service/TransactionLog.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2017-present 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.service;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Function;
+
+import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import org.onosproject.store.primitives.TransactionId;
+
+/**
+ * Collection of transaction updates to be applied atomically.
+ *
+ * @param <T> log record type
+ */
+public class TransactionLog<T> {
+    private final TransactionId transactionId;
+    private final List<T> records;
+
+    public TransactionLog(TransactionId transactionId, List<T> records) {
+        this.transactionId = transactionId;
+        this.records = ImmutableList.copyOf(records);
+    }
+
+    /**
+     * Returns the transaction identifier.
+     *
+     * @return transaction id
+     */
+    public TransactionId transactionId() {
+        return transactionId;
+    }
+
+    /**
+     * Returns the list of transaction log records.
+     *
+     * @return a list of transaction log records
+     */
+    public List<T> records() {
+        return records;
+    }
+
+    @Override
+    public boolean equals(Object object) {
+        if (object instanceof TransactionLog) {
+            TransactionLog that = (TransactionLog) object;
+            return this.transactionId.equals(that.transactionId)
+                    && this.records.equals(that.records);
+        }
+        return false;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(transactionId, records);
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(getClass())
+                .add("transactionId", transactionId)
+                .add("records", records)
+                .toString();
+    }
+
+    /**
+     * Maps this instance to another {@code MapTransaction} with different key and value types.
+     *
+     * @param mapper function for mapping record types
+     * @return newly typed instance
+     *
+     * @param <U> record type of returned instance
+     */
+    public <U> TransactionLog<U> map(Function<T, U> mapper) {
+        return new TransactionLog<>(transactionId, Lists.transform(records, mapper::apply));
+    }
+}
\ No newline at end of file
diff --git a/core/api/src/main/java/org/onosproject/store/service/Transactional.java b/core/api/src/main/java/org/onosproject/store/service/Transactional.java
new file mode 100644
index 0000000..a6e1fbd
--- /dev/null
+++ b/core/api/src/main/java/org/onosproject/store/service/Transactional.java
@@ -0,0 +1,69 @@
+/*
+ * Copyright 2017-present 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.service;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.onosproject.store.primitives.TransactionId;
+
+/**
+ * Interface for transactional primitives.
+ */
+public interface Transactional<T> {
+
+    /**
+     * Begins the transaction.
+     *
+     * @param transactionId the transaction identifier for the transaction to begin
+     * @return a completable future to be completed with the lock version
+     */
+    CompletableFuture<Version> begin(TransactionId transactionId);
+
+    /**
+     * Prepares a transaction for commitment.
+     *
+     * @param transactionLog transaction log
+     * @return {@code true} if prepare is successful and transaction is ready to be committed
+     * {@code false} otherwise
+     */
+    CompletableFuture<Boolean> prepare(TransactionLog<T> transactionLog);
+
+    /**
+     * Prepares and commits a transaction.
+     *
+     * @param transactionLog transaction log
+     * @return {@code true} if prepare is successful and transaction was committed
+     * {@code false} otherwise
+     */
+    CompletableFuture<Boolean> prepareAndCommit(TransactionLog<T> transactionLog);
+
+    /**
+     * Commits a previously prepared transaction and unlocks the object.
+     *
+     * @param transactionId transaction identifier
+     * @return future that will be completed when the operation finishes
+     */
+    CompletableFuture<Void> commit(TransactionId transactionId);
+
+    /**
+     * Aborts a previously prepared transaction and unlocks the object.
+     *
+     * @param transactionId transaction identifier
+     * @return future that will be completed when the operation finishes
+     */
+    CompletableFuture<Void> rollback(TransactionId transactionId);
+
+}
\ No newline at end of file
diff --git a/core/api/src/main/java/org/onosproject/store/service/Version.java b/core/api/src/main/java/org/onosproject/store/service/Version.java
new file mode 100644
index 0000000..58a1d3d
--- /dev/null
+++ b/core/api/src/main/java/org/onosproject/store/service/Version.java
@@ -0,0 +1,83 @@
+/*
+ * Copyright 2017-present 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.service;
+
+import java.util.Objects;
+
+import com.google.common.base.MoreObjects;
+import com.google.common.collect.ComparisonChain;
+import org.onosproject.store.Timestamp;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+/**
+ * Logical timestamp for versions.
+ * <p>
+ * The version is a logical timestamp that represents a point in logical time at which an event occurs.
+ * This is used in both pessimistic and optimistic locking protocols to ensure that the state of a shared resource
+ * has not changed at the end of a transaction.
+ */
+public class Version implements Timestamp {
+    private final long version;
+
+    public Version(long version) {
+        this.version = version;
+    }
+
+    @Override
+    public int compareTo(Timestamp o) {
+        checkArgument(o instanceof Version,
+                "Must be LockVersion", o);
+        Version that = (Version) o;
+
+        return ComparisonChain.start()
+                .compare(this.version, that.version)
+                .result();
+    }
+
+    @Override
+    public int hashCode() {
+        return Long.hashCode(version);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (!(obj instanceof Version)) {
+            return false;
+        }
+        Version that = (Version) obj;
+        return Objects.equals(this.version, that.version);
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(getClass())
+                .add("version", version)
+                .toString();
+    }
+
+    /**
+     * Returns the lock version.
+     *
+     * @return the lock version
+     */
+    public long value() {
+        return this.version;
+    }
+}
\ No newline at end of file
diff --git a/core/api/src/test/java/org/onosproject/store/primitives/MapUpdateTest.java b/core/api/src/test/java/org/onosproject/store/primitives/MapUpdateTest.java
index 6fecd8c..53cc160 100644
--- a/core/api/src/test/java/org/onosproject/store/primitives/MapUpdateTest.java
+++ b/core/api/src/test/java/org/onosproject/store/primitives/MapUpdateTest.java
@@ -33,7 +33,6 @@
             .withValue("2".getBytes())
             .withCurrentVersion(3)
             .withKey("4")
-            .withMapName("5")
             .withType(MapUpdate.Type.PUT)
             .build();
 
@@ -42,7 +41,6 @@
             .withValue("2".getBytes())
             .withCurrentVersion(3)
             .withKey("4")
-            .withMapName("5")
             .withType(MapUpdate.Type.REMOVE)
             .build();
 
@@ -51,7 +49,6 @@
             .withValue("2".getBytes())
             .withCurrentVersion(3)
             .withKey("4")
-            .withMapName("5")
             .withType(MapUpdate.Type.REMOVE_IF_VALUE_MATCH)
             .build();
 
@@ -60,7 +57,6 @@
             .withValue("2".getBytes())
             .withCurrentVersion(3)
             .withKey("4")
-            .withMapName("5")
             .withType(MapUpdate.Type.REMOVE_IF_VERSION_MATCH)
             .build();
 
@@ -69,7 +65,6 @@
             .withValue("2".getBytes())
             .withCurrentVersion(3)
             .withKey("4")
-            .withMapName("5")
             .withType(MapUpdate.Type.PUT_IF_VALUE_MATCH)
             .build();
 
@@ -78,7 +73,6 @@
             .withValue("2".getBytes())
             .withCurrentVersion(3)
             .withKey("4")
-            .withMapName("5")
             .withType(MapUpdate.Type.PUT_IF_VERSION_MATCH)
             .build();
 
@@ -91,7 +85,6 @@
         assertThat(stats1.value(), is("2".getBytes()));
         assertThat(stats1.currentVersion(), is(3L));
         assertThat(stats1.key(), is("4"));
-        assertThat(stats1.mapName(), is("5"));
         assertThat(stats1.type(), is(MapUpdate.Type.PUT));
     }
 
diff --git a/core/api/src/test/java/org/onosproject/store/service/AsyncConsistentMapAdapter.java b/core/api/src/test/java/org/onosproject/store/service/AsyncConsistentMapAdapter.java
index c9161a3..90b4f51 100644
--- a/core/api/src/test/java/org/onosproject/store/service/AsyncConsistentMapAdapter.java
+++ b/core/api/src/test/java/org/onosproject/store/service/AsyncConsistentMapAdapter.java
@@ -24,6 +24,7 @@
 import java.util.function.BiFunction;
 import java.util.function.Predicate;
 
+import org.onosproject.store.primitives.MapUpdate;
 import org.onosproject.store.primitives.TransactionId;
 
 /**
@@ -143,7 +144,17 @@
     }
 
     @Override
-    public CompletableFuture<Boolean> prepare(MapTransaction<K, V> transaction) {
+    public CompletableFuture<Version> begin(TransactionId transactionId) {
+        return null;
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepare(TransactionLog<MapUpdate<K, V>> transactionLog) {
+        return null;
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepareAndCommit(TransactionLog<MapUpdate<K, V>> transactionLog) {
         return null;
     }
 
@@ -156,10 +167,5 @@
     public CompletableFuture<Void> rollback(TransactionId transactionId) {
         return null;
     }
-
-    @Override
-    public CompletableFuture<Boolean> prepareAndCommit(MapTransaction<K, V> transaction) {
-        return null;
-    }
 }
 
diff --git a/core/api/src/test/java/org/onosproject/store/service/VersionTest.java b/core/api/src/test/java/org/onosproject/store/service/VersionTest.java
new file mode 100644
index 0000000..1495ec3
--- /dev/null
+++ b/core/api/src/test/java/org/onosproject/store/service/VersionTest.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright 2017-present 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.service;
+
+import org.junit.Test;
+
+import static junit.framework.TestCase.assertFalse;
+import static junit.framework.TestCase.assertTrue;
+
+/**
+ * Version test.
+ */
+public class VersionTest {
+
+    @Test
+    public void testVersion() {
+        Version version1 = new Version(1);
+        Version version2 = new Version(1);
+        assertTrue(version1.equals(version2));
+        assertTrue(version1.hashCode() == version2.hashCode());
+        assertTrue(version1.value() == version2.value());
+
+        Version version3 = new Version(2);
+        assertFalse(version1.equals(version3));
+        assertFalse(version1.hashCode() == version3.hashCode());
+        assertFalse(version1.value() == version3.value());
+    }
+
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CatalystSerializers.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CatalystSerializers.java
index 645215c..d676ab1 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CatalystSerializers.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CatalystSerializers.java
@@ -50,7 +50,7 @@
 import org.onosproject.store.service.DocumentPath;
 import org.onosproject.store.service.DocumentTreeEvent;
 import org.onosproject.store.service.MapEvent;
-import org.onosproject.store.service.MapTransaction;
+import org.onosproject.store.service.TransactionLog;
 import org.onosproject.store.service.Task;
 import org.onosproject.store.service.Versioned;
 import org.onosproject.store.service.WorkQueueStats;
@@ -97,7 +97,7 @@
         serializer.register(TransactionId.class, factory);
         serializer.register(MapUpdate.class, factory);
         serializer.register(MapUpdate.Type.class, factory);
-        serializer.register(MapTransaction.class, factory);
+        serializer.register(TransactionLog.class, factory);
         serializer.register(Versioned.class, factory);
         serializer.register(MapEvent.class, factory);
         serializer.register(Task.class, factory);
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContext.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContext.java
index fccf48e..2f3fe11 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContext.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContext.java
@@ -15,11 +15,9 @@
  */
 package org.onosproject.store.primitives.impl;
 
-import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.onosproject.store.primitives.DistributedPrimitiveCreator;
 import org.onosproject.store.primitives.TransactionId;
 import org.onosproject.store.service.CommitStatus;
 import org.onosproject.store.service.Serializer;
@@ -27,7 +25,7 @@
 import org.onosproject.store.service.TransactionalMap;
 import org.onosproject.utils.MeteringAgent;
 
-import com.google.common.collect.Sets;
+import static com.google.common.base.MoreObjects.toStringHelper;
 
 /**
  * Default implementation of transaction context.
@@ -35,17 +33,12 @@
 public class DefaultTransactionContext implements TransactionContext {
 
     private final AtomicBoolean isOpen = new AtomicBoolean(false);
-    private final DistributedPrimitiveCreator creator;
     private final TransactionId transactionId;
     private final TransactionCoordinator transactionCoordinator;
-    private final Set<TransactionParticipant> txParticipants = Sets.newConcurrentHashSet();
     private final MeteringAgent monitor;
 
-    public DefaultTransactionContext(TransactionId transactionId,
-            DistributedPrimitiveCreator creator,
-            TransactionCoordinator transactionCoordinator) {
+    public DefaultTransactionContext(TransactionId transactionId, TransactionCoordinator transactionCoordinator) {
         this.transactionId = transactionId;
-        this.creator = creator;
         this.transactionCoordinator = transactionCoordinator;
         this.monitor = new MeteringAgent("transactionContext", "*", true);
     }
@@ -75,8 +68,7 @@
     @Override
     public CompletableFuture<CommitStatus> commit() {
         final MeteringAgent.Context timer = monitor.startTimer("commit");
-        return transactionCoordinator.commit(transactionId, txParticipants)
-                                     .whenComplete((r, e) -> timer.stop(e));
+        return transactionCoordinator.commit().whenComplete((r, e) -> timer.stop(e));
     }
 
     @Override
@@ -85,14 +77,14 @@
     }
 
     @Override
-    public <K, V> TransactionalMap<K, V> getTransactionalMap(String mapName,
-            Serializer serializer) {
-        // FIXME: Do not create duplicates.
-        DefaultTransactionalMap<K, V> txMap = new DefaultTransactionalMap<K, V>(mapName,
-                DistributedPrimitives.newMeteredMap(creator.<K, V>newAsyncConsistentMap(mapName, serializer)),
-                this,
-                serializer);
-        txParticipants.add(txMap);
-        return txMap;
+    public <K, V> TransactionalMap<K, V> getTransactionalMap(String mapName, Serializer serializer) {
+        return transactionCoordinator.getTransactionalMap(mapName, serializer);
+    }
+
+    @Override
+    public String toString() {
+        return toStringHelper(this)
+                .add("transactionId", transactionId)
+                .toString();
     }
 }
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContextBuilder.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContextBuilder.java
index c599306..12f36a0 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContextBuilder.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContextBuilder.java
@@ -15,31 +15,26 @@
  */
 package org.onosproject.store.primitives.impl;
 
-import org.onosproject.store.primitives.DistributedPrimitiveCreator;
 import org.onosproject.store.primitives.TransactionId;
 import org.onosproject.store.service.TransactionContext;
 import org.onosproject.store.service.TransactionContextBuilder;
+
 /**
  * Default Transaction Context Builder.
  */
 public class DefaultTransactionContextBuilder extends TransactionContextBuilder {
 
     private final TransactionId transactionId;
-    private final DistributedPrimitiveCreator primitiveCreator;
-    private final TransactionCoordinator transactionCoordinator;
+    private final TransactionManager transactionManager;
 
-    public DefaultTransactionContextBuilder(TransactionId transactionId,
-            DistributedPrimitiveCreator primitiveCreator,
-            TransactionCoordinator transactionCoordinator) {
+    public DefaultTransactionContextBuilder(TransactionId transactionId, TransactionManager transactionManager) {
         this.transactionId = transactionId;
-        this.primitiveCreator = primitiveCreator;
-        this.transactionCoordinator = transactionCoordinator;
+        this.transactionManager = transactionManager;
     }
 
     @Override
     public TransactionContext build() {
         return new DefaultTransactionContext(transactionId,
-                primitiveCreator,
-                transactionCoordinator);
+                new TransactionCoordinator(transactionId, transactionManager));
     }
 }
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionalMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionalMap.java
deleted file mode 100644
index d138c9b..0000000
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionalMap.java
+++ /dev/null
@@ -1,294 +0,0 @@
-/*
- * Copyright 2016-present 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.primitives.impl;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-import org.apache.commons.lang3.tuple.Pair;
-import org.onlab.util.HexString;
-import org.onosproject.store.primitives.MapUpdate;
-import org.onosproject.store.service.AsyncConsistentMap;
-import org.onosproject.store.service.ConsistentMap;
-import org.onosproject.store.service.MapTransaction;
-import org.onosproject.store.service.Serializer;
-import org.onosproject.store.service.TransactionContext;
-import org.onosproject.store.service.TransactionalMap;
-import org.onosproject.store.service.Versioned;
-
-import static com.google.common.base.Preconditions.*;
-
-import com.google.common.base.MoreObjects;
-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;
-
-/**
- * Default Transactional Map implementation that provides a repeatable reads
- * transaction isolation level.
- *
- * @param <K> key type
- * @param <V> value type.
- */
-public class DefaultTransactionalMap<K, V> implements TransactionalMap<K, V>, TransactionParticipant {
-
-    private final TransactionContext txContext;
-    private static final String TX_CLOSED_ERROR = "Transaction is closed";
-    private final AsyncConsistentMap<K, V> backingMap;
-    private final ConsistentMap<K, V> backingConsistentMap;
-    private final String name;
-    private final Serializer serializer;
-    private final Map<K, Versioned<V>> readCache = Maps.newConcurrentMap();
-    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,
-            AsyncConsistentMap<K, V> backingMap,
-            TransactionContext txContext,
-            Serializer serializer) {
-        this.name = name;
-        this.backingMap = backingMap;
-        this.backingConsistentMap = backingMap.asConsistentMap();
-        this.txContext = txContext;
-        this.serializer = serializer;
-    }
-
-    @Override
-    public V get(K key) {
-        checkState(txContext.isOpen(), TX_CLOSED_ERROR);
-        checkNotNull(key, ERROR_NULL_KEY);
-        if (deleteSet.contains(key)) {
-            return null;
-        }
-        V latest = writeCache.get(key);
-        if (latest != null) {
-            return latest;
-        } else {
-            Versioned<V> v = readCache.computeIfAbsent(key, k -> backingConsistentMap.get(k));
-            return v != null ? v.value() : null;
-        }
-    }
-
-    @Override
-    public boolean containsKey(K key) {
-        return get(key) != null;
-    }
-
-    @Override
-    public V put(K key, V value) {
-        checkState(txContext.isOpen(), TX_CLOSED_ERROR);
-        checkNotNull(value, ERROR_NULL_VALUE);
-
-        V latest = get(key);
-        writeCache.put(key, value);
-        deleteSet.remove(key);
-        return latest;
-    }
-
-    @Override
-    public V remove(K key) {
-        checkState(txContext.isOpen(), TX_CLOSED_ERROR);
-        V latest = get(key);
-        if (latest != null) {
-            writeCache.remove(key);
-            deleteSet.add(key);
-        }
-        return latest;
-    }
-
-    @Override
-    public boolean remove(K key, V value) {
-        checkState(txContext.isOpen(), TX_CLOSED_ERROR);
-        checkNotNull(value, ERROR_NULL_VALUE);
-        V latest = get(key);
-        if (Objects.equal(value, latest)) {
-            remove(key);
-            return true;
-        }
-        return false;
-    }
-
-    @Override
-    public boolean replace(K key, V oldValue, V newValue) {
-        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;
-        }
-        return false;
-    }
-
-    @Override
-    public V putIfAbsent(K key, V value) {
-        checkState(txContext.isOpen(), TX_CLOSED_ERROR);
-        checkNotNull(value, ERROR_NULL_VALUE);
-        V latest = get(key);
-        if (latest == null) {
-            put(key, value);
-        }
-        return latest;
-    }
-
-    @Override
-    public CompletableFuture<Boolean> prepare() {
-        return backingMap.prepare(new MapTransaction<>(txContext.transactionId(), updates()));
-    }
-
-    @Override
-    public CompletableFuture<Void> commit() {
-        return backingMap.commit(txContext.transactionId());
-    }
-
-    @Override
-    public CompletableFuture<Void> rollback() {
-        return backingMap.rollback(txContext.transactionId());
-    }
-
-    @Override
-    public CompletableFuture<Boolean> prepareAndCommit() {
-        return backingMap.prepareAndCommit(new MapTransaction<>(txContext.transactionId(), updates()));
-    }
-
-    @Override
-    public int totalUpdates() {
-        return updates().size();
-    }
-
-    @Override
-    public boolean hasPendingUpdates() {
-        return updatesStream().findAny().isPresent();
-    }
-
-    protected Stream<MapUpdate<K, V>> updatesStream() {
-        return Stream.concat(
-            // 1st stream: delete ops
-            deleteSet.stream()
-                .map(key -> Pair.of(key, readCache.get(key)))
-                .filter(e -> e.getValue() != null)
-                .map(e -> MapUpdate.<K, V>newBuilder()
-                 .withMapName(name)
-                 .withType(MapUpdate.Type.REMOVE_IF_VERSION_MATCH)
-                 .withKey(e.getKey())
-                 .withCurrentVersion(e.getValue().version())
-                 .build()),
-            // 2nd stream: write ops
-            writeCache.entrySet().stream()
-                    .map(e -> {
-                        Versioned<V> original = readCache.get(e.getKey());
-                        if (original == null) {
-                            return MapUpdate.<K, V>newBuilder()
-                                    .withMapName(name)
-                                    .withType(MapUpdate.Type.PUT_IF_ABSENT)
-                                    .withKey(e.getKey())
-                                    .withValue(e.getValue())
-                                    .build();
-                        } else {
-                            return MapUpdate.<K, V>newBuilder()
-                                    .withMapName(name)
-                                    .withType(MapUpdate.Type.PUT_IF_VERSION_MATCH)
-                                    .withKey(e.getKey())
-                                    .withCurrentVersion(original.version())
-                                    .withValue(e.getValue())
-                                    .build();
-                        }
-                    }));
-    }
-
-    protected List<MapUpdate<K, V>> updates() {
-        return updatesStream().collect(Collectors.toList());
-    }
-
-    protected List<MapUpdate<String, byte[]>> toMapUpdates() {
-        List<MapUpdate<String, byte[]>> updates = Lists.newLinkedList();
-        deleteSet.forEach(key -> {
-            Versioned<V> original = readCache.get(key);
-            if (original != null) {
-                updates.add(MapUpdate.<String, byte[]>newBuilder()
-                        .withMapName(name)
-                        .withType(MapUpdate.Type.REMOVE_IF_VERSION_MATCH)
-                        .withKey(keyCache.getUnchecked(key))
-                        .withCurrentVersion(original.version())
-                        .build());
-            }
-        });
-        writeCache.forEach((key, value) -> {
-            Versioned<V> original = readCache.get(key);
-            if (original == null) {
-                updates.add(MapUpdate.<String, byte[]>newBuilder()
-                        .withMapName(name)
-                        .withType(MapUpdate.Type.PUT_IF_ABSENT)
-                        .withKey(keyCache.getUnchecked(key))
-                        .withValue(serializer.encode(value))
-                        .build());
-            } else {
-                updates.add(MapUpdate.<String, byte[]>newBuilder()
-                        .withMapName(name)
-                        .withType(MapUpdate.Type.PUT_IF_VERSION_MATCH)
-                        .withKey(keyCache.getUnchecked(key))
-                        .withCurrentVersion(original.version())
-                        .withValue(serializer.encode(value))
-                        .build());
-            }
-        });
-        return updates;
-    }
-
-    @Override
-    public String toString() {
-        return MoreObjects.toStringHelper(this)
-                .add("backingMap", backingMap)
-                .add("updates", updates())
-                .toString();
-    }
-
-    /**
-     * Discards all changes made to this transactional map.
-     */
-    protected void abort() {
-        readCache.clear();
-        writeCache.clear();
-        deleteSet.clear();
-    }
-}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionalMapParticipant.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionalMapParticipant.java
new file mode 100644
index 0000000..9acc377
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionalMapParticipant.java
@@ -0,0 +1,85 @@
+/*
+ * Copyright 2017-present 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.primitives.impl;
+
+import java.util.Map;
+import java.util.stream.Stream;
+
+import com.google.common.collect.Maps;
+import org.apache.commons.lang3.tuple.Pair;
+import org.onosproject.store.primitives.MapUpdate;
+import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.Versioned;
+
+/**
+ * Repeatable read based map participant.
+ */
+public class DefaultTransactionalMapParticipant<K, V> extends TransactionalMapParticipant<K, V> {
+    private final Map<K, Versioned<V>> readCache = Maps.newConcurrentMap();
+
+    public DefaultTransactionalMapParticipant(
+            ConsistentMap<K, V> backingMap, Transaction<MapUpdate<K, V>> transaction) {
+        super(backingMap, transaction);
+    }
+
+    @Override
+    protected V read(K key) {
+        Versioned<V> value = readCache.computeIfAbsent(key, backingMap::get);
+        return value != null ? value.value() : null;
+    }
+
+    @Override
+    protected Stream<MapUpdate<K, V>> records() {
+        return Stream.concat(deleteStream(), writeStream());
+    }
+
+    /**
+     * Returns a transaction record stream for deleted keys.
+     */
+    private Stream<MapUpdate<K, V>> deleteStream() {
+        return deleteSet.stream()
+                .map(key -> Pair.of(key, readCache.get(key)))
+                .filter(e -> e.getValue() != null)
+                .map(e -> MapUpdate.<K, V>newBuilder()
+                        .withType(MapUpdate.Type.REMOVE_IF_VERSION_MATCH)
+                        .withKey(e.getKey())
+                        .withCurrentVersion(e.getValue().version())
+                        .build());
+    }
+
+    /**
+     * Returns a transaction record stream for updated keys.
+     */
+    private Stream<MapUpdate<K, V>> writeStream() {
+        return writeCache.entrySet().stream().map(entry -> {
+            Versioned<V> original = readCache.get(entry.getKey());
+            if (original == null) {
+                return MapUpdate.<K, V>newBuilder()
+                        .withType(MapUpdate.Type.PUT_IF_ABSENT)
+                        .withKey(entry.getKey())
+                        .withValue(entry.getValue())
+                        .build();
+            } else {
+                return MapUpdate.<K, V>newBuilder()
+                        .withType(MapUpdate.Type.PUT_IF_VERSION_MATCH)
+                        .withKey(entry.getKey())
+                        .withCurrentVersion(original.version())
+                        .withValue(entry.getValue())
+                        .build();
+            }
+        });
+    }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DelegatingAsyncConsistentMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DelegatingAsyncConsistentMap.java
index 2f4a665..c0b323a 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DelegatingAsyncConsistentMap.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DelegatingAsyncConsistentMap.java
@@ -29,10 +29,12 @@
 import java.util.function.Predicate;
 
 import org.onosproject.core.ApplicationId;
+import org.onosproject.store.primitives.MapUpdate;
 import org.onosproject.store.primitives.TransactionId;
 import org.onosproject.store.service.AsyncConsistentMap;
 import org.onosproject.store.service.MapEventListener;
-import org.onosproject.store.service.MapTransaction;
+import org.onosproject.store.service.TransactionLog;
+import org.onosproject.store.service.Version;
 import org.onosproject.store.service.Versioned;
 
 import com.google.common.base.MoreObjects;
@@ -170,8 +172,18 @@
     }
 
     @Override
-    public CompletableFuture<Boolean> prepare(MapTransaction<K, V> transaction) {
-        return delegateMap.prepare(transaction);
+    public CompletableFuture<Version> begin(TransactionId transactionId) {
+        return delegateMap.begin(transactionId);
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepare(TransactionLog<MapUpdate<K, V>> transactionLog) {
+        return delegateMap.prepare(transactionLog);
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepareAndCommit(TransactionLog<MapUpdate<K, V>> transactionLog) {
+        return delegateMap.prepareAndCommit(transactionLog);
     }
 
     @Override
@@ -185,11 +197,6 @@
     }
 
     @Override
-    public CompletableFuture<Boolean> prepareAndCommit(MapTransaction<K, V> transaction) {
-        return delegateMap.prepareAndCommit(transaction);
-    }
-
-    @Override
     public void addStatusChangeListener(Consumer<Status> listener) {
         delegateMap.addStatusChangeListener(listener);
     }
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DelegatingAsyncConsistentTreeMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DelegatingAsyncConsistentTreeMap.java
index 48c0b85..f474f7b 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DelegatingAsyncConsistentTreeMap.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DelegatingAsyncConsistentTreeMap.java
@@ -16,10 +16,12 @@
 
 package org.onosproject.store.primitives.impl;
 
+import org.onosproject.store.primitives.MapUpdate;
 import org.onosproject.store.primitives.TransactionId;
 import org.onosproject.store.service.AsyncConsistentTreeMap;
 import org.onosproject.store.service.MapEventListener;
-import org.onosproject.store.service.MapTransaction;
+import org.onosproject.store.service.TransactionLog;
+import org.onosproject.store.service.Version;
 import org.onosproject.store.service.Versioned;
 
 import java.util.Collection;
@@ -253,9 +255,18 @@
     }
 
     @Override
-    public CompletableFuture<Boolean> prepare(
-            MapTransaction<String, V> transaction) {
-        return delegateMap.prepare(transaction);
+    public CompletableFuture<Version> begin(TransactionId transactionId) {
+        return delegateMap.begin(transactionId);
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepare(TransactionLog<MapUpdate<String, V>> transactionLog) {
+        return delegateMap.prepare(transactionLog);
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepareAndCommit(TransactionLog<MapUpdate<String, V>> transactionLog) {
+        return delegateMap.prepareAndCommit(transactionLog);
     }
 
     @Override
@@ -269,12 +280,6 @@
     }
 
     @Override
-    public CompletableFuture<Boolean> prepareAndCommit(
-            MapTransaction<String, V> transaction) {
-        return delegateMap.prepareAndCommit(transaction);
-    }
-
-    @Override
     public boolean equals(Object other) {
         if (other instanceof DelegatingAsyncConsistentTreeMap) {
             DelegatingAsyncConsistentTreeMap<V> that =
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MeteredAsyncConsistentMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MeteredAsyncConsistentMap.java
index f8844fb..ccffead 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MeteredAsyncConsistentMap.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MeteredAsyncConsistentMap.java
@@ -26,11 +26,13 @@
 import java.util.function.Function;
 import java.util.function.Predicate;
 
+import org.onosproject.store.primitives.MapUpdate;
 import org.onosproject.store.primitives.TransactionId;
 import org.onosproject.store.service.AsyncConsistentMap;
 import org.onosproject.store.service.MapEvent;
 import org.onosproject.store.service.MapEventListener;
-import org.onosproject.store.service.MapTransaction;
+import org.onosproject.store.service.TransactionLog;
+import org.onosproject.store.service.Version;
 import org.onosproject.store.service.Versioned;
 
 import com.google.common.base.Throwables;
@@ -63,6 +65,7 @@
     private static final String ENTRY_SET = "entrySet";
     private static final String REPLACE = "replace";
     private static final String COMPUTE_IF_ABSENT = "computeIfAbsent";
+    private static final String BEGIN = "begin";
     private static final String PREPARE = "prepare";
     private static final String COMMIT = "commit";
     private static final String ROLLBACK = "rollback";
@@ -249,10 +252,17 @@
     }
 
     @Override
-    public CompletableFuture<Boolean> prepare(MapTransaction<K, V> transaction) {
+    public CompletableFuture<Version> begin(TransactionId transactionId) {
+        final MeteringAgent.Context timer = monitor.startTimer(BEGIN);
+        return super.begin(transactionId)
+                .whenComplete((r, e) -> timer.stop(e));
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepare(TransactionLog<MapUpdate<K, V>> transactionLog) {
         final MeteringAgent.Context timer = monitor.startTimer(PREPARE);
-        return super.prepare(transaction)
-                    .whenComplete((r, e) -> timer.stop(e));
+        return super.prepare(transactionLog)
+                .whenComplete((r, e) -> timer.stop(e));
     }
 
     @Override
@@ -270,10 +280,10 @@
     }
 
     @Override
-    public CompletableFuture<Boolean> prepareAndCommit(MapTransaction<K, V> transaction) {
+    public CompletableFuture<Boolean> prepareAndCommit(TransactionLog<MapUpdate<K, V>> transactionLog) {
         final MeteringAgent.Context timer = monitor.startTimer(PREPARE_AND_COMMIT);
-        return super.prepareAndCommit(transaction)
-                    .whenComplete((r, e) -> timer.stop(e));
+        return super.prepareAndCommit(transactionLog)
+                .whenComplete((r, e) -> timer.stop(e));
     }
 
     private class InternalMeteredMapEventListener implements MapEventListener<K, V> {
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/PartitionedAsyncConsistentMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/PartitionedAsyncConsistentMap.java
index 9adcb33..bf77105 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/PartitionedAsyncConsistentMap.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/PartitionedAsyncConsistentMap.java
@@ -18,7 +18,6 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import java.util.Collection;
-import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
@@ -37,12 +36,12 @@
 import org.onosproject.store.primitives.TransactionId;
 import org.onosproject.store.service.AsyncConsistentMap;
 import org.onosproject.store.service.MapEventListener;
-import org.onosproject.store.service.MapTransaction;
+import org.onosproject.store.service.TransactionLog;
+import org.onosproject.store.service.Version;
 import org.onosproject.store.service.Versioned;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 
 /**
@@ -200,54 +199,28 @@
     }
 
     @Override
-    public CompletableFuture<Boolean> prepare(MapTransaction<K, V> transaction) {
+    public CompletableFuture<Version> begin(TransactionId transactionId) {
+        throw new UnsupportedOperationException();
+    }
 
-        Map<AsyncConsistentMap<K, V>, List<MapUpdate<K, V>>> updatesGroupedByMap = Maps.newIdentityHashMap();
-        transaction.updates().forEach(update -> {
-            AsyncConsistentMap<K, V> map = getMap(update.key());
-            updatesGroupedByMap.computeIfAbsent(map, k -> Lists.newLinkedList()).add(update);
-        });
-        Map<AsyncConsistentMap<K, V>, MapTransaction<K, V>> transactionsByMap =
-                Maps.transformValues(updatesGroupedByMap,
-                                     list -> new MapTransaction<>(transaction.transactionId(), list));
+    @Override
+    public CompletableFuture<Boolean> prepare(TransactionLog<MapUpdate<K, V>> transactionLog) {
+        throw new UnsupportedOperationException();
+    }
 
-        return Tools.allOf(transactionsByMap.entrySet()
-                         .stream()
-                         .map(e -> e.getKey().prepare(e.getValue()))
-                         .collect(Collectors.toList()))
-                    .thenApply(list -> list.stream().reduce(Boolean::logicalAnd).orElse(true));
+    @Override
+    public CompletableFuture<Boolean> prepareAndCommit(TransactionLog<MapUpdate<K, V>> transactionLog) {
+        throw new UnsupportedOperationException();
     }
 
     @Override
     public CompletableFuture<Void> commit(TransactionId transactionId) {
-        return CompletableFuture.allOf(getMaps().stream()
-                                                .map(e -> e.commit(transactionId))
-                                                .toArray(CompletableFuture[]::new));
+        throw new UnsupportedOperationException();
     }
 
     @Override
     public CompletableFuture<Void> rollback(TransactionId transactionId) {
-        return CompletableFuture.allOf(getMaps().stream()
-                .map(e -> e.rollback(transactionId))
-                .toArray(CompletableFuture[]::new));
-    }
-
-    @Override
-    public CompletableFuture<Boolean> prepareAndCommit(MapTransaction<K, V> transaction) {
-        Map<AsyncConsistentMap<K, V>, List<MapUpdate<K, V>>> updatesGroupedByMap = Maps.newIdentityHashMap();
-        transaction.updates().forEach(update -> {
-            AsyncConsistentMap<K, V> map = getMap(update.key());
-            updatesGroupedByMap.computeIfAbsent(map, k -> Lists.newLinkedList()).add(update);
-        });
-        Map<AsyncConsistentMap<K, V>, MapTransaction<K, V>> transactionsByMap =
-                Maps.transformValues(updatesGroupedByMap,
-                                     list -> new MapTransaction<>(transaction.transactionId(), list));
-
-        return Tools.allOf(transactionsByMap.entrySet()
-                                            .stream()
-                                            .map(e -> e.getKey().prepareAndCommit(e.getValue()))
-                                            .collect(Collectors.toList()))
-                    .thenApply(list -> list.stream().reduce(Boolean::logicalAnd).orElse(true));
+        throw new UnsupportedOperationException();
     }
 
     @Override
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/PartitionedTransactionalMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/PartitionedTransactionalMap.java
new file mode 100644
index 0000000..90b3524
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/PartitionedTransactionalMap.java
@@ -0,0 +1,99 @@
+/*
+ * Copyright 2017-present 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.primitives.impl;
+
+import java.util.Collection;
+import java.util.Map;
+
+import com.google.common.base.MoreObjects;
+import org.onosproject.cluster.PartitionId;
+import org.onosproject.store.service.TransactionalMap;
+
+/**
+ * Partitioned transactional map.
+ */
+public class PartitionedTransactionalMap<K, V> implements TransactionalMap<K, V> {
+    protected final Map<PartitionId, TransactionalMapParticipant<K, V>> partitions;
+    protected final Hasher<K> hasher;
+
+    public PartitionedTransactionalMap(
+            Map<PartitionId, TransactionalMapParticipant<K, V>> partitions, Hasher<K> hasher) {
+        this.partitions = partitions;
+        this.hasher = hasher;
+    }
+
+    /**
+     * Returns the collection of map partitions.
+     *
+     * @return a collection of map partitions
+     */
+    @SuppressWarnings("unchecked")
+    Collection<TransactionParticipant> participants() {
+        return (Collection) partitions.values();
+    }
+
+    /**
+     * Returns the partition for the given key.
+     *
+     * @param key the key for which to return the partition
+     * @return the partition for the given key
+     */
+    private TransactionalMap<K, V> partition(K key) {
+        return partitions.get(hasher.hash(key));
+    }
+
+    @Override
+    public V get(K key) {
+        return partition(key).get(key);
+    }
+
+    @Override
+    public boolean containsKey(K key) {
+        return partition(key).containsKey(key);
+    }
+
+    @Override
+    public V put(K key, V value) {
+        return partition(key).put(key, value);
+    }
+
+    @Override
+    public V remove(K key) {
+        return partition(key).remove(key);
+    }
+
+    @Override
+    public V putIfAbsent(K key, V value) {
+        return partition(key).putIfAbsent(key, value);
+    }
+
+    @Override
+    public boolean remove(K key, V value) {
+        return partition(key).remove(key, value);
+    }
+
+    @Override
+    public boolean replace(K key, V oldValue, V newValue) {
+        return partition(key).replace(key, oldValue, newValue);
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(this)
+                .add("partitions", partitions.values())
+                .toString();
+    }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StorageManager.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StorageManager.java
index 48d9b7e..1fc821d 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StorageManager.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StorageManager.java
@@ -42,7 +42,6 @@
 import org.onosproject.store.primitives.TransactionId;
 import org.onosproject.store.serializers.KryoNamespaces;
 import org.onosproject.store.service.AsyncAtomicValue;
-import org.onosproject.store.service.AsyncConsistentMap;
 import org.onosproject.store.service.AsyncDocumentTree;
 import org.onosproject.store.service.AsyncConsistentMultimap;
 import org.onosproject.store.service.AsyncConsistentTreeMap;
@@ -69,7 +68,6 @@
 import org.slf4j.Logger;
 
 import com.google.common.collect.Maps;
-import com.google.common.util.concurrent.Futures;
 
 /**
  * Implementation for {@code StorageService} and {@code StorageAdminService}.
@@ -98,8 +96,7 @@
     private final Supplier<TransactionId> transactionIdGenerator =
             () -> TransactionId.from(UUID.randomUUID().toString());
     private DistributedPrimitiveCreator federatedPrimitiveCreator;
-    private AsyncConsistentMap<TransactionId, Transaction.State> transactions;
-    private TransactionCoordinator transactionCoordinator;
+    private TransactionManager transactionManager;
 
     @Activate
     public void activate() {
@@ -108,13 +105,7 @@
             .filter(id -> !id.equals(PartitionId.from(0)))
             .forEach(id -> partitionMap.put(id, partitionService.getDistributedPrimitiveCreator(id)));
         federatedPrimitiveCreator = new FederatedDistributedPrimitiveCreator(partitionMap);
-        transactions = this.<TransactionId, Transaction.State>consistentMapBuilder()
-                    .withName("onos-transactions")
-                    .withSerializer(Serializer.using(KryoNamespaces.API,
-                            Transaction.class,
-                            Transaction.State.class))
-                    .buildAsyncMap();
-        transactionCoordinator = new TransactionCoordinator(transactions);
+        transactionManager = new TransactionManager(this, partitionService);
         log.info("Started");
     }
 
@@ -187,9 +178,7 @@
     @Override
     public TransactionContextBuilder transactionContextBuilder() {
         checkPermission(STORAGE_WRITE);
-        return new DefaultTransactionContextBuilder(transactionIdGenerator.get(),
-                federatedPrimitiveCreator,
-                transactionCoordinator);
+        return new DefaultTransactionContextBuilder(transactionIdGenerator.get(), transactionManager);
     }
 
     @Override
@@ -259,7 +248,7 @@
 
     @Override
     public Collection<TransactionId> getPendingTransactions() {
-        return Futures.getUnchecked(transactions.keySet());
+        return transactionManager.getPendingTransactions();
     }
 
     private List<MapInfo> listMapInfo(DistributedPrimitiveCreator creator) {
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Transaction.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Transaction.java
index be5f05a..82b3919 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Transaction.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Transaction.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-present Open Networking Laboratory
+ * Copyright 2017-present 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.
@@ -16,88 +16,263 @@
 package org.onosproject.store.primitives.impl;
 
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.onosproject.store.primitives.MapUpdate;
 import org.onosproject.store.primitives.TransactionId;
+import org.onosproject.store.service.TransactionLog;
+import org.onosproject.store.service.Transactional;
+import org.onosproject.store.service.Version;
+import org.onosproject.store.service.TransactionContext;
+import org.onosproject.store.service.TransactionException;
 
-import com.google.common.base.MoreObjects;
-import com.google.common.collect.ImmutableList;
+import static com.google.common.base.MoreObjects.toStringHelper;
+import static com.google.common.base.Preconditions.checkState;
 
 /**
- * An immutable transaction object.
+ * Manages a transaction within the context of a single primitive.
+ * <p>
+ * The {@code Transaction} object is used to manage the transaction for a single partition primitive that implements
+ * the {@link Transactional} interface. It's used as a proxy for {@link TransactionContext}s to manage the transaction
+ * as it relates to a single piece of atomic state.
  */
-public class Transaction {
+public class Transaction<T> {
 
+    /**
+     * Transaction state.
+     * <p>
+     * The transaction state is used to indicate the phase within which the transaction is currently running.
+     */
     enum State {
+
         /**
-         * Indicates a new transaction that is about to be prepared. All transactions
-         * start their life in this state.
+         * Active transaction state.
+         * <p>
+         * The {@code ACTIVE} state represents a transaction in progress. Active transactions may or may not affect
+         * concurrently running transactions depending on the transaction's isolation level.
+         */
+        ACTIVE,
+
+        /**
+         * Preparing transaction state.
+         * <p>
+         * Once a transaction commitment begins, it enters the {@code PREPARING} phase of the two-phase commit protocol.
          */
         PREPARING,
 
         /**
-         * Indicates a transaction that is successfully prepared i.e. all participants voted to commit
+         * Prepared transaction state.
+         * <p>
+         * Once the first phase of the two-phase commit protocol is complete, the transaction's state is set to
+         * {@code PREPARED}.
          */
         PREPARED,
 
         /**
-         * Indicates a transaction that is about to be committed.
+         * Committing transaction state.
+         * <p>
+         * The {@code COMMITTING} state represents a transaction within the second phase of the two-phase commit
+         * protocol.
          */
         COMMITTING,
 
         /**
-         * Indicates a transaction that has successfully committed.
+         * Committed transaction state.
+         * <p>
+         * Once the second phase of the two-phase commit protocol is complete, the transaction's state is set to
+         * {@code COMMITTED}.
          */
         COMMITTED,
 
         /**
-         * Indicates a transaction that is about to be rolled back.
+         * Rolling back transaction state.
+         * <p>
+         * In the event of a two-phase lock failure, when the transaction is rolled back it will enter the
+         * {@code ROLLING_BACK} state while the rollback is in progress.
          */
-        ROLLINGBACK,
+        ROLLING_BACK,
 
         /**
-         * Indicates a transaction that has been rolled back and all locks are released.
+         * Rolled back transaction state.
+         * <p>
+         * Once a transaction has been rolled back, it will enter the {@code ROLLED_BACK} state.
          */
-        ROLLEDBACK
+        ROLLED_BACK,
     }
 
-    private final TransactionId transactionId;
-    private final List<MapUpdate<String, byte[]>> updates;
-    private final State state;
+    private static final String TX_OPEN_ERROR = "transaction already open";
+    private static final String TX_CLOSED_ERROR = "transaction not open";
+    private static final String TX_INACTIVE_ERROR = "transaction is not active";
+    private static final String TX_UNPREPARED_ERROR = "transaction has not been prepared";
 
-    public Transaction(TransactionId transactionId, List<MapUpdate<String, byte[]>> updates) {
-        this(transactionId, updates, State.PREPARING);
-    }
+    protected final TransactionId transactionId;
+    protected final Transactional<T> transactionalObject;
+    private final AtomicBoolean open = new AtomicBoolean();
+    private volatile State state = State.ACTIVE;
 
-    private Transaction(TransactionId transactionId,
-            List<MapUpdate<String, byte[]>> updates,
-            State state) {
+    public Transaction(TransactionId transactionId, Transactional<T> transactionalObject) {
         this.transactionId = transactionId;
-        this.updates = ImmutableList.copyOf(updates);
-        this.state = state;
+        this.transactionalObject = transactionalObject;
     }
 
-    public TransactionId id() {
+    /**
+     * Returns the transaction identifier.
+     *
+     * @return the transaction identifier
+     */
+    public TransactionId transactionId() {
         return transactionId;
     }
 
-    public List<MapUpdate<String, byte[]>> updates() {
-        return updates;
-    }
-
+    /**
+     * Returns the current transaction state.
+     *
+     * @return the current transaction state
+     */
     public State state() {
         return state;
     }
 
-    public Transaction transition(State newState) {
-        return new Transaction(transactionId, updates, newState);
+    /**
+     * Returns a boolean indicating whether the transaction is open.
+     *
+     * @return indicates whether the transaction is open
+     */
+    public boolean isOpen() {
+        return open.get();
+    }
+
+    /**
+     * Opens the transaction, throwing an {@link IllegalStateException} if it's already open.
+     */
+    protected void open() {
+        if (!open.compareAndSet(false, true)) {
+            throw new IllegalStateException(TX_OPEN_ERROR);
+        }
+    }
+
+    /**
+     * Checks that the transaction is open and throws an {@link IllegalStateException} if not.
+     */
+    protected void checkOpen() {
+        checkState(isOpen(), TX_CLOSED_ERROR);
+    }
+
+    /**
+     * Checks that the transaction state is {@code ACTIVE} and throws an {@link IllegalStateException} if not.
+     */
+    protected void checkActive() {
+        checkState(state == State.ACTIVE, TX_INACTIVE_ERROR);
+    }
+
+    /**
+     * Checks that the transaction state is {@code PREPARED} and throws an {@link IllegalStateException} if not.
+     */
+    protected void checkPrepared() {
+        checkState(state == State.PREPARED, TX_UNPREPARED_ERROR);
+    }
+
+    /**
+     * Updates the transaction state.
+     *
+     * @param state the updated transaction state
+     */
+    protected void setState(State state) {
+        this.state = state;
+    }
+
+    /**
+     * Begins the transaction.
+     * <p>
+     * Locks are acquired when the transaction is begun to prevent concurrent transactions from operating on the shared
+     * resource to which this transaction relates.
+     *
+     * @return a completable future to be completed once the transaction has been started
+     */
+    public CompletableFuture<Version> begin() {
+        open();
+        return transactionalObject.begin(transactionId);
+    }
+
+    /**
+     * Prepares the transaction.
+     * <p>
+     * When preparing the transaction, the given list of updates for the shared resource will be prepared, and
+     * concurrent modification checks will be performed. The returned future may be completed with a
+     * {@link TransactionException} if a concurrent modification is detected for an isolation level that does
+     * not allow such modifications.
+     *
+     * @param updates the transaction updates
+     * @return a completable future to be completed once the transaction has been prepared
+     */
+    public CompletableFuture<Boolean> prepare(List<T> updates) {
+        checkOpen();
+        checkActive();
+        setState(State.PREPARING);
+        return transactionalObject.prepare(new TransactionLog<T>(transactionId, updates))
+                .thenApply(succeeded -> {
+                    setState(State.PREPARED);
+                    return succeeded;
+                });
+    }
+
+    /**
+     * Prepares and commits the transaction in a single atomic operation.
+     * <p>
+     * Both the prepare and commit phases of the protocol must be executed within a single atomic operation. This method
+     * is used to optimize committing transactions that operate only on a single partition within a single primitive.
+     *
+     * @param updates the transaction updates
+     * @return a completable future to be completed once the transaction has been prepared
+     */
+    public CompletableFuture<Boolean> prepareAndCommit(List<T> updates) {
+        checkOpen();
+        checkActive();
+        setState(State.PREPARING);
+        return transactionalObject.prepareAndCommit(new TransactionLog<T>(transactionId, updates))
+                .thenApply(succeeded -> {
+                    setState(State.COMMITTED);
+                    return succeeded;
+                });
+    }
+
+    /**
+     * Commits the transaction.
+     * <p>
+     * Performs the second phase of the two-phase commit protocol, committing the previously
+     * {@link #prepare(List) prepared} updates.
+     *
+     * @return a completable future to be completed once the transaction has been committed
+     */
+    public CompletableFuture<Void> commit() {
+        checkOpen();
+        checkPrepared();
+        setState(State.COMMITTING);
+        return transactionalObject.commit(transactionId).thenRun(() -> {
+            setState(State.COMMITTED);
+        });
+    }
+
+    /**
+     * Rolls back the transaction.
+     * <p>
+     * Rolls back the first phase of the two-phase commit protocol, cancelling prepared updates.
+     *
+     * @return a completable future to be completed once the transaction has been rolled back
+     */
+    public CompletableFuture<Void> rollback() {
+        checkOpen();
+        checkPrepared();
+        setState(State.ROLLING_BACK);
+        return transactionalObject.rollback(transactionId).thenRun(() -> {
+            setState(State.ROLLED_BACK);
+        });
     }
 
     @Override
     public String toString() {
-        return MoreObjects.toStringHelper(getClass())
+        return toStringHelper(this)
                 .add("transactionId", transactionId)
-                .add("updates", updates)
                 .add("state", state)
                 .toString();
     }
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionCoordinator.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionCoordinator.java
index 160ecf9..9826266 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionCoordinator.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionCoordinator.java
@@ -19,78 +19,123 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
 
+import com.google.common.collect.Sets;
 import org.onlab.util.Tools;
 import org.onosproject.store.primitives.TransactionId;
-import org.onosproject.store.service.AsyncConsistentMap;
 import org.onosproject.store.service.CommitStatus;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.TransactionalMap;
+
+import static com.google.common.base.MoreObjects.toStringHelper;
 
 /**
- * Coordinator for a two-phase commit protocol.
+ * Transaction coordinator.
  */
 public class TransactionCoordinator {
+    protected final TransactionId transactionId;
+    protected final TransactionManager transactionManager;
+    protected final Set<TransactionParticipant> transactionParticipants = Sets.newConcurrentHashSet();
 
-    private final AsyncConsistentMap<TransactionId, Transaction.State> transactions;
-
-    public TransactionCoordinator(AsyncConsistentMap<TransactionId, Transaction.State> transactions) {
-        this.transactions = transactions;
+    public TransactionCoordinator(TransactionId transactionId, TransactionManager transactionManager) {
+        this.transactionId = transactionId;
+        this.transactionManager = transactionManager;
     }
 
     /**
-     * Commits a transaction.
+     * Returns a transactional map for this transaction.
      *
-     * @param transactionId transaction identifier
-     * @param transactionParticipants set of transaction participants
-     * @return future for commit result
+     * @param name the transactional map name
+     * @param serializer the serializer
+     * @param <K> key type
+     * @param <V> value type
+     * @return a transactional map for this transaction
      */
-    CompletableFuture<CommitStatus> commit(TransactionId transactionId,
-                                           Set<TransactionParticipant> transactionParticipants) {
-        int totalUpdates = transactionParticipants.stream()
-                                                  .map(TransactionParticipant::totalUpdates)
-                                                  .reduce(Math::addExact)
-                                                  .orElse(0);
+    public <K, V> TransactionalMap<K, V> getTransactionalMap(String name, Serializer serializer) {
+        PartitionedTransactionalMap<K, V> map = transactionManager.getTransactionalMap(name, serializer, this);
+        transactionParticipants.addAll(map.participants());
+        return map;
+    }
 
-        if (totalUpdates == 0) {
+    /**
+     * Commits the transaction.
+     *
+     * @return the transaction commit status
+     */
+    public CompletableFuture<CommitStatus> commit() {
+        long totalParticipants = transactionParticipants.stream()
+                .filter(TransactionParticipant::hasPendingUpdates)
+                .count();
+
+        if (totalParticipants == 0) {
             return CompletableFuture.completedFuture(CommitStatus.SUCCESS);
-        } else if (totalUpdates == 1) {
+        } else if (totalParticipants == 1) {
             return transactionParticipants.stream()
-                                          .filter(p -> p.totalUpdates() == 1)
-                                          .findFirst()
-                                          .get()
-                                          .prepareAndCommit()
-                                          .thenApply(v -> v ? CommitStatus.SUCCESS : CommitStatus.FAILURE);
+                    .filter(TransactionParticipant::hasPendingUpdates)
+                    .findFirst()
+                    .get()
+                    .prepareAndCommit()
+                    .thenApply(v -> v ? CommitStatus.SUCCESS : CommitStatus.FAILURE);
         } else {
-            CompletableFuture<CommitStatus> status =  transactions.put(transactionId, Transaction.State.PREPARING)
-                    .thenCompose(v -> this.doPrepare(transactionParticipants))
+            Set<TransactionParticipant> transactionParticipants = this.transactionParticipants.stream()
+                    .filter(TransactionParticipant::hasPendingUpdates)
+                    .collect(Collectors.toSet());
+
+            CompletableFuture<CommitStatus> status = transactionManager.updateState(
+                    transactionId, Transaction.State.PREPARING)
+                    .thenCompose(v -> prepare(transactionParticipants))
                     .thenCompose(result -> result
-                            ? transactions.put(transactionId, Transaction.State.COMMITTING)
-                                          .thenCompose(v -> doCommit(transactionParticipants))
-                                          .thenApply(v -> CommitStatus.SUCCESS)
-                            : transactions.put(transactionId, Transaction.State.ROLLINGBACK)
-                                          .thenCompose(v -> doRollback(transactionParticipants))
-                                          .thenApply(v -> CommitStatus.FAILURE));
-            return status.thenCompose(v -> transactions.remove(transactionId).thenApply(u -> v));
+                            ? transactionManager.updateState(transactionId, Transaction.State.COMMITTING)
+                            .thenCompose(v -> commit(transactionParticipants))
+                            .thenApply(v -> CommitStatus.SUCCESS)
+                            : transactionManager.updateState(transactionId, Transaction.State.ROLLING_BACK)
+                            .thenCompose(v -> rollback(transactionParticipants))
+                            .thenApply(v -> CommitStatus.FAILURE));
+            return status.thenCompose(v -> transactionManager.remove(transactionId).thenApply(u -> v));
         }
     }
 
-    private CompletableFuture<Boolean> doPrepare(Set<TransactionParticipant> transactionParticipants) {
+    /**
+     * Performs the prepare phase of the two-phase commit protocol for the given transaction participants.
+     *
+     * @param transactionParticipants the transaction participants for which to prepare the transaction
+     * @return a completable future indicating whether <em>all</em> prepares succeeded
+     */
+    protected CompletableFuture<Boolean> prepare(Set<TransactionParticipant> transactionParticipants) {
         return Tools.allOf(transactionParticipants.stream()
-                                                  .filter(TransactionParticipant::hasPendingUpdates)
-                                                  .map(TransactionParticipant::prepare)
-                                                  .collect(Collectors.toList()))
-                    .thenApply(list -> list.stream().reduce(Boolean::logicalAnd).orElse(true));
+                .map(TransactionParticipant::prepare)
+                .collect(Collectors.toList()))
+                .thenApply(list -> list.stream().reduce(Boolean::logicalAnd).orElse(true));
     }
 
-    private CompletableFuture<Void> doCommit(Set<TransactionParticipant> transactionParticipants) {
+    /**
+     * Performs the commit phase of the two-phase commit protocol for the given transaction participants.
+     *
+     * @param transactionParticipants the transaction participants for which to commit the transaction
+     * @return a completable future to be completed once the commits are complete
+     */
+    protected CompletableFuture<Void> commit(Set<TransactionParticipant> transactionParticipants) {
         return CompletableFuture.allOf(transactionParticipants.stream()
-                                                              .filter(TransactionParticipant::hasPendingUpdates)
-                                                              .map(TransactionParticipant::commit)
-                                                              .toArray(CompletableFuture[]::new));
+                .map(TransactionParticipant::commit)
+                .toArray(CompletableFuture[]::new));
     }
 
-    private CompletableFuture<Void> doRollback(Set<TransactionParticipant> transactionParticipants) {
+    /**
+     * Rolls back transactions for the given participants.
+     *
+     * @param transactionParticipants the transaction participants for which to roll back the transaction
+     * @return a completable future to be completed once the rollbacks are complete
+     */
+    protected CompletableFuture<Void> rollback(Set<TransactionParticipant> transactionParticipants) {
         return CompletableFuture.allOf(transactionParticipants.stream()
-                                                              .filter(TransactionParticipant::hasPendingUpdates)
-                                                              .map(TransactionParticipant::rollback)
-                                                              .toArray(CompletableFuture[]::new));
+                .map(TransactionParticipant::rollback)
+                .toArray(CompletableFuture[]::new));
+    }
+
+    @Override
+    public String toString() {
+        return toStringHelper(this)
+                .add("transactionId", transactionId)
+                .add("participants", transactionParticipants)
+                .toString();
     }
 }
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionManager.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionManager.java
new file mode 100644
index 0000000..5c246de
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionManager.java
@@ -0,0 +1,153 @@
+/*
+ * Copyright 2017-present 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.primitives.impl;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.hash.Hashing;
+import com.google.common.util.concurrent.Futures;
+import org.onosproject.cluster.PartitionId;
+import org.onosproject.store.primitives.MapUpdate;
+import org.onosproject.store.primitives.PartitionService;
+import org.onosproject.store.primitives.TransactionId;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.AsyncConsistentMap;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.TransactionException;
+
+/**
+ * Transaction manager for managing state shared across multiple transactions.
+ */
+public class TransactionManager {
+    private static final int DEFAULT_CACHE_SIZE = 100;
+
+    private final PartitionService partitionService;
+    private final List<PartitionId> sortedPartitions;
+    private final AsyncConsistentMap<TransactionId, Transaction.State> transactions;
+    private final int cacheSize;
+    private final Map<PartitionId, Cache<String, AsyncConsistentMap>> partitionCache = Maps.newConcurrentMap();
+
+    public TransactionManager(StorageService storageService, PartitionService partitionService) {
+        this(storageService, partitionService, DEFAULT_CACHE_SIZE);
+    }
+
+    public TransactionManager(StorageService storageService, PartitionService partitionService, int cacheSize) {
+        this.partitionService = partitionService;
+        this.cacheSize = cacheSize;
+        this.transactions = storageService.<TransactionId, Transaction.State>consistentMapBuilder()
+                .withName("onos-transactions")
+                .withSerializer(Serializer.using(KryoNamespaces.API,
+                        Transaction.class,
+                        Transaction.State.class))
+                .buildAsyncMap();
+        this.sortedPartitions = Lists.newArrayList(partitionService.getAllPartitionIds());
+        Collections.sort(sortedPartitions);
+    }
+
+    /**
+     * Returns the collection of currently pending transactions.
+     *
+     * @return a collection of currently pending transactions
+     */
+    public Collection<TransactionId> getPendingTransactions() {
+        return Futures.getUnchecked(transactions.keySet());
+    }
+
+    /**
+     * Returns a partitioned transactional map for use within a transaction context.
+     * <p>
+     * The transaction coordinator will return a map that takes advantage of caching that's shared across transaction
+     * contexts.
+     *
+     * @param name the map name
+     * @param serializer the map serializer
+     * @param transactionCoordinator the transaction coordinator for which the map is being created
+     * @param <K> key type
+     * @param <V> value type
+     * @return a partitioned transactional map
+     */
+    <K, V> PartitionedTransactionalMap<K, V> getTransactionalMap(
+            String name,
+            Serializer serializer,
+            TransactionCoordinator transactionCoordinator) {
+        Map<PartitionId, TransactionalMapParticipant<K, V>> partitions = new HashMap<>();
+        for (PartitionId partitionId : partitionService.getAllPartitionIds()) {
+            partitions.put(partitionId, getTransactionalMapPartition(
+                    name, partitionId, serializer, transactionCoordinator));
+        }
+
+        Hasher<K> hasher = key -> {
+            int hashCode = Hashing.sha256().hashBytes(serializer.encode(key)).asInt();
+            return sortedPartitions.get(Math.abs(hashCode) % sortedPartitions.size());
+        };
+        return new PartitionedTransactionalMap<>(partitions, hasher);
+    }
+
+    @SuppressWarnings("unchecked")
+    private <K, V> TransactionalMapParticipant<K, V> getTransactionalMapPartition(
+            String mapName,
+            PartitionId partitionId,
+            Serializer serializer,
+            TransactionCoordinator transactionCoordinator) {
+        Cache<String, AsyncConsistentMap> mapCache = partitionCache.computeIfAbsent(partitionId, p ->
+                CacheBuilder.newBuilder().maximumSize(cacheSize / partitionService.getNumberOfPartitions()).build());
+        try {
+            AsyncConsistentMap<K, V> baseMap = partitionService.getDistributedPrimitiveCreator(partitionId)
+                            .newAsyncConsistentMap(mapName, serializer);
+            AsyncConsistentMap<K, V> asyncMap = mapCache.get(mapName, () ->
+                    DistributedPrimitives.newCachingMap(baseMap));
+
+            Transaction<MapUpdate<K, V>> transaction = new Transaction<>(
+                    transactionCoordinator.transactionId,
+                    baseMap);
+            return new DefaultTransactionalMapParticipant<>(asyncMap.asConsistentMap(), transaction);
+        } catch (ExecutionException e) {
+            throw new TransactionException(e);
+        }
+    }
+
+    /**
+     * Updates the state of a transaction in the transaction registry.
+     *
+     * @param transactionId the transaction identifier
+     * @param state the state of the transaction
+     * @return a completable future to be completed once the transaction state has been updated in the registry
+     */
+    CompletableFuture<Void> updateState(TransactionId transactionId, Transaction.State state) {
+        return transactions.put(transactionId, state).thenApply(v -> null);
+    }
+
+    /**
+     * Removes the given transaction from the transaction registry.
+     *
+     * @param transactionId the transaction identifier
+     * @return a completable future to be completed once the transaction state has been removed from the registry
+     */
+    CompletableFuture<Void> remove(TransactionId transactionId) {
+        return transactions.remove(transactionId).thenApply(v -> null);
+    }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionParticipant.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionParticipant.java
index 0780bf6..f97d448 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionParticipant.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionParticipant.java
@@ -23,40 +23,38 @@
 public interface TransactionParticipant {
 
     /**
-     * Returns if this participant has updates that need to be committed.
-     * @return {@code true} if yes; {@code false} otherwise
+     * Returns a boolean indicating whether the participant has pending updates.
+     *
+     * @return indicates whether the participant has pending updates
      */
-    default boolean hasPendingUpdates() {
-        return totalUpdates() > 0;
-    }
-
-    /**
-     * Returns the number of updates that need to committed for this participant.
-     * @return update count.
-     */
-    int totalUpdates();
-
-    /**
-     * Executes the prepare and commit steps in a single go.
-     * @return {@code true} is successful i.e updates are committed; {@code false} otherwise
-     */
-    CompletableFuture<Boolean> prepareAndCommit();
+    boolean hasPendingUpdates();
 
     /**
      * Executes the prepare phase.
+     *
      * @return {@code true} is successful; {@code false} otherwise
      */
     CompletableFuture<Boolean> prepare();
 
     /**
      * Attempts to execute the commit phase for previously prepared transaction.
+     *
      * @return future that is completed when the operation completes
      */
     CompletableFuture<Void> commit();
 
     /**
+     * Executes the prepare and commit phases atomically.
+     *
+     * @return {@code true} is successful; {@code false} otherwise
+     */
+    CompletableFuture<Boolean> prepareAndCommit();
+
+    /**
      * Attempts to execute the rollback phase for previously prepared transaction.
+     *
      * @return future that is completed when the operation completes
      */
     CompletableFuture<Void> rollback();
+
 }
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionalMapParticipant.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionalMapParticipant.java
new file mode 100644
index 0000000..4429a1b
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionalMapParticipant.java
@@ -0,0 +1,233 @@
+/*
+ * Copyright 2017-present 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.primitives.impl;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import com.google.common.base.MoreObjects;
+import com.google.common.base.Objects;
+import com.google.common.base.Throwables;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.onosproject.store.primitives.MapUpdate;
+import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.DistributedPrimitive;
+import org.onosproject.store.service.TransactionException;
+import org.onosproject.store.service.TransactionalMap;
+import org.onosproject.store.service.Version;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+
+/**
+ * Base class for participants within a single {@link TransactionalMap}.
+ * <p>
+ * This class provides the basic functionality required by transactional map participants and provides methods
+ * for defining operations specific to individual isolation levels.
+ *
+ * @param <K> key type
+ * @param <V> value type.
+ */
+public abstract class TransactionalMapParticipant<K, V> implements TransactionalMap<K, V>, TransactionParticipant {
+    private static final String TX_CLOSED_ERROR = "Transaction is closed";
+    private static final String ERROR_NULL_VALUE = "Null values are not allowed";
+    private static final String ERROR_NULL_KEY = "Null key is not allowed";
+
+    protected final ConsistentMap<K, V> backingMap;
+    protected final Transaction<MapUpdate<K, V>> transaction;
+    protected final Map<K, V> writeCache = Maps.newConcurrentMap();
+    protected final Set<K> deleteSet = Sets.newConcurrentHashSet();
+    protected final List<MapUpdate<K, V>> log = new ArrayList<>();
+    protected volatile Version lock;
+
+    protected TransactionalMapParticipant(
+            ConsistentMap<K, V> backingMap,
+            Transaction<MapUpdate<K, V>> transaction) {
+        this.backingMap = backingMap;
+        this.transaction = transaction;
+    }
+
+    /**
+     * Starts the transaction for this partition when a read occurs.
+     * <p>
+     * Acquiring a pessimistic lock at the start of the transaction ensures that underlying cached maps have been
+     * synchronized prior to a read.
+     */
+    private void beginTransaction() {
+        if (!transaction.isOpen()) {
+            try {
+                lock = transaction.begin()
+                        .get(DistributedPrimitive.DEFAULT_OPERTATION_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new TransactionException.Interrupted();
+            } catch (TimeoutException e) {
+                throw new TransactionException.Timeout();
+            } catch (ExecutionException e) {
+                Throwables.propagateIfPossible(e.getCause());
+                throw new TransactionException(e.getCause());
+            }
+        }
+    }
+
+    @Override
+    public V get(K key) {
+        // Start the transaction for this primitive/partition if necessary.
+        beginTransaction();
+
+        checkState(transaction.isOpen(), TX_CLOSED_ERROR);
+        checkNotNull(key, ERROR_NULL_KEY);
+
+        if (deleteSet.contains(key)) {
+            return null;
+        }
+
+        V latest = writeCache.get(key);
+        if (latest != null) {
+            return latest;
+        } else {
+            return read(key);
+        }
+    }
+
+    /**
+     * Executes a get operation based on the transaction isolation level.
+     *
+     * @param key the key to look up
+     * @return the value
+     */
+    protected abstract V read(K key);
+
+    @Override
+    public boolean containsKey(K key) {
+        return get(key) != null;
+    }
+
+    @Override
+    public V put(K key, V value) {
+        checkNotNull(value, ERROR_NULL_VALUE);
+
+        V latest = get(key);
+        writeCache.put(key, value);
+        deleteSet.remove(key);
+        return latest;
+    }
+
+    @Override
+    public V remove(K key) {
+        V latest = get(key);
+        if (latest != null) {
+            writeCache.remove(key);
+            deleteSet.add(key);
+        }
+        return latest;
+    }
+
+    @Override
+    public boolean remove(K key, V value) {
+        checkNotNull(value, ERROR_NULL_VALUE);
+
+        V latest = get(key);
+        if (Objects.equal(value, latest)) {
+            remove(key);
+            return true;
+        }
+        return false;
+    }
+
+    @Override
+    public boolean replace(K key, V oldValue, V newValue) {
+        checkNotNull(oldValue, ERROR_NULL_VALUE);
+        checkNotNull(newValue, ERROR_NULL_VALUE);
+
+        V latest = get(key);
+        if (Objects.equal(oldValue, latest)) {
+            put(key, newValue);
+            return true;
+        }
+        return false;
+    }
+
+    @Override
+    public V putIfAbsent(K key, V value) {
+        checkNotNull(value, ERROR_NULL_VALUE);
+
+        V latest = get(key);
+        if (latest == null) {
+            put(key, value);
+        }
+        return latest;
+    }
+
+    @Override
+    public boolean hasPendingUpdates() {
+        return records().findAny().isPresent();
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepare() {
+        return transaction.prepare(log());
+    }
+
+    @Override
+    public CompletableFuture<Void> commit() {
+        return transaction.commit();
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepareAndCommit() {
+        return transaction.prepareAndCommit(log());
+    }
+
+    @Override
+    public CompletableFuture<Void> rollback() {
+        return transaction.rollback();
+    }
+
+    /**
+     * Returns a list of updates performed within this map partition.
+     *
+     * @return a list of map updates
+     */
+    protected List<MapUpdate<K, V>> log() {
+        return records().collect(Collectors.toList());
+    }
+
+    /**
+     * Returns a stream of updates performed within this map partition.
+     *
+     * @return a stream of map updates
+     */
+    protected abstract Stream<MapUpdate<K, V>> records();
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(this)
+                .add("backingMap", backingMap)
+                .add("updates", log())
+                .toString();
+    }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TranscodingAsyncConsistentMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TranscodingAsyncConsistentMap.java
index 2afb5df..c2f167a 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TranscodingAsyncConsistentMap.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TranscodingAsyncConsistentMap.java
@@ -29,11 +29,13 @@
 import java.util.stream.Collectors;
 
 import org.onlab.util.Tools;
+import org.onosproject.store.primitives.MapUpdate;
 import org.onosproject.store.primitives.TransactionId;
 import org.onosproject.store.service.AsyncConsistentMap;
 import org.onosproject.store.service.MapEvent;
 import org.onosproject.store.service.MapEventListener;
-import org.onosproject.store.service.MapTransaction;
+import org.onosproject.store.service.TransactionLog;
+import org.onosproject.store.service.Version;
 import org.onosproject.store.service.Versioned;
 
 import com.google.common.collect.Maps;
@@ -266,9 +268,27 @@
     }
 
     @Override
-    public CompletableFuture<Boolean> prepare(MapTransaction<K1, V1> transaction) {
+    public CompletableFuture<Version> begin(TransactionId transactionId) {
         try {
-            return backingMap.prepare(transaction.map(keyEncoder, valueEncoder));
+            return backingMap.begin(transactionId);
+        } catch (Exception e) {
+            return Tools.exceptionalFuture(e);
+        }
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepare(TransactionLog<MapUpdate<K1, V1>> transactionLog) {
+        try {
+            return backingMap.prepare(transactionLog.map(record -> record.map(keyEncoder, valueEncoder)));
+        } catch (Exception e) {
+            return Tools.exceptionalFuture(e);
+        }
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepareAndCommit(TransactionLog<MapUpdate<K1, V1>> transactionLog) {
+        try {
+            return backingMap.prepareAndCommit(transactionLog.map(record -> record.map(keyEncoder, valueEncoder)));
         } catch (Exception e) {
             return Tools.exceptionalFuture(e);
         }
@@ -276,18 +296,17 @@
 
     @Override
     public CompletableFuture<Void> commit(TransactionId transactionId) {
-        return backingMap.commit(transactionId);
+        try {
+            return backingMap.commit(transactionId);
+        } catch (Exception e) {
+            return Tools.exceptionalFuture(e);
+        }
     }
 
     @Override
     public CompletableFuture<Void> rollback(TransactionId transactionId) {
-        return backingMap.rollback(transactionId);
-    }
-
-    @Override
-    public CompletableFuture<Boolean> prepareAndCommit(MapTransaction<K1, V1> transaction) {
         try {
-            return backingMap.prepareAndCommit(transaction.map(keyEncoder, valueEncoder));
+            return backingMap.rollback(transactionId);
         } catch (Exception e) {
             return Tools.exceptionalFuture(e);
         }
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TranscodingAsyncConsistentTreeMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TranscodingAsyncConsistentTreeMap.java
index 747008f..2f0683e 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TranscodingAsyncConsistentTreeMap.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TranscodingAsyncConsistentTreeMap.java
@@ -18,11 +18,13 @@
 
 import com.google.common.collect.Maps;
 import org.onlab.util.Tools;
+import org.onosproject.store.primitives.MapUpdate;
 import org.onosproject.store.primitives.TransactionId;
 import org.onosproject.store.service.AsyncConsistentTreeMap;
 import org.onosproject.store.service.MapEvent;
 import org.onosproject.store.service.MapEventListener;
-import org.onosproject.store.service.MapTransaction;
+import org.onosproject.store.service.TransactionLog;
+import org.onosproject.store.service.Version;
 import org.onosproject.store.service.Versioned;
 
 import java.util.Collection;
@@ -361,27 +363,30 @@
     }
 
     @Override
-    public CompletableFuture<Boolean> prepare(
-            MapTransaction<String, V1> transaction) {
-        throw new UnsupportedOperationException("This operation is not yet " +
-                                                        "supported.");
+    public CompletableFuture<Version> begin(TransactionId transactionId) {
+        throw new UnsupportedOperationException("This operation is not yet supported.");
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepare(TransactionLog<MapUpdate<String, V1>> transactionLog) {
+        throw new UnsupportedOperationException("This operation is not yet supported.");
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepareAndCommit(TransactionLog<MapUpdate<String, V1>> transactionLog) {
+        throw new UnsupportedOperationException("This operation is not yet supported.");
     }
 
     @Override
     public CompletableFuture<Void> commit(TransactionId transactionId) {
-        throw new UnsupportedOperationException("This operation is not yet " +
-                                                        "supported.");    }
+        throw new UnsupportedOperationException("This operation is not yet supported.");
+    }
 
     @Override
     public CompletableFuture<Void> rollback(TransactionId transactionId) {
-        throw new UnsupportedOperationException("This operation is not yet " +
-                                                        "supported.");    }
+        throw new UnsupportedOperationException("This operation is not yet supported.");
+    }
 
-    @Override
-    public CompletableFuture<Boolean> prepareAndCommit(
-            MapTransaction<String, V1> transaction) {
-        throw new UnsupportedOperationException("This operation is not yet " +
-                                                        "supported.");    }
     private class InternalBackingMapEventListener
             implements MapEventListener<String, V2> {
 
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMap.java
index afa67f5..8a13c01 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMap.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMap.java
@@ -36,6 +36,7 @@
 
 import org.onlab.util.Match;
 import org.onlab.util.Tools;
+import org.onosproject.store.primitives.MapUpdate;
 import org.onosproject.store.primitives.TransactionId;
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.Clear;
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.ContainsKey;
@@ -47,6 +48,7 @@
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.KeySet;
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.Listen;
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.Size;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.TransactionBegin;
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.TransactionCommit;
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.TransactionPrepare;
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.TransactionPrepareAndCommit;
@@ -58,7 +60,8 @@
 import org.onosproject.store.service.ConsistentMapException;
 import org.onosproject.store.service.MapEvent;
 import org.onosproject.store.service.MapEventListener;
-import org.onosproject.store.service.MapTransaction;
+import org.onosproject.store.service.TransactionLog;
+import org.onosproject.store.service.Version;
 import org.onosproject.store.service.Versioned;
 
 import com.google.common.collect.ImmutableSet;
@@ -294,8 +297,22 @@
     }
 
     @Override
-    public CompletableFuture<Boolean> prepare(MapTransaction<String, byte[]> transaction) {
-        return client.submit(new TransactionPrepare(transaction)).thenApply(v -> v == PrepareResult.OK);
+    public CompletableFuture<Version> begin(TransactionId transactionId) {
+        return client.submit(new TransactionBegin()).thenApply(Version::new);
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepare(
+            TransactionLog<MapUpdate<String, byte[]>> transactionLog) {
+        return client.submit(new TransactionPrepare(transactionLog))
+                .thenApply(v -> v == PrepareResult.OK);
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepareAndCommit(
+            TransactionLog<MapUpdate<String, byte[]>> transactionLog) {
+        return client.submit(new TransactionPrepareAndCommit(transactionLog))
+                .thenApply(v -> v == PrepareResult.OK);
     }
 
     @Override
@@ -305,13 +322,7 @@
 
     @Override
     public CompletableFuture<Void> rollback(TransactionId transactionId) {
-        return client.submit(new TransactionRollback(transactionId))
-                .thenApply(v -> null);
-    }
-
-    @Override
-    public CompletableFuture<Boolean> prepareAndCommit(MapTransaction<String, byte[]> transaction) {
-        return client.submit(new TransactionPrepareAndCommit(transaction)).thenApply(v -> v == PrepareResult.OK);
+        return client.submit(new TransactionRollback(transactionId)).thenApply(v -> null);
     }
 
     @Override
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMapCommands.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMapCommands.java
index 20b13f1..0c5dd76 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMapCommands.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMapCommands.java
@@ -30,8 +30,9 @@
 import java.util.Set;
 
 import org.onlab.util.Match;
+import org.onosproject.store.primitives.MapUpdate;
 import org.onosproject.store.primitives.TransactionId;
-import org.onosproject.store.service.MapTransaction;
+import org.onosproject.store.service.TransactionLog;
 import org.onosproject.store.service.Versioned;
 
 import com.google.common.base.MoreObjects;
@@ -202,39 +203,49 @@
     }
 
     /**
+     * Transaction begin query.
+     */
+    public static class TransactionBegin extends MapQuery<Long> {
+        @Override
+        public ConsistencyLevel consistency() {
+            return ConsistencyLevel.LINEARIZABLE;
+        }
+    }
+
+    /**
      * Map prepare command.
      */
     @SuppressWarnings("serial")
     public static class TransactionPrepare extends MapCommand<PrepareResult> {
-        private MapTransaction<String, byte[]> mapTransaction;
+        private TransactionLog<MapUpdate<String, byte[]>> transactionLog;
 
         public TransactionPrepare() {
         }
 
-        public TransactionPrepare(MapTransaction<String, byte[]> mapTransaction) {
-            this.mapTransaction = mapTransaction;
+        public TransactionPrepare(TransactionLog<MapUpdate<String, byte[]>> transactionLog) {
+            this.transactionLog = transactionLog;
         }
 
-        public MapTransaction<String, byte[]> transaction() {
-            return mapTransaction;
+        public TransactionLog<MapUpdate<String, byte[]>> transactionLog() {
+            return transactionLog;
         }
 
         @Override
         public void writeObject(BufferOutput<?> buffer, Serializer serializer) {
             super.writeObject(buffer, serializer);
-            serializer.writeObject(mapTransaction, buffer);
+            serializer.writeObject(transactionLog, buffer);
         }
 
         @Override
         public void readObject(BufferInput<?> buffer, Serializer serializer) {
             super.readObject(buffer, serializer);
-            mapTransaction = serializer.readObject(buffer);
+            transactionLog = serializer.readObject(buffer);
         }
 
         @Override
         public String toString() {
             return MoreObjects.toStringHelper(getClass())
-                    .add("mapTransaction", mapTransaction)
+                    .add("transactionLog", transactionLog)
                     .toString();
         }
     }
@@ -247,8 +258,8 @@
         public TransactionPrepareAndCommit() {
         }
 
-        public TransactionPrepareAndCommit(MapTransaction<String, byte[]> mapTransaction) {
-            super(mapTransaction);
+        public TransactionPrepareAndCommit(TransactionLog<MapUpdate<String, byte[]>> transactionLog) {
+            super(transactionLog);
         }
 
         @Override
@@ -592,6 +603,7 @@
             registry.register(Size.class, -769);
             registry.register(Listen.class, -770);
             registry.register(Unlisten.class, -771);
+            registry.register(TransactionBegin.class, -777);
             registry.register(TransactionPrepare.class, -772);
             registry.register(TransactionCommit.class, -773);
             registry.register(TransactionRollback.class, -774);
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMapState.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMapState.java
index 3a99b02..e844b8f 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMapState.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMapState.java
@@ -52,6 +52,7 @@
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.KeySet;
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.Listen;
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.Size;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.TransactionBegin;
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.TransactionCommit;
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.TransactionPrepare;
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.TransactionPrepareAndCommit;
@@ -60,7 +61,7 @@
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.UpdateAndGet;
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.Values;
 import org.onosproject.store.service.MapEvent;
-import org.onosproject.store.service.MapTransaction;
+import org.onosproject.store.service.TransactionLog;
 import org.onosproject.store.service.Versioned;
 import org.slf4j.Logger;
 
@@ -110,6 +111,7 @@
         // Commands
         executor.register(UpdateAndGet.class, this::updateAndGet);
         executor.register(AtomixConsistentMapCommands.Clear.class, this::clear);
+        executor.register(TransactionBegin.class, this::begin);
         executor.register(TransactionPrepare.class, this::prepare);
         executor.register(TransactionCommit.class, this::commit);
         executor.register(TransactionRollback.class, this::rollback);
@@ -374,6 +376,20 @@
     }
 
     /**
+     * Handles a begin commit.
+     *
+     * @param commit transaction begin commit
+     * @return transaction state version
+     */
+    protected long begin(Commit<? extends TransactionBegin> commit) {
+        try {
+            return commit.index();
+        } finally {
+            commit.close();
+        }
+    }
+
+    /**
      * Handles an prepare and commit commit.
      *
      * @param commit transaction prepare and commit commit
@@ -382,7 +398,7 @@
     protected PrepareResult prepareAndCommit(Commit<? extends TransactionPrepareAndCommit> commit) {
         PrepareResult prepareResult = prepare(commit);
         if (prepareResult == PrepareResult.OK) {
-            commitInternal(commit.operation().transaction().transactionId(), commit.index());
+            commitInternal(commit.operation().transactionLog().transactionId(), commit.index());
         }
         return prepareResult;
     }
@@ -396,8 +412,8 @@
     protected PrepareResult prepare(Commit<? extends TransactionPrepare> commit) {
         boolean ok = false;
         try {
-            MapTransaction<String, byte[]> transaction = commit.operation().transaction();
-            for (MapUpdate<String, byte[]> update : transaction.updates()) {
+            TransactionLog<MapUpdate<String, byte[]>> transaction = commit.operation().transactionLog();
+            for (MapUpdate<String, byte[]> update : transaction.records()) {
                 String key = update.key();
                 if (preparedKeys.contains(key)) {
                     return PrepareResult.CONCURRENT_TRANSACTION;
@@ -416,7 +432,7 @@
             // No violations detected. Add to pendingTransactions and mark
             // modified keys as locked for updates.
             pendingTransactions.put(transaction.transactionId(), commit);
-            transaction.updates().forEach(u -> preparedKeys.add(u.key()));
+            transaction.records().forEach(u -> preparedKeys.add(u.key()));
             ok = true;
             return PrepareResult.OK;
         } catch (Exception e) {
@@ -453,16 +469,16 @@
         if (prepareCommit == null) {
             return CommitResult.UNKNOWN_TRANSACTION_ID;
         }
-        MapTransaction<String, byte[]> transaction = prepareCommit.operation().transaction();
+        TransactionLog<MapUpdate<String, byte[]>> transaction = prepareCommit.operation().transactionLog();
         long totalReferencesToCommit = transaction
-                .updates()
+                .records()
                 .stream()
                 .filter(update -> update.type() != MapUpdate.Type.REMOVE_IF_VERSION_MATCH)
                 .count();
         CountDownCompleter<Commit<? extends TransactionPrepare>> completer =
                 new CountDownCompleter<>(prepareCommit, totalReferencesToCommit, Commit::close);
         List<MapEvent<String, byte[]>> eventsToPublish = Lists.newArrayList();
-        for (MapUpdate<String, byte[]> update : transaction.updates()) {
+        for (MapUpdate<String, byte[]> update : transaction.records()) {
             String key = update.key();
             checkState(preparedKeys.remove(key), "key is not prepared");
             MapEntryValue previousValue = mapEntries.remove(key);
@@ -496,8 +512,8 @@
                 return RollbackResult.UNKNOWN_TRANSACTION_ID;
             } else {
                 prepareCommit.operation()
-                             .transaction()
-                             .updates()
+                             .transactionLog()
+                             .records()
                              .forEach(u -> preparedKeys.remove(u.key()));
                 prepareCommit.close();
                 return RollbackResult.OK;
@@ -648,7 +664,7 @@
 
         @Override
         public byte[] value() {
-            MapTransaction<String, byte[]> transaction = completer.object().operation().transaction();
+            TransactionLog<MapUpdate<String, byte[]>> transaction = completer.object().operation().transactionLog();
             return valueForKey(key, transaction);
         }
 
@@ -662,8 +678,8 @@
             completer.countDown();
         }
 
-        private byte[] valueForKey(String key, MapTransaction<String, byte[]> transaction) {
-            MapUpdate<String, byte[]>  update = transaction.updates()
+        private byte[] valueForKey(String key, TransactionLog<MapUpdate<String, byte[]>> transaction) {
+            MapUpdate<String, byte[]>  update = transaction.records()
                                                            .stream()
                                                            .filter(u -> u.key().equals(key))
                                                            .findFirst()
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentTreeMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentTreeMap.java
index 3dd34fd..bf462c9 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentTreeMap.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentTreeMap.java
@@ -21,6 +21,7 @@
 import io.atomix.resource.AbstractResource;
 import io.atomix.resource.ResourceTypeInfo;
 import org.onlab.util.Match;
+import org.onosproject.store.primitives.MapUpdate;
 import org.onosproject.store.primitives.TransactionId;
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentTreeMapCommands.FirstKey;
 import org.onosproject.store.primitives.resources.impl.AtomixConsistentTreeMapCommands.FloorEntry;
@@ -30,7 +31,8 @@
 import org.onosproject.store.service.AsyncConsistentTreeMap;
 import org.onosproject.store.service.MapEvent;
 import org.onosproject.store.service.MapEventListener;
-import org.onosproject.store.service.MapTransaction;
+import org.onosproject.store.service.TransactionLog;
+import org.onosproject.store.service.Version;
 import org.onosproject.store.service.Versioned;
 
 import java.util.Collection;
@@ -368,40 +370,38 @@
 
     @Override
     public CompletableFuture<NavigableSet<String>> navigableKeySet() {
-        throw new UnsupportedOperationException("This operation is not yet " +
-                                                        "supported.");
+        throw new UnsupportedOperationException("This operation is not yet supported.");
     }
 
     @Override
     public CompletableFuture<NavigableMap<String, byte[]>> subMap(
             String upperKey, String lowerKey, boolean inclusiveUpper,
             boolean inclusiveLower) {
-        throw new UnsupportedOperationException("This operation is not yet " +
-                                                        "supported.");    }
-
-    @Override
-    public CompletableFuture<Boolean> prepareAndCommit(MapTransaction<String,
-            byte[]> transaction) {
-        throw new UnsupportedOperationException("This operation is not yet " +
-                                                        "supported.");
+        throw new UnsupportedOperationException("This operation is not yet supported.");
     }
 
     @Override
-    public CompletableFuture<Boolean> prepare(MapTransaction<String, byte[]>
-                                                          transaction) {
-        throw new UnsupportedOperationException("This operation is not yet " +
-                                                        "supported.");
+    public CompletableFuture<Version> begin(TransactionId transactionId) {
+        throw new UnsupportedOperationException("This operation is not yet supported.");
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepare(TransactionLog<MapUpdate<String, byte[]>> transactionLog) {
+        throw new UnsupportedOperationException("This operation is not yet supported.");
+    }
+
+    @Override
+    public CompletableFuture<Boolean> prepareAndCommit(TransactionLog<MapUpdate<String, byte[]>> transactionLog) {
+        throw new UnsupportedOperationException("This operation is not yet supported.");
     }
 
     @Override
     public CompletableFuture<Void> commit(TransactionId transactionId) {
-        throw new UnsupportedOperationException("This operation is not yet " +
-                                                        "supported.");
+        throw new UnsupportedOperationException("This operation is not yet supported.");
     }
 
     @Override
     public CompletableFuture<Void> rollback(TransactionId transactionId) {
-        throw new UnsupportedOperationException("This operation is not yet " +
-                                                        "supported.");
+        throw new UnsupportedOperationException("This operation is not yet supported.");
     }
 }
diff --git a/core/store/primitives/src/test/java/org/onosproject/store/primitives/impl/TransactionManagerTest.java b/core/store/primitives/src/test/java/org/onosproject/store/primitives/impl/TransactionManagerTest.java
new file mode 100644
index 0000000..9245a03
--- /dev/null
+++ b/core/store/primitives/src/test/java/org/onosproject/store/primitives/impl/TransactionManagerTest.java
@@ -0,0 +1,102 @@
+/*
+ * Copyright 2017-present 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.primitives.impl;
+
+import java.util.HashMap;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.function.Consumer;
+
+import com.google.common.collect.Sets;
+import org.junit.Test;
+import org.onosproject.cluster.PartitionId;
+import org.onosproject.store.primitives.DistributedPrimitiveCreator;
+import org.onosproject.store.primitives.PartitionService;
+import org.onosproject.store.primitives.TransactionId;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.AsyncConsistentMap;
+import org.onosproject.store.service.ConsistentMapBuilder;
+import org.onosproject.store.service.MapEventListener;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.StorageService;
+
+import static org.easymock.EasyMock.anyObject;
+import static org.easymock.EasyMock.anyString;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.expectLastCall;
+import static org.easymock.EasyMock.mock;
+import static org.easymock.EasyMock.replay;
+import static org.junit.Assert.assertSame;
+
+/**
+ * Transaction manager test.
+ */
+public class TransactionManagerTest {
+
+    @Test
+    public void testTransactionMapCaching() throws Exception {
+        AsyncConsistentMap asyncMap = mock(AsyncConsistentMap.class);
+        expect(asyncMap.name()).andReturn("foo");
+        expect(asyncMap.addListener(anyObject(MapEventListener.class), anyObject(Executor.class)))
+                .andReturn(CompletableFuture.completedFuture(null)).anyTimes();
+        asyncMap.addStatusChangeListener(anyObject(Consumer.class));
+        expectLastCall().anyTimes();
+        expect(asyncMap.entrySet()).andReturn(CompletableFuture.completedFuture(new HashMap<>().entrySet())).anyTimes();
+
+        ConsistentMapBuilder mapBuilder = mock(ConsistentMapBuilder.class);
+        expect(mapBuilder.withName(anyString())).andReturn(mapBuilder).anyTimes();
+        expect(mapBuilder.withSerializer(anyObject(Serializer.class))).andReturn(mapBuilder).anyTimes();
+        expect(mapBuilder.buildAsyncMap()).andReturn(asyncMap).anyTimes();
+
+        DistributedPrimitiveCreator primitiveCreator = mock(DistributedPrimitiveCreator.class);
+        expect(primitiveCreator.newAsyncConsistentMap(anyString(), anyObject(Serializer.class)))
+                .andReturn(asyncMap).anyTimes();
+
+        StorageService storageService = mock(StorageService.class);
+        expect(storageService.consistentMapBuilder()).andReturn(mapBuilder);
+
+        PartitionService partitionService = mock(PartitionService.class);
+        Set<PartitionId> partitionIds = Sets.newHashSet(PartitionId.from(1), PartitionId.from(2), PartitionId.from(3));
+        expect(partitionService.getAllPartitionIds())
+                .andReturn(partitionIds).anyTimes();
+        expect(partitionService.getNumberOfPartitions())
+                .andReturn(partitionIds.size()).anyTimes();
+        expect(partitionService.getDistributedPrimitiveCreator(anyObject(PartitionId.class)))
+                .andReturn(primitiveCreator).anyTimes();
+
+        replay(storageService, partitionService, asyncMap, primitiveCreator, mapBuilder);
+
+        TransactionManager transactionManager = new TransactionManager(storageService, partitionService);
+        TransactionId transactionId = TransactionId.from(UUID.randomUUID().toString());
+        TransactionCoordinator transactionCoordinator = new TransactionCoordinator(transactionId, transactionManager);
+        Serializer serializer = Serializer.using(KryoNamespaces.API);
+
+        PartitionedTransactionalMap<String, String> transactionalMap1 = (PartitionedTransactionalMap)
+                transactionManager.getTransactionalMap("foo", serializer, transactionCoordinator);
+        PartitionedTransactionalMap<String, String> transactionalMap2 = (PartitionedTransactionalMap)
+                transactionManager.getTransactionalMap("foo", serializer, transactionCoordinator);
+
+        assertSame(transactionalMap1.partitions.get(PartitionId.from(1)).transaction.transactionalObject,
+                transactionalMap2.partitions.get(PartitionId.from(1)).transaction.transactionalObject);
+        assertSame(transactionalMap1.partitions.get(PartitionId.from(2)).transaction.transactionalObject,
+                transactionalMap2.partitions.get(PartitionId.from(2)).transaction.transactionalObject);
+        assertSame(transactionalMap1.partitions.get(PartitionId.from(3)).transaction.transactionalObject,
+                transactionalMap2.partitions.get(PartitionId.from(3)).transaction.transactionalObject);
+    }
+
+}
diff --git a/core/store/primitives/src/test/java/org/onosproject/store/primitives/impl/TransactionTest.java b/core/store/primitives/src/test/java/org/onosproject/store/primitives/impl/TransactionTest.java
new file mode 100644
index 0000000..533ea97
--- /dev/null
+++ b/core/store/primitives/src/test/java/org/onosproject/store/primitives/impl/TransactionTest.java
@@ -0,0 +1,397 @@
+/*
+ * Copyright 2017-present 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.primitives.impl;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.function.Predicate;
+
+import com.google.common.hash.Hashing;
+import org.junit.Test;
+import org.onosproject.cluster.PartitionId;
+import org.onosproject.store.primitives.MapUpdate;
+import org.onosproject.store.primitives.TransactionId;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.AsyncConsistentMap;
+import org.onosproject.store.service.CommitStatus;
+import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.MapEventListener;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.TransactionLog;
+import org.onosproject.store.service.Version;
+import org.onosproject.store.service.Versioned;
+
+import static junit.framework.TestCase.assertNull;
+import static org.easymock.EasyMock.anyObject;
+import static org.easymock.EasyMock.anyString;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.mock;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.strictMock;
+import static org.easymock.EasyMock.verify;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Transaction test.
+ */
+public class TransactionTest {
+
+    @Test
+    public void testTransaction() throws Exception {
+        AsyncConsistentMap<String, String> asyncMap = strictMock(AsyncConsistentMap.class);
+        TransactionId transactionId = TransactionId.from("foo");
+        List<MapUpdate<String, String>> updates = Collections.singletonList(new MapUpdate<>());
+        Transaction<MapUpdate<String, String>> transaction = new Transaction<>(transactionId, asyncMap);
+        assertEquals(transactionId, transaction.transactionId());
+
+        expect(asyncMap.begin(transactionId))
+                .andReturn(CompletableFuture.completedFuture(new Version(1)));
+        expect(asyncMap.prepare(new TransactionLog<>(transactionId, updates)))
+                .andReturn(CompletableFuture.completedFuture(true));
+        expect(asyncMap.commit(transactionId))
+                .andReturn(CompletableFuture.completedFuture(null));
+        replay(asyncMap);
+
+        assertEquals(Transaction.State.ACTIVE, transaction.state());
+        assertEquals(1, transaction.begin().join().value());
+        assertEquals(Transaction.State.ACTIVE, transaction.state());
+        assertTrue(transaction.prepare(updates).join());
+        assertEquals(Transaction.State.PREPARED, transaction.state());
+        transaction.commit();
+        assertEquals(Transaction.State.COMMITTED, transaction.state());
+        verify(asyncMap);
+    }
+
+    @Test
+    public void testTransactionFailOnOutOfOrderCalls() throws Exception {
+        AsyncConsistentMap<String, String> asyncMap = strictMock(AsyncConsistentMap.class);
+        TransactionId transactionId = TransactionId.from("foo");
+        List<MapUpdate<String, String>> updates = Collections.singletonList(new MapUpdate<>());
+        Transaction<MapUpdate<String, String>> transaction = new Transaction<>(transactionId, asyncMap);
+
+        try {
+            transaction.prepare(updates);
+            fail();
+        } catch (IllegalStateException e) {
+        }
+
+        try {
+            transaction.commit();
+            fail();
+        } catch (IllegalStateException e) {
+        }
+
+        try {
+            transaction.rollback();
+            fail();
+        } catch (IllegalStateException e) {
+        }
+
+        expect(asyncMap.begin(transactionId))
+                .andReturn(CompletableFuture.completedFuture(new Version(1)));
+        expect(asyncMap.prepare(new TransactionLog<>(transactionId, updates)))
+                .andReturn(CompletableFuture.completedFuture(true));
+        replay(asyncMap);
+
+        assertFalse(transaction.isOpen());
+        assertEquals(Transaction.State.ACTIVE, transaction.state());
+        assertEquals(1, transaction.begin().join().value());
+        assertTrue(transaction.isOpen());
+        assertEquals(Transaction.State.ACTIVE, transaction.state());
+        assertTrue(transaction.prepare(updates).join());
+        assertEquals(Transaction.State.PREPARED, transaction.state());
+
+        try {
+            transaction.begin();
+            fail();
+        } catch (IllegalStateException e) {
+        }
+        verify(asyncMap);
+    }
+
+    @Test
+    public void testCoordinatedMapTransaction() throws Exception {
+        List<Object> mocks = new ArrayList<>();
+
+        Map<PartitionId, DefaultTransactionalMapParticipant<String, String>> participants = new HashMap<>();
+        List<PartitionId> sortedParticipants = new ArrayList<>();
+        TransactionId transactionId = TransactionId.from(UUID.randomUUID().toString());
+        for (int i = 1; i <= 3; i++) {
+            AsyncConsistentMap<String, String> asyncMap = mock(AsyncConsistentMap.class);
+            mocks.add(asyncMap);
+
+            ConsistentMap<String, String> consistentMap = new TestConsistentMap<>();
+            Transaction<MapUpdate<String, String>> transaction = new Transaction<>(transactionId, asyncMap);
+            PartitionId partitionId = PartitionId.from(i);
+            participants.put(partitionId, new DefaultTransactionalMapParticipant<>(consistentMap, transaction));
+            sortedParticipants.add(partitionId);
+        }
+
+        expect(participants.get(PartitionId.from(1)).transaction.transactionalObject
+                .begin(anyObject(TransactionId.class)))
+                .andReturn(CompletableFuture.completedFuture(new Version(1)));
+
+        expect(participants.get(PartitionId.from(1)).transaction.transactionalObject.prepare(
+                new TransactionLog<>(transactionId, Arrays.asList(
+                        MapUpdate.<String, String>newBuilder()
+                                .withType(MapUpdate.Type.REMOVE_IF_VERSION_MATCH)
+                                .withKey("foo")
+                                .withCurrentVersion(1)
+                                .build(),
+                        MapUpdate.<String, String>newBuilder()
+                                .withType(MapUpdate.Type.REMOVE_IF_VERSION_MATCH)
+                                .withKey("baz")
+                                .withCurrentVersion(2)
+                                .build()
+                )))).andReturn(CompletableFuture.completedFuture(true));
+
+        expect(participants.get(PartitionId.from(1)).transaction.transactionalObject.commit(transactionId))
+                .andReturn(CompletableFuture.completedFuture(null));
+
+        expect(participants.get(PartitionId.from(3)).transaction.transactionalObject
+                .begin(anyObject(TransactionId.class)))
+                .andReturn(CompletableFuture.completedFuture(new Version(1)));
+
+        expect(participants.get(PartitionId.from(3)).transaction.transactionalObject.prepare(
+                new TransactionLog<>(transactionId, Arrays.asList(
+                        MapUpdate.<String, String>newBuilder()
+                                .withType(MapUpdate.Type.PUT_IF_ABSENT)
+                                .withKey("bar")
+                                .withValue("baz")
+                                .build()
+                )))).andReturn(CompletableFuture.completedFuture(true));
+
+        expect(participants.get(PartitionId.from(3)).transaction.transactionalObject.commit(transactionId))
+                .andReturn(CompletableFuture.completedFuture(null));
+
+        TransactionManager transactionManager = mock(TransactionManager.class);
+        expect(transactionManager.updateState(anyObject(TransactionId.class), anyObject(Transaction.State.class)))
+                .andReturn(CompletableFuture.completedFuture(null))
+                .anyTimes();
+        expect(transactionManager.remove(anyObject(TransactionId.class)))
+                .andReturn(CompletableFuture.completedFuture(null))
+                .anyTimes();
+        mocks.add(transactionManager);
+
+        TransactionCoordinator transactionCoordinator = new TransactionCoordinator(transactionId, transactionManager);
+
+        Hasher<String> hasher = key -> {
+            int hashCode = Hashing.sha256().hashBytes(key.getBytes()).asInt();
+            return sortedParticipants.get(Math.abs(hashCode) % sortedParticipants.size());
+        };
+
+        expect(transactionManager.<String, String>getTransactionalMap(anyString(), anyObject(), anyObject()))
+                .andReturn(new PartitionedTransactionalMap(participants, hasher));
+
+        replay(mocks.toArray());
+
+        PartitionedTransactionalMap<String, String> transactionalMap = (PartitionedTransactionalMap)
+                transactionCoordinator.getTransactionalMap("foo", Serializer.using(KryoNamespaces.API));
+
+        // Sneak a couple entries in the first partition.
+        transactionalMap.partitions.get(PartitionId.from(1)).backingMap.put("foo", "bar");
+        transactionalMap.partitions.get(PartitionId.from(1)).backingMap.put("baz", "foo");
+
+        assertTrue(transactionalMap.containsKey("foo"));
+        assertEquals("bar", transactionalMap.remove("foo"));
+        assertFalse(transactionalMap.containsKey("bar"));
+        assertNull(transactionalMap.put("bar", "baz"));
+        assertTrue(transactionalMap.containsKey("bar"));
+        assertTrue(transactionalMap.containsKey("baz"));
+        assertFalse(transactionalMap.remove("baz", "baz"));
+        assertTrue(transactionalMap.remove("baz", "foo"));
+        assertFalse(transactionalMap.containsKey("baz"));
+
+        assertEquals(CommitStatus.SUCCESS, transactionCoordinator.commit().join());
+        verify(mocks.toArray());
+    }
+
+    private static class TestConsistentMap<K, V> implements ConsistentMap<K, V> {
+        private final Map<K, Versioned<V>> map = new HashMap<>();
+        private final AtomicLong version = new AtomicLong();
+
+        @Override
+        public String name() {
+            return null;
+        }
+
+        @Override
+        public Type primitiveType() {
+            return Type.CONSISTENT_MAP;
+        }
+
+        private long nextVersion() {
+            return version.incrementAndGet();
+        }
+
+        @Override
+        public int size() {
+            return map.size();
+        }
+
+        @Override
+        public boolean isEmpty() {
+            return map.isEmpty();
+        }
+
+        @Override
+        public boolean containsKey(K key) {
+            return map.containsKey(key);
+        }
+
+        @Override
+        public boolean containsValue(V value) {
+            return map.containsValue(value);
+        }
+
+        @Override
+        public Versioned<V> get(K key) {
+            return map.get(key);
+        }
+
+        @Override
+        public Versioned<V> getOrDefault(K key, V defaultValue) {
+            return map.getOrDefault(key, new Versioned<>(defaultValue, 0));
+        }
+
+        @Override
+        public Versioned<V> computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Versioned<V> compute(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Versioned<V> computeIfPresent(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Versioned<V> computeIf(K key,
+                Predicate<? super V> condition, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Versioned<V> put(K key, V value) {
+            return map.put(key, new Versioned<>(value, nextVersion()));
+        }
+
+        @Override
+        public Versioned<V> putAndGet(K key, V value) {
+            return put(key, value);
+        }
+
+        @Override
+        public Versioned<V> remove(K key) {
+            return map.remove(key);
+        }
+
+        @Override
+        public void clear() {
+            map.clear();
+        }
+
+        @Override
+        public Set<K> keySet() {
+            return map.keySet();
+        }
+
+        @Override
+        public Collection<Versioned<V>> values() {
+            return map.values();
+        }
+
+        @Override
+        public Set<Map.Entry<K, Versioned<V>>> entrySet() {
+            return map.entrySet();
+        }
+
+        @Override
+        public Versioned<V> putIfAbsent(K key, V value) {
+            return map.putIfAbsent(key, new Versioned<>(value, nextVersion()));
+        }
+
+        @Override
+        public boolean remove(K key, V value) {
+            return map.remove(key, value);
+        }
+
+        @Override
+        public boolean remove(K key, long version) {
+            Versioned<V> value = map.get(key);
+            if (value != null && value.version() == version) {
+                map.remove(key);
+                return true;
+            }
+            return false;
+        }
+
+        @Override
+        public Versioned<V> replace(K key, V value) {
+            return map.replace(key, new Versioned<>(value, nextVersion()));
+        }
+
+        @Override
+        public boolean replace(K key, V oldValue, V newValue) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean replace(K key, long oldVersion, V newValue) {
+            Versioned<V> value = map.get(key);
+            if (value != null && value.version() == oldVersion) {
+                map.put(key, new Versioned<>(newValue, nextVersion()));
+                return true;
+            }
+            return false;
+        }
+
+        @Override
+        public void addListener(MapEventListener<K, V> listener, Executor executor) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public void removeListener(MapEventListener<K, V> listener) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Map<K, V> asJavaMap() {
+            throw new UnsupportedOperationException();
+        }
+    }
+
+}
diff --git a/core/store/primitives/src/test/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMapTest.java b/core/store/primitives/src/test/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMapTest.java
index 23d528b..ac4c60d 100644
--- a/core/store/primitives/src/test/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMapTest.java
+++ b/core/store/primitives/src/test/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMapTest.java
@@ -26,7 +26,7 @@
 import org.onosproject.store.primitives.TransactionId;
 import org.onosproject.store.service.MapEvent;
 import org.onosproject.store.service.MapEventListener;
-import org.onosproject.store.service.MapTransaction;
+import org.onosproject.store.service.TransactionLog;
 import org.onosproject.store.service.Versioned;
 
 import java.util.Arrays;
@@ -366,7 +366,8 @@
                 .withValue(value1)
                 .build();
 
-        MapTransaction<String, byte[]> tx = new MapTransaction<>(TransactionId.from("tx1"), Arrays.asList(update1));
+        TransactionLog<MapUpdate<String, byte[]>> tx =
+                new TransactionLog<>(TransactionId.from("tx1"), Arrays.asList(update1));
 
         map.prepare(tx).thenAccept(result -> {
             assertEquals(true, result);
@@ -416,7 +417,7 @@
                 .withCurrentVersion(currFooVersion)
                 .build();
 
-        tx = new MapTransaction<>(TransactionId.from("tx2"), Arrays.asList(remove1));
+        tx = new TransactionLog<>(TransactionId.from("tx2"), Arrays.asList(remove1));
 
         map.prepare(tx).thenAccept(result -> {
             assertTrue("prepare should succeed", result);
@@ -459,7 +460,8 @@
                 .withKey("foo")
                 .withValue(value1)
                 .build();
-        MapTransaction<String, byte[]> tx = new MapTransaction<>(TransactionId.from("tx1"), Arrays.asList(update1));
+        TransactionLog<MapUpdate<String, byte[]>> tx =
+                new TransactionLog<>(TransactionId.from("tx1"), Arrays.asList(update1));
         map.prepare(tx).thenAccept(result -> {
             assertEquals(true, result);
         }).join();
diff --git a/core/store/serializers/src/main/java/org/onosproject/store/serializers/KryoNamespaces.java b/core/store/serializers/src/main/java/org/onosproject/store/serializers/KryoNamespaces.java
index dd3ef19..d638f5f 100644
--- a/core/store/serializers/src/main/java/org/onosproject/store/serializers/KryoNamespaces.java
+++ b/core/store/serializers/src/main/java/org/onosproject/store/serializers/KryoNamespaces.java
@@ -220,7 +220,7 @@
 import org.onosproject.store.primitives.MapUpdate;
 import org.onosproject.store.primitives.TransactionId;
 import org.onosproject.store.service.MapEvent;
-import org.onosproject.store.service.MapTransaction;
+import org.onosproject.store.service.TransactionLog;
 import org.onosproject.store.service.SetEvent;
 import org.onosproject.store.service.Task;
 import org.onosproject.store.service.Versioned;
@@ -528,7 +528,7 @@
             .register(ExtensionSelectorType.class)
             .register(ExtensionTreatmentType.class)
             .register(TransactionId.class)
-            .register(MapTransaction.class)
+            .register(TransactionLog.class)
             .register(MapUpdate.class)
             .register(MapUpdate.Type.class)
             .register(Versioned.class)