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 e04f10e..f720ded 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
@@ -31,7 +31,6 @@
  *
  * @param <K> map key type
  * @param <V> map value type
- *
  */
 public final class MapUpdate<K, V> {
 
@@ -39,47 +38,38 @@
      * Type of database update operation.
      */
     public enum Type {
-        /**
-         * Insert/Update entry without any checks.
-         */
-        PUT,
 
         /**
-         * Insert an entry iff there is no existing entry for that key.
+         * Acquires a read lock on a key.
+         * <p>
+         * This record type will check to ensure that the lock version matches the current version for the key
+         * in the map and acquire a read lock on the key for the duration of the transaction.
          */
-        PUT_IF_ABSENT,
+        LOCK,
 
         /**
-         * Update entry if the current version matches specified version.
+         * Checks the version of a key without locking the key.
+         * <p>
+         * This record type will perform a simple version check during the prepare phase of the two-phase commit
+         * protocol to ensure that the key has not changed during a transaction.
+         */
+        VERSION_MATCH,
+
+        /**
+         * Updates an entry if the current version matches specified version.
          */
         PUT_IF_VERSION_MATCH,
 
         /**
-         * Update entry if the current value matches specified value.
-         */
-        PUT_IF_VALUE_MATCH,
-
-        /**
-         * Remove entry without any checks.
-         */
-        REMOVE,
-
-        /**
-         * Remove entry if the current version matches specified version.
+         * Removes an entry if the current version matches specified version.
          */
         REMOVE_IF_VERSION_MATCH,
-
-        /**
-         * Remove entry if the current value matches specified value.
-         */
-        REMOVE_IF_VALUE_MATCH,
     }
 
     private Type type;
     private K key;
     private V value;
-    private V currentValue;
-    private long currentVersion = -1;
+    private long version = -1;
 
     /**
      * Returns the type of update operation.
@@ -106,19 +96,11 @@
     }
 
     /**
-     * Returns the expected current value for the key.
-     * @return current value in database.
-     */
-    public V currentValue() {
-        return currentValue;
-    }
-
-    /**
      * Returns the expected current version in the database for the key.
      * @return expected version.
      */
-    public long currentVersion() {
-        return currentVersion;
+    public long version() {
+        return version;
     }
 
     /**
@@ -135,14 +117,13 @@
                 .withType(type)
                 .withKey(keyMapper.apply(key))
                 .withValue(value == null ? null : valueMapper.apply(value))
-                .withCurrentValue(currentValue == null ? null : valueMapper.apply(currentValue))
-                .withCurrentVersion(currentVersion)
+                .withVersion(version)
                 .build();
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(type, key, value, currentValue, currentVersion);
+        return Objects.hash(type, key, value, version);
     }
 
     @Override
@@ -152,8 +133,7 @@
             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);
+                    && Objects.equals(this.version, that.version);
         }
         return false;
     }
@@ -164,8 +144,7 @@
             .add("type", type)
             .add("key", key)
             .add("value", value instanceof byte[] ? new ByteArraySizeHashPrinter((byte[]) value) : value)
-            .add("currentValue", currentValue)
-            .add("currentVersion", currentVersion)
+            .add("version", version)
             .toString();
     }
 
@@ -205,48 +184,38 @@
             return this;
         }
 
-        public Builder<K, V> withCurrentValue(V value) {
-            update.currentValue = value;
-            return this;
-        }
-
         public Builder<K, V> withValue(V value) {
             update.value = value;
             return this;
         }
 
-        public Builder<K, V> withCurrentVersion(long version) {
-            update.currentVersion = version;
+        public Builder<K, V> withVersion(long version) {
+            update.version = version;
             return this;
         }
 
         private void validateInputs() {
             checkNotNull(update.type, "type must be specified");
-            checkNotNull(update.key, "key must be specified");
             switch (update.type) {
-            case PUT:
-            case PUT_IF_ABSENT:
-                checkNotNull(update.value, "value must be specified.");
-                break;
-            case PUT_IF_VERSION_MATCH:
-                checkNotNull(update.value, "value must be specified.");
-                checkState(update.currentVersion >= 0, "current version must be specified");
-                break;
-            case PUT_IF_VALUE_MATCH:
-                checkNotNull(update.value, "value must be specified.");
-                checkNotNull(update.currentValue, "currentValue must be specified.");
-                break;
-            case REMOVE:
-                break;
-            case REMOVE_IF_VERSION_MATCH:
-                checkState(update.currentVersion >= 0, "current version must be specified");
-                break;
-            case REMOVE_IF_VALUE_MATCH:
-                checkNotNull(update.currentValue, "currentValue must be specified.");
-                break;
-            default:
-                throw new IllegalStateException("Unknown operation type");
+                case VERSION_MATCH:
+                    break;
+                case LOCK:
+                    checkNotNull(update.key, "key must be specified");
+                    checkState(update.version >= 0, "version must be specified");
+                    break;
+                case PUT_IF_VERSION_MATCH:
+                    checkNotNull(update.key, "key must be specified");
+                    checkNotNull(update.value, "value must be specified.");
+                    checkState(update.version >= 0, "version must be specified");
+                    break;
+                case REMOVE_IF_VERSION_MATCH:
+                    checkNotNull(update.key, "key must be specified");
+                    checkState(update.version >= 0, "version must be specified");
+                    break;
+                default:
+                    throw new IllegalStateException("Unknown operation type");
             }
+
         }
     }
 }
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
index c367f5d..ab84c3a 100644
--- a/core/api/src/main/java/org/onosproject/store/service/TransactionLog.java
+++ b/core/api/src/main/java/org/onosproject/store/service/TransactionLog.java
@@ -31,10 +31,12 @@
  */
 public class TransactionLog<T> {
     private final TransactionId transactionId;
+    private final long version;
     private final List<T> records;
 
-    public TransactionLog(TransactionId transactionId, List<T> records) {
+    public TransactionLog(TransactionId transactionId, long version, List<T> records) {
         this.transactionId = transactionId;
+        this.version = version;
         this.records = ImmutableList.copyOf(records);
     }
 
@@ -48,6 +50,15 @@
     }
 
     /**
+     * Returns the transaction lock version.
+     *
+     * @return the transaction lock version
+     */
+    public long version() {
+        return version;
+    }
+
+    /**
      * Returns the list of transaction log records.
      *
      * @return a list of transaction log records
@@ -75,6 +86,7 @@
     public String toString() {
         return MoreObjects.toStringHelper(getClass())
                 .add("transactionId", transactionId)
+                .add("version", version)
                 .add("records", records)
                 .toString();
     }
@@ -88,6 +100,6 @@
      * @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));
+        return new TransactionLog<>(transactionId, version, Lists.transform(records, mapper::apply));
     }
 }
\ 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 53cc160..6f16e97 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
@@ -29,49 +29,27 @@
 public class MapUpdateTest {
 
     private final MapUpdate<String, byte[]> stats1 = MapUpdate.<String, byte[]>newBuilder()
-            .withCurrentValue("1".getBytes())
             .withValue("2".getBytes())
-            .withCurrentVersion(3)
+            .withVersion(3)
             .withKey("4")
-            .withType(MapUpdate.Type.PUT)
+            .withType(MapUpdate.Type.PUT_IF_VERSION_MATCH)
             .build();
 
     private final MapUpdate<String, byte[]> stats2 = MapUpdate.<String, byte[]>newBuilder()
-            .withCurrentValue("1".getBytes())
-            .withValue("2".getBytes())
-            .withCurrentVersion(3)
-            .withKey("4")
-            .withType(MapUpdate.Type.REMOVE)
+            .withType(MapUpdate.Type.VERSION_MATCH)
+            .withVersion(10)
             .build();
 
     private final MapUpdate<String, byte[]> stats3 = MapUpdate.<String, byte[]>newBuilder()
-            .withCurrentValue("1".getBytes())
             .withValue("2".getBytes())
-            .withCurrentVersion(3)
-            .withKey("4")
-            .withType(MapUpdate.Type.REMOVE_IF_VALUE_MATCH)
-            .build();
-
-    private final MapUpdate<String, byte[]> stats4 = MapUpdate.<String, byte[]>newBuilder()
-            .withCurrentValue("1".getBytes())
-            .withValue("2".getBytes())
-            .withCurrentVersion(3)
+            .withVersion(3)
             .withKey("4")
             .withType(MapUpdate.Type.REMOVE_IF_VERSION_MATCH)
             .build();
 
-    private final MapUpdate<String, byte[]> stats5 = MapUpdate.<String, byte[]>newBuilder()
-            .withCurrentValue("1".getBytes())
+    private final MapUpdate<String, byte[]> stats4 = MapUpdate.<String, byte[]>newBuilder()
             .withValue("2".getBytes())
-            .withCurrentVersion(3)
-            .withKey("4")
-            .withType(MapUpdate.Type.PUT_IF_VALUE_MATCH)
-            .build();
-
-    private final MapUpdate<String, byte[]> stats6 = MapUpdate.<String, byte[]>newBuilder()
-            .withCurrentValue("1".getBytes())
-            .withValue("2".getBytes())
-            .withCurrentVersion(3)
+            .withVersion(3)
             .withKey("4")
             .withType(MapUpdate.Type.PUT_IF_VERSION_MATCH)
             .build();
@@ -81,11 +59,10 @@
      */
     @Test
     public void testConstruction() {
-        assertThat(stats1.currentValue(), is("1".getBytes()));
         assertThat(stats1.value(), is("2".getBytes()));
-        assertThat(stats1.currentVersion(), is(3L));
+        assertThat(stats1.version(), is(3L));
         assertThat(stats1.key(), is("4"));
-        assertThat(stats1.type(), is(MapUpdate.Type.PUT));
+        assertThat(stats1.type(), is(MapUpdate.Type.PUT_IF_VERSION_MATCH));
     }
 
     /**
@@ -102,11 +79,6 @@
                 .addEqualityGroup(stats3, stats3)
                 .addEqualityGroup(stats4)
                 .testEquals();
-
-        new EqualsTester()
-                .addEqualityGroup(stats5, stats5)
-                .addEqualityGroup(stats6)
-                .testEquals();
     }
 
     /**
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
index 9acc377..19f6e0c 100644
--- 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
@@ -22,6 +22,7 @@
 import org.apache.commons.lang3.tuple.Pair;
 import org.onosproject.store.primitives.MapUpdate;
 import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.Version;
 import org.onosproject.store.service.Versioned;
 
 /**
@@ -37,13 +38,19 @@
 
     @Override
     protected V read(K key) {
-        Versioned<V> value = readCache.computeIfAbsent(key, backingMap::get);
-        return value != null ? value.value() : null;
+        Versioned<V> value = backingMap.getOrDefault(key, null);
+        readCache.put(key, value);
+        return value.value();
     }
 
     @Override
-    protected Stream<MapUpdate<K, V>> records() {
-        return Stream.concat(deleteStream(), writeStream());
+    public boolean hasPendingUpdates() {
+        return !writeCache.isEmpty() || !deleteSet.isEmpty();
+    }
+
+    @Override
+    protected Stream<MapUpdate<K, V>> records(Version lockVersion) {
+        return Stream.concat(deleteStream(), writeStream(lockVersion));
     }
 
     /**
@@ -52,34 +59,26 @@
     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())
+                        .withVersion(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();
-            }
-        });
+    private Stream<MapUpdate<K, V>> writeStream(Version lockVersion) {
+        return writeCache.entrySet().stream()
+                .map(entry -> {
+                    Versioned<V> original = readCache.get(entry.getKey());
+                    return MapUpdate.<K, V>newBuilder()
+                            .withType(MapUpdate.Type.PUT_IF_VERSION_MATCH)
+                            .withKey(entry.getKey())
+                            .withValue(entry.getValue())
+                            .withVersion(Math.max(original.version(), lockVersion.value()))
+                            .build();
+                });
     }
 }
\ No newline at end of file
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 82b3919..8ad8cb8 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
@@ -109,6 +109,7 @@
     protected final Transactional<T> transactionalObject;
     private final AtomicBoolean open = new AtomicBoolean();
     private volatile State state = State.ACTIVE;
+    private volatile Version lock;
 
     public Transaction(TransactionId transactionId, Transactional<T> transactionalObject) {
         this.transactionId = transactionId;
@@ -191,7 +192,10 @@
      */
     public CompletableFuture<Version> begin() {
         open();
-        return transactionalObject.begin(transactionId);
+        return transactionalObject.begin(transactionId).thenApply(lock -> {
+            this.lock = lock;
+            return lock;
+        });
     }
 
     /**
@@ -208,8 +212,10 @@
     public CompletableFuture<Boolean> prepare(List<T> updates) {
         checkOpen();
         checkActive();
+        Version lock = this.lock;
+        checkState(lock != null, TX_INACTIVE_ERROR);
         setState(State.PREPARING);
-        return transactionalObject.prepare(new TransactionLog<T>(transactionId, updates))
+        return transactionalObject.prepare(new TransactionLog<T>(transactionId, lock.value(), updates))
                 .thenApply(succeeded -> {
                     setState(State.PREPARED);
                     return succeeded;
@@ -228,8 +234,10 @@
     public CompletableFuture<Boolean> prepareAndCommit(List<T> updates) {
         checkOpen();
         checkActive();
+        Version lock = this.lock;
+        checkState(lock != null, TX_INACTIVE_ERROR);
         setState(State.PREPARING);
-        return transactionalObject.prepareAndCommit(new TransactionLog<T>(transactionId, updates))
+        return transactionalObject.prepareAndCommit(new TransactionLog<T>(transactionId, lock.value(), updates))
                 .thenApply(succeeded -> {
                     setState(State.COMMITTED);
                     return succeeded;
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
index 4429a1b..977312a 100644
--- 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
@@ -16,7 +16,6 @@
 
 package org.onosproject.store.primitives.impl;
 
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -60,7 +59,6 @@
     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(
@@ -77,18 +75,22 @@
      * 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());
+        if (lock == null) {
+            synchronized (this) {
+                if (lock == null) {
+                    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());
+                    }
+                }
             }
         }
     }
@@ -183,13 +185,8 @@
     }
 
     @Override
-    public boolean hasPendingUpdates() {
-        return records().findAny().isPresent();
-    }
-
-    @Override
     public CompletableFuture<Boolean> prepare() {
-        return transaction.prepare(log());
+        return transaction.prepare(log(lock));
     }
 
     @Override
@@ -199,7 +196,7 @@
 
     @Override
     public CompletableFuture<Boolean> prepareAndCommit() {
-        return transaction.prepareAndCommit(log());
+        return transaction.prepareAndCommit(log(lock));
     }
 
     @Override
@@ -210,24 +207,25 @@
     /**
      * Returns a list of updates performed within this map partition.
      *
+     * @param lockVersion the global transaction lock version
      * @return a list of map updates
      */
-    protected List<MapUpdate<K, V>> log() {
-        return records().collect(Collectors.toList());
+    protected List<MapUpdate<K, V>> log(Version lockVersion) {
+        return records(lockVersion).collect(Collectors.toList());
     }
 
     /**
      * Returns a stream of updates performed within this map partition.
      *
+     * @param lockVersion the global transaction lock version
      * @return a stream of map updates
      */
-    protected abstract Stream<MapUpdate<K, V>> records();
+    protected abstract Stream<MapUpdate<K, V>> records(Version lockVersion);
 
     @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/resources/impl/AtomixConsistentMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMap.java
index 8a13c01..7220c37 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
@@ -298,7 +298,7 @@
 
     @Override
     public CompletableFuture<Version> begin(TransactionId transactionId) {
-        return client.submit(new TransactionBegin()).thenApply(Version::new);
+        return client.submit(new TransactionBegin(transactionId)).thenApply(Version::new);
     }
 
     @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 0c5dd76..8dd4096 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
@@ -15,6 +15,11 @@
  */
 package org.onosproject.store.primitives.resources.impl;
 
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.MoreObjects;
 import io.atomix.catalyst.buffer.BufferInput;
 import io.atomix.catalyst.buffer.BufferOutput;
 import io.atomix.catalyst.serializer.CatalystSerializable;
@@ -24,19 +29,12 @@
 import io.atomix.catalyst.util.Assert;
 import io.atomix.copycat.Command;
 import io.atomix.copycat.Query;
-
-import java.util.Collection;
-import java.util.Map;
-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.TransactionLog;
 import org.onosproject.store.service.Versioned;
 
-import com.google.common.base.MoreObjects;
-
 /**
  * {@link AtomixConsistentMap} resource state machine operations.
  */
@@ -203,12 +201,32 @@
     }
 
     /**
-     * Transaction begin query.
+     * Transaction begin command.
      */
-    public static class TransactionBegin extends MapQuery<Long> {
+    public static class TransactionBegin extends MapCommand<Long> {
+        private TransactionId transactionId;
+
+        public TransactionBegin() {
+        }
+
+        public TransactionBegin(TransactionId transactionId) {
+            this.transactionId = transactionId;
+        }
+
+        public TransactionId transactionId() {
+            return transactionId;
+        }
+
         @Override
-        public ConsistencyLevel consistency() {
-            return ConsistencyLevel.LINEARIZABLE;
+        public void writeObject(BufferOutput<?> buffer, Serializer serializer) {
+            super.writeObject(buffer, serializer);
+            serializer.writeObject(transactionId, buffer);
+        }
+
+        @Override
+        public void readObject(BufferInput<?> buffer, Serializer serializer) {
+            super.readObject(buffer, serializer);
+            transactionId = serializer.readObject(buffer);
         }
     }
 
@@ -264,7 +282,7 @@
 
         @Override
         public CompactionMode compaction() {
-            return CompactionMode.QUORUM;
+            return CompactionMode.TOMBSTONE;
         }
     }
 
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 e844b8f..a85d51b 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
@@ -79,7 +79,8 @@
     private final Map<Long, Commit<? extends Listen>> listeners = new HashMap<>();
     private final Map<String, MapEntryValue> mapEntries = new HashMap<>();
     private final Set<String> preparedKeys = Sets.newHashSet();
-    private final Map<TransactionId, Commit<? extends TransactionPrepare>> pendingTransactions = Maps.newHashMap();
+    private final Map<TransactionId, TransactionScope> activeTransactions = Maps.newHashMap();
+    private long currentVersion;
 
     public AtomixConsistentMapState(Properties properties) {
         super(properties);
@@ -137,7 +138,8 @@
      */
     protected boolean containsKey(Commit<? extends ContainsKey> commit) {
         try {
-            return toVersioned(mapEntries.get(commit.operation().key())) != null;
+            MapEntryValue value = mapEntries.get(commit.operation().key());
+            return value != null && value.type() != MapEntryValue.Type.TOMBSTONE;
         } finally {
             commit.close();
         }
@@ -151,9 +153,9 @@
      */
     protected boolean containsValue(Commit<? extends ContainsValue> commit) {
         try {
-            Match<byte[]> valueMatch = Match
-                    .ifValue(commit.operation().value());
+            Match<byte[]> valueMatch = Match.ifValue(commit.operation().value());
             return mapEntries.values().stream()
+                    .filter(value -> value.type() != MapEntryValue.Type.TOMBSTONE)
                     .anyMatch(value -> valueMatch.matches(value.value()));
         } finally {
             commit.close();
@@ -182,8 +184,14 @@
      */
     protected Versioned<byte[]> getOrDefault(Commit<? extends GetOrDefault> commit) {
         try {
-            Versioned<byte[]> value = toVersioned(mapEntries.get(commit.operation().key()));
-            return value != null ? value : new Versioned<>(commit.operation().defaultValue(), 0);
+            MapEntryValue value = mapEntries.get(commit.operation().key());
+            if (value == null) {
+                return new Versioned<>(commit.operation().defaultValue(), 0);
+            } else if (value.type() == MapEntryValue.Type.TOMBSTONE) {
+                return new Versioned<>(commit.operation().defaultValue(), value.version);
+            } else {
+                return new Versioned<>(value.value(), value.version);
+            }
         } finally {
             commit.close();
         }
@@ -197,7 +205,9 @@
      */
     protected int size(Commit<? extends Size> commit) {
         try {
-            return mapEntries.size();
+            return (int) mapEntries.values().stream()
+                    .filter(value -> value.type() != MapEntryValue.Type.TOMBSTONE)
+                    .count();
         } finally {
             commit.close();
         }
@@ -211,7 +221,8 @@
      */
     protected boolean isEmpty(Commit<? extends IsEmpty> commit) {
         try {
-            return mapEntries.isEmpty();
+            return mapEntries.values().stream()
+                    .noneMatch(value -> value.type() != MapEntryValue.Type.TOMBSTONE);
         } finally {
             commit.close();
         }
@@ -225,7 +236,10 @@
      */
     protected Set<String> keySet(Commit<? extends KeySet> commit) {
         try {
-            return mapEntries.keySet().stream().collect(Collectors.toSet());
+            return mapEntries.entrySet().stream()
+                    .filter(entry -> entry.getValue().type() != MapEntryValue.Type.TOMBSTONE)
+                    .map(Map.Entry::getKey)
+                    .collect(Collectors.toSet());
         } finally {
             commit.close();
         }
@@ -239,7 +253,10 @@
      */
     protected Collection<Versioned<byte[]>> values(Commit<? extends Values> commit) {
         try {
-            return mapEntries.values().stream().map(this::toVersioned).collect(Collectors.toList());
+            return mapEntries.entrySet().stream()
+                    .filter(entry -> entry.getValue().type() != MapEntryValue.Type.TOMBSTONE)
+                    .map(entry -> toVersioned(entry.getValue()))
+                    .collect(Collectors.toList());
         } finally {
             commit.close();
         }
@@ -254,11 +271,9 @@
      */
     protected Set<Map.Entry<String, Versioned<byte[]>>> entrySet(Commit<? extends EntrySet> commit) {
         try {
-            return mapEntries
-                    .entrySet()
-                    .stream()
-                    .map(e -> Maps.immutableEntry(e.getKey(),
-                            toVersioned(e.getValue())))
+            return mapEntries.entrySet().stream()
+                    .filter(entry -> entry.getValue().type() != MapEntryValue.Type.TOMBSTONE)
+                    .map(e -> Maps.immutableEntry(e.getKey(), toVersioned(e.getValue())))
                     .collect(Collectors.toSet());
         } finally {
             commit.close();
@@ -285,21 +300,34 @@
             }
 
             byte[] newValue = commit.operation().value();
-            long newVersion = commit.index();
+            currentVersion = commit.index();
             Versioned<byte[]> newMapValue = newValue == null ? null
-                    : new Versioned<>(newValue, newVersion);
+                    : new Versioned<>(newValue, currentVersion);
 
             MapEvent.Type updateType = newValue == null ? REMOVE
                     : oldCommitValue == null ? INSERT : UPDATE;
+
+            // If a value existed in the map, remove and discard the value to ensure disk can be freed.
             if (updateType == REMOVE || updateType == UPDATE) {
                 mapEntries.remove(key);
                 oldCommitValue.discard();
             }
+
+            // If this is an insert/update commit, add the commit to the map entries.
             if (updateType == INSERT || updateType == UPDATE) {
-                mapEntries.put(key, new NonTransactionalCommit(newVersion, commit));
+                mapEntries.put(key, new NonTransactionalCommit(commit));
+            } else if (!activeTransactions.isEmpty()) {
+                // If this is a delete but transactions are currently running, ensure tombstones are retained
+                // for version checks.
+                TombstoneCommit tombstone = new TombstoneCommit(
+                        commit.index(),
+                        new CountDownCompleter<>(commit, 1, Commit::close));
+                mapEntries.put(key, tombstone);
             } else {
+                // If no transactions are in progress, we can safely delete the key from memory.
                 commit.close();
             }
+
             publish(Lists.newArrayList(new MapEvent<>("", key, newMapValue, oldMapValue)));
             return new MapEntryUpdateResult<>(updateStatus, "", key, oldMapValue,
                     newMapValue);
@@ -383,7 +411,9 @@
      */
     protected long begin(Commit<? extends TransactionBegin> commit) {
         try {
-            return commit.index();
+            long version = commit.index();
+            activeTransactions.put(commit.operation().transactionId(), new TransactionScope(version));
+            return version;
         } finally {
             commit.close();
         }
@@ -397,10 +427,18 @@
      */
     protected PrepareResult prepareAndCommit(Commit<? extends TransactionPrepareAndCommit> commit) {
         PrepareResult prepareResult = prepare(commit);
+        TransactionScope transactionScope =
+                activeTransactions.remove(commit.operation().transactionLog().transactionId());
+        this.currentVersion = commit.index();
         if (prepareResult == PrepareResult.OK) {
-            commitInternal(commit.operation().transactionLog().transactionId(), commit.index());
+            transactionScope = transactionScope.prepared(commit);
+            commit(transactionScope);
+        } else {
+            transactionScope.close();
         }
+        discardTombstones();
         return prepareResult;
+
     }
 
     /**
@@ -411,30 +449,72 @@
      */
     protected PrepareResult prepare(Commit<? extends TransactionPrepare> commit) {
         boolean ok = false;
+
         try {
-            TransactionLog<MapUpdate<String, byte[]>> transaction = commit.operation().transactionLog();
-            for (MapUpdate<String, byte[]> update : transaction.records()) {
-                String key = update.key();
+            TransactionLog<MapUpdate<String, byte[]>> transactionLog = commit.operation().transactionLog();
+
+            // Iterate through records in the transaction log and perform isolation checks.
+            for (MapUpdate<String, byte[]> record : transactionLog.records()) {
+                String key = record.key();
+
+                // If the record is a VERSION_MATCH then check that the record's version matches the current
+                // version of the state machine.
+                if (record.type() == MapUpdate.Type.VERSION_MATCH && key == null) {
+                    if (record.version() > currentVersion) {
+                        return PrepareResult.OPTIMISTIC_LOCK_FAILURE;
+                    } else {
+                        continue;
+                    }
+                }
+
+                // If the prepared keys already contains the key contained within the record, that indicates a
+                // conflict with a concurrent transaction.
                 if (preparedKeys.contains(key)) {
                     return PrepareResult.CONCURRENT_TRANSACTION;
                 }
+
+                // Read the existing value from the map.
                 MapEntryValue existingValue = mapEntries.get(key);
+
+                // Note: if the existing value is null, that means the key has not changed during the transaction,
+                // otherwise a tombstone would have been retained.
                 if (existingValue == null) {
-                    if (update.type() != MapUpdate.Type.PUT_IF_ABSENT) {
+                    // If the value is null, ensure the version is equal to the transaction version.
+                    if (record.version() != transactionLog.version()) {
                         return PrepareResult.OPTIMISTIC_LOCK_FAILURE;
                     }
                 } else {
-                    if (existingValue.version() != update.currentVersion()) {
+                    // If the value is non-null, compare the current version with the record version.
+                    if (existingValue.version() > record.version()) {
                         return PrepareResult.OPTIMISTIC_LOCK_FAILURE;
                     }
                 }
             }
-            // No violations detected. Add to pendingTransactions and mark
-            // modified keys as locked for updates.
-            pendingTransactions.put(transaction.transactionId(), commit);
-            transaction.records().forEach(u -> preparedKeys.add(u.key()));
+
+            // No violations detected. Mark modified keys locked for transactions.
+            transactionLog.records().forEach(record -> {
+                if (record.type() != MapUpdate.Type.VERSION_MATCH) {
+                    preparedKeys.add(record.key());
+                }
+            });
+
             ok = true;
-            return PrepareResult.OK;
+
+            // Update the transaction scope. If the transaction scope is not set on this node, that indicates the
+            // coordinator is communicating with another node. Transactions assume that the client is communicating
+            // with a single leader in order to limit the overhead of retaining tombstones.
+            TransactionScope transactionScope = activeTransactions.get(transactionLog.transactionId());
+            if (transactionScope == null) {
+                activeTransactions.put(
+                        transactionLog.transactionId(),
+                        new TransactionScope(transactionLog.version(), commit));
+                return PrepareResult.PARTIAL_FAILURE;
+            } else {
+                activeTransactions.put(
+                        transactionLog.transactionId(),
+                        transactionScope.prepared(commit));
+                return PrepareResult.OK;
+            }
         } catch (Exception e) {
             log.warn("Failure applying {}", commit, e);
             throw Throwables.propagate(e);
@@ -453,43 +533,72 @@
      */
     protected CommitResult commit(Commit<? extends TransactionCommit> commit) {
         TransactionId transactionId = commit.operation().transactionId();
+        TransactionScope transactionScope = activeTransactions.remove(transactionId);
+        if (transactionScope == null) {
+            return CommitResult.UNKNOWN_TRANSACTION_ID;
+        }
+
         try {
-            return commitInternal(transactionId, commit.index());
+            this.currentVersion = commit.index();
+            return commit(transactionScope.committed(commit));
         } catch (Exception e) {
             log.warn("Failure applying {}", commit, e);
             throw Throwables.propagate(e);
         } finally {
-            commit.close();
+            discardTombstones();
         }
     }
 
-    private CommitResult commitInternal(TransactionId transactionId, long version) {
-        Commit<? extends TransactionPrepare> prepareCommit = pendingTransactions
-                .remove(transactionId);
-        if (prepareCommit == null) {
-            return CommitResult.UNKNOWN_TRANSACTION_ID;
-        }
-        TransactionLog<MapUpdate<String, byte[]>> transaction = prepareCommit.operation().transactionLog();
-        long totalReferencesToCommit = transaction
-                .records()
-                .stream()
-                .filter(update -> update.type() != MapUpdate.Type.REMOVE_IF_VERSION_MATCH)
+    /**
+     * Applies committed operations to the state machine.
+     */
+    private CommitResult commit(TransactionScope transactionScope) {
+        TransactionLog<MapUpdate<String, byte[]>> transactionLog = transactionScope.transactionLog();
+        boolean retainTombstones = !activeTransactions.isEmpty();
+
+        // Count the total number of keys that will be set by this transaction. This is necessary to do reference
+        // counting for garbage collection.
+        long totalReferencesToCommit = transactionLog.records().stream()
+                // No keys are set for version checks. For deletes, references are only retained of tombstones
+                // need to be retained for concurrent transactions.
+                .filter(record -> record.type() != MapUpdate.Type.VERSION_MATCH && record.type() != MapUpdate.Type.LOCK
+                        && (record.type() != MapUpdate.Type.REMOVE_IF_VERSION_MATCH || retainTombstones))
                 .count();
-        CountDownCompleter<Commit<? extends TransactionPrepare>> completer =
-                new CountDownCompleter<>(prepareCommit, totalReferencesToCommit, Commit::close);
+
+        // Create a count down completer that counts references to the transaction commit for garbage collection.
+        CountDownCompleter<TransactionScope> completer = new CountDownCompleter<>(
+                transactionScope, totalReferencesToCommit, TransactionScope::close);
+
         List<MapEvent<String, byte[]>> eventsToPublish = Lists.newArrayList();
-        for (MapUpdate<String, byte[]> update : transaction.records()) {
-            String key = update.key();
+        for (MapUpdate<String, byte[]> record : transactionLog.records()) {
+            if (record.type() == MapUpdate.Type.VERSION_MATCH) {
+                continue;
+            }
+
+            String key = record.key();
             checkState(preparedKeys.remove(key), "key is not prepared");
+
+            if (record.type() == MapUpdate.Type.LOCK) {
+                continue;
+            }
+
             MapEntryValue previousValue = mapEntries.remove(key);
             MapEntryValue newValue = null;
-            if (update.type() != MapUpdate.Type.REMOVE_IF_VERSION_MATCH) {
-                newValue = new TransactionalCommit(key, version, completer);
+
+            // If the record is not a delete, create a transactional commit.
+            if (record.type() != MapUpdate.Type.REMOVE_IF_VERSION_MATCH) {
+                newValue = new TransactionalCommit(currentVersion, record.value(), completer);
+            } else if (retainTombstones) {
+                // For deletes, if tombstones need to be retained then create and store a tombstone commit.
+                newValue = new TombstoneCommit(currentVersion, completer);
             }
+
             eventsToPublish.add(new MapEvent<>("", key, toVersioned(newValue), toVersioned(previousValue)));
+
             if (newValue != null) {
                 mapEntries.put(key, newValue);
             }
+
             if (previousValue != null) {
                 previousValue.discard();
             }
@@ -506,20 +615,57 @@
      */
     protected RollbackResult rollback(Commit<? extends TransactionRollback> commit) {
         TransactionId transactionId = commit.operation().transactionId();
-        try {
-            Commit<? extends TransactionPrepare> prepareCommit = pendingTransactions.remove(transactionId);
-            if (prepareCommit == null) {
-                return RollbackResult.UNKNOWN_TRANSACTION_ID;
-            } else {
-                prepareCommit.operation()
-                             .transactionLog()
-                             .records()
-                             .forEach(u -> preparedKeys.remove(u.key()));
-                prepareCommit.close();
-                return RollbackResult.OK;
-            }
-        } finally {
+        TransactionScope transactionScope = activeTransactions.remove(transactionId);
+        if (transactionScope == null) {
+            return RollbackResult.UNKNOWN_TRANSACTION_ID;
+        } else if (!transactionScope.isPrepared()) {
+            discardTombstones();
+            transactionScope.close();
             commit.close();
+            return RollbackResult.OK;
+        } else {
+            try {
+                transactionScope.transactionLog().records()
+                        .forEach(record -> {
+                            if (record.type() != MapUpdate.Type.VERSION_MATCH) {
+                                preparedKeys.remove(record.key());
+                            }
+                        });
+                return RollbackResult.OK;
+            } finally {
+                discardTombstones();
+                transactionScope.close();
+                commit.close();
+            }
+        }
+
+    }
+
+    /**
+     * Discards tombstones no longer needed by active transactions.
+     */
+    private void discardTombstones() {
+        if (activeTransactions.isEmpty()) {
+            Iterator<Map.Entry<String, MapEntryValue>> iterator = mapEntries.entrySet().iterator();
+            while (iterator.hasNext()) {
+                MapEntryValue value = iterator.next().getValue();
+                if (value.type() == MapEntryValue.Type.TOMBSTONE) {
+                    iterator.remove();
+                    value.discard();
+                }
+            }
+        } else {
+            long lowWaterMark = activeTransactions.values().stream()
+                    .mapToLong(TransactionScope::version)
+                    .min().getAsLong();
+            Iterator<Map.Entry<String, MapEntryValue>> iterator = mapEntries.entrySet().iterator();
+            while (iterator.hasNext()) {
+                MapEntryValue value = iterator.next().getValue();
+                if (value.type() == MapEntryValue.Type.TOMBSTONE && value.version < lowWaterMark) {
+                    iterator.remove();
+                    value.discard();
+                }
+            }
         }
     }
 
@@ -553,7 +699,8 @@
      * @return versioned instance
      */
     private Versioned<byte[]> toVersioned(MapEntryValue value) {
-        return value == null ? null : new Versioned<>(value.value(), value.version());
+        return value != null && value.type() != MapEntryValue.Type.TOMBSTONE
+                ? new Versioned<>(value.value(), value.version()) : null;
     }
 
     /**
@@ -594,52 +741,73 @@
     /**
      * Interface implemented by map values.
      */
-    private interface MapEntryValue {
+    private abstract static class MapEntryValue {
+        protected final Type type;
+        protected final long version;
+
+        MapEntryValue(Type type, long version) {
+            this.type = type;
+            this.version = version;
+        }
+
         /**
-         * Returns the raw {@code byte[]}.
+         * Returns the value type.
          *
-         * @return raw value
+         * @return the value type
          */
-        byte[] value();
+        Type type() {
+            return type;
+        }
 
         /**
          * Returns the version of the value.
          *
          * @return version
          */
-        long version();
+        long version() {
+            return version;
+        }
+
+        /**
+         * Returns the raw {@code byte[]}.
+         *
+         * @return raw value
+         */
+        abstract byte[] value();
 
         /**
          * Discards the value by invoke appropriate clean up actions.
          */
-        void discard();
+        abstract void discard();
+
+        /**
+         * Value type.
+         */
+        enum Type {
+            VALUE,
+            TOMBSTONE,
+        }
     }
 
     /**
      * A {@code MapEntryValue} that is derived from a non-transactional update
      * i.e. via any standard map update operation.
      */
-    private class NonTransactionalCommit implements MapEntryValue {
-        private final long version;
+    private static class NonTransactionalCommit extends MapEntryValue {
         private final Commit<? extends UpdateAndGet> commit;
 
-        public NonTransactionalCommit(long version, Commit<? extends UpdateAndGet> commit) {
-            this.version = version;
+        NonTransactionalCommit(Commit<? extends UpdateAndGet> commit) {
+            super(Type.VALUE, commit.index());
             this.commit = commit;
         }
 
         @Override
-        public byte[] value() {
+        byte[] value() {
             return commit.operation().value();
         }
 
         @Override
-        public long version() {
-            return version;
-        }
-
-        @Override
-        public void discard() {
+        void discard() {
             commit.close();
         }
     }
@@ -648,43 +816,135 @@
      * A {@code MapEntryValue} that is derived from updates submitted via a
      * transaction.
      */
-    private class TransactionalCommit implements MapEntryValue {
-        private final String key;
+    private static class TransactionalCommit extends MapEntryValue {
+        private final byte[] value;
+        private final CountDownCompleter<?> completer;
+
+        TransactionalCommit(long version, byte[] value, CountDownCompleter<?> completer) {
+            super(Type.VALUE, version);
+            this.value = value;
+            this.completer = completer;
+        }
+
+        @Override
+        byte[] value() {
+            return value;
+        }
+
+        @Override
+        void discard() {
+            completer.countDown();
+        }
+    }
+
+    /**
+     * A {@code MapEntryValue} that represents a deleted entry.
+     */
+    private static class TombstoneCommit extends MapEntryValue {
+        private final CountDownCompleter<?> completer;
+
+        public TombstoneCommit(long version, CountDownCompleter<?> completer) {
+            super(Type.TOMBSTONE, version);
+            this.completer = completer;
+        }
+
+        @Override
+        byte[] value() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        void discard() {
+            completer.countDown();
+        }
+    }
+
+    /**
+     * Map transaction scope.
+     */
+    private static final class TransactionScope {
         private final long version;
-        private final CountDownCompleter<Commit<? extends TransactionPrepare>> completer;
+        private final Commit<? extends TransactionPrepare> prepareCommit;
+        private final Commit<? extends TransactionCommit> commitCommit;
 
-        public TransactionalCommit(
-                String key,
+        private TransactionScope(long version) {
+            this(version, null, null);
+        }
+
+        private TransactionScope(
                 long version,
-                CountDownCompleter<Commit<? extends TransactionPrepare>> commit) {
-            this.key = key;
+                Commit<? extends TransactionPrepare> prepareCommit) {
+            this(version, prepareCommit, null);
+        }
+
+        private TransactionScope(
+                long version,
+                Commit<? extends TransactionPrepare> prepareCommit,
+                Commit<? extends TransactionCommit> commitCommit) {
             this.version = version;
-            this.completer = commit;
+            this.prepareCommit = prepareCommit;
+            this.commitCommit = commitCommit;
         }
 
-        @Override
-        public byte[] value() {
-            TransactionLog<MapUpdate<String, byte[]>> transaction = completer.object().operation().transactionLog();
-            return valueForKey(key, transaction);
-        }
-
-        @Override
-        public long version() {
+        /**
+         * Returns the transaction version.
+         *
+         * @return the transaction version
+         */
+        long version() {
             return version;
         }
 
-        @Override
-        public void discard() {
-            completer.countDown();
+        /**
+         * Returns whether this is a prepared transaction scope.
+         *
+         * @return whether this is a prepared transaction scope
+         */
+        boolean isPrepared() {
+            return prepareCommit != null;
         }
 
-        private byte[] valueForKey(String key, TransactionLog<MapUpdate<String, byte[]>> transaction) {
-            MapUpdate<String, byte[]>  update = transaction.records()
-                                                           .stream()
-                                                           .filter(u -> u.key().equals(key))
-                                                           .findFirst()
-                                                           .orElse(null);
-            return update == null ? null : update.value();
+        /**
+         * Returns the transaction commit log.
+         *
+         * @return the transaction commit log
+         */
+        TransactionLog<MapUpdate<String, byte[]>> transactionLog() {
+            checkState(isPrepared());
+            return prepareCommit.operation().transactionLog();
+        }
+
+        /**
+         * Returns a new transaction scope with a prepare commit.
+         *
+         * @param commit the prepare commit
+         * @return new transaction scope updated with the prepare commit
+         */
+        TransactionScope prepared(Commit<? extends TransactionPrepare> commit) {
+            return new TransactionScope(version, commit);
+        }
+
+        /**
+         * Returns a new transaction scope with a commit commit.
+         *
+         * @param commit the commit commit ;-)
+         * @return new transaction scope updated with the commit commit
+         */
+        TransactionScope committed(Commit<? extends TransactionCommit> commit) {
+            checkState(isPrepared());
+            return new TransactionScope(version, prepareCommit, commit);
+        }
+
+        /**
+         * Closes the transaction and all associated commits.
+         */
+        void close() {
+            if (prepareCommit != null) {
+                prepareCommit.close();
+            }
+            if (commitCommit != null) {
+                commitCommit.close();
+            }
         }
     }
 }
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
index 533ea97..ef03935 100644
--- 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
@@ -74,7 +74,7 @@
 
         expect(asyncMap.begin(transactionId))
                 .andReturn(CompletableFuture.completedFuture(new Version(1)));
-        expect(asyncMap.prepare(new TransactionLog<>(transactionId, updates)))
+        expect(asyncMap.prepare(new TransactionLog<>(transactionId, 1, updates)))
                 .andReturn(CompletableFuture.completedFuture(true));
         expect(asyncMap.commit(transactionId))
                 .andReturn(CompletableFuture.completedFuture(null));
@@ -117,7 +117,7 @@
 
         expect(asyncMap.begin(transactionId))
                 .andReturn(CompletableFuture.completedFuture(new Version(1)));
-        expect(asyncMap.prepare(new TransactionLog<>(transactionId, updates)))
+        expect(asyncMap.prepare(new TransactionLog<>(transactionId, 1, updates)))
                 .andReturn(CompletableFuture.completedFuture(true));
         replay(asyncMap);
 
@@ -160,16 +160,16 @@
                 .andReturn(CompletableFuture.completedFuture(new Version(1)));
 
         expect(participants.get(PartitionId.from(1)).transaction.transactionalObject.prepare(
-                new TransactionLog<>(transactionId, Arrays.asList(
+                new TransactionLog<>(transactionId, 1, Arrays.asList(
                         MapUpdate.<String, String>newBuilder()
                                 .withType(MapUpdate.Type.REMOVE_IF_VERSION_MATCH)
                                 .withKey("foo")
-                                .withCurrentVersion(1)
+                                .withVersion(1)
                                 .build(),
                         MapUpdate.<String, String>newBuilder()
                                 .withType(MapUpdate.Type.REMOVE_IF_VERSION_MATCH)
                                 .withKey("baz")
-                                .withCurrentVersion(2)
+                                .withVersion(2)
                                 .build()
                 )))).andReturn(CompletableFuture.completedFuture(true));
 
@@ -181,11 +181,12 @@
                 .andReturn(CompletableFuture.completedFuture(new Version(1)));
 
         expect(participants.get(PartitionId.from(3)).transaction.transactionalObject.prepare(
-                new TransactionLog<>(transactionId, Arrays.asList(
+                new TransactionLog<>(transactionId, 1, Arrays.asList(
                         MapUpdate.<String, String>newBuilder()
-                                .withType(MapUpdate.Type.PUT_IF_ABSENT)
+                                .withType(MapUpdate.Type.PUT_IF_VERSION_MATCH)
                                 .withKey("bar")
                                 .withValue("baz")
+                                .withVersion(1)
                                 .build()
                 )))).andReturn(CompletableFuture.completedFuture(true));
 
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 ac4c60d..284b57b 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
@@ -27,6 +27,7 @@
 import org.onosproject.store.service.MapEvent;
 import org.onosproject.store.service.MapEventListener;
 import org.onosproject.store.service.TransactionLog;
+import org.onosproject.store.service.Version;
 import org.onosproject.store.service.Versioned;
 
 import java.util.Arrays;
@@ -41,7 +42,6 @@
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
@@ -92,6 +92,14 @@
     }
 
     /**
+     * Tests map transaction prepare.
+     */
+    @Test
+    public void testTransactionPrepare() throws Throwable {
+        transactionPrepareTests();
+    }
+
+    /**
      * Tests map transaction commit.
      */
     @Test
@@ -112,22 +120,12 @@
         final byte[] rawBarValue = Tools.getBytesUtf8("Hello bar!");
 
         AtomixConsistentMap map = createAtomixClient().getResource("testBasicMapOperationMap",
-                                                                   AtomixConsistentMap.class).join();
+                AtomixConsistentMap.class).join();
 
         map.isEmpty().thenAccept(result -> {
             assertTrue(result);
         }).join();
 
-        map.getOrDefault("nothing", null).thenAccept(result -> {
-            assertEquals(0, result.version());
-            assertNull(result.value());
-        }).join();
-
-        map.getOrDefault("foo", "bar".getBytes()).thenAccept(result -> {
-            assertEquals(0, result.version());
-            assertArrayEquals("bar".getBytes(), result.value());
-        }).join();
-
         map.put("foo", rawFooValue).thenAccept(result -> {
             assertNull(result);
         }).join();
@@ -175,11 +173,6 @@
             assertTrue(Arrays.equals(Versioned.valueOrElse(result, null), rawFooValue));
         }).join();
 
-        map.getOrDefault("foo", "bar".getBytes()).thenAccept(result -> {
-            assertNotEquals(0, result.version());
-            assertArrayEquals(rawFooValue, result.value());
-        }).join();
-
         map.remove("foo").thenAccept(result -> {
             assertTrue(Arrays.equals(Versioned.valueOrElse(result, null), rawFooValue));
         }).join();
@@ -257,7 +250,7 @@
         final byte[] value3 = Tools.getBytesUtf8("value3");
 
         AtomixConsistentMap map = createAtomixClient().getResource("testMapComputeOperationsMap",
-                                                                   AtomixConsistentMap.class).join();
+                AtomixConsistentMap.class).join();
 
         map.computeIfAbsent("foo", k -> value1).thenAccept(result -> {
             assertTrue(Arrays.equals(Versioned.valueOrElse(result, null), value1));
@@ -295,7 +288,7 @@
         final byte[] value3 = Tools.getBytesUtf8("value3");
 
         AtomixConsistentMap map = createAtomixClient().getResource("testMapListenerMap",
-                                                                   AtomixConsistentMap.class).join();
+                AtomixConsistentMap.class).join();
         TestMapEventListener listener = new TestMapEventListener();
 
         // add listener; insert new value into map and verify an INSERT event is received.
@@ -349,27 +342,104 @@
         map.removeListener(listener).join();
     }
 
+    protected void transactionPrepareTests() throws Throwable {
+        AtomixConsistentMap map = createAtomixClient().getResource("testPrepareTestsMap",
+                AtomixConsistentMap.class).join();
+
+        TransactionId transactionId1 = TransactionId.from("tx1");
+        TransactionId transactionId2 = TransactionId.from("tx2");
+        TransactionId transactionId3 = TransactionId.from("tx3");
+        TransactionId transactionId4 = TransactionId.from("tx4");
+
+        Version lock1 = map.begin(transactionId1).join();
+
+        MapUpdate<String, byte[]> update1 =
+                MapUpdate.<String, byte[]>newBuilder()
+                        .withType(MapUpdate.Type.LOCK)
+                        .withKey("foo")
+                        .withVersion(lock1.value())
+                        .build();
+        MapUpdate<String, byte[]> update2 =
+                MapUpdate.<String, byte[]>newBuilder()
+                        .withType(MapUpdate.Type.LOCK)
+                        .withKey("bar")
+                        .withVersion(lock1.value())
+                        .build();
+
+        map.prepare(new TransactionLog<>(transactionId1, lock1.value(), Arrays.asList(update1, update2)))
+                .thenAccept(result -> {
+                    assertTrue(result);
+                }).join();
+
+        Version lock2 = map.begin(transactionId2).join();
+
+        MapUpdate<String, byte[]> update3 =
+                MapUpdate.<String, byte[]>newBuilder()
+                        .withType(MapUpdate.Type.LOCK)
+                        .withKey("foo")
+                        .withVersion(lock2.value())
+                        .build();
+
+        map.prepare(new TransactionLog<>(transactionId2, lock2.value(), Arrays.asList(update3)))
+                .thenAccept(result -> {
+                    assertFalse(result);
+                }).join();
+        map.rollback(transactionId2).join();
+
+        Version lock3 = map.begin(transactionId3).join();
+
+        MapUpdate<String, byte[]> update4 =
+                MapUpdate.<String, byte[]>newBuilder()
+                        .withType(MapUpdate.Type.LOCK)
+                        .withKey("baz")
+                        .withVersion(0)
+                        .build();
+
+        map.prepare(new TransactionLog<>(transactionId3, lock3.value(), Arrays.asList(update4)))
+                .thenAccept(result -> {
+                    assertFalse(result);
+                }).join();
+        map.rollback(transactionId3).join();
+
+        Version lock4 = map.begin(transactionId4).join();
+
+        MapUpdate<String, byte[]> update5 =
+                MapUpdate.<String, byte[]>newBuilder()
+                        .withType(MapUpdate.Type.LOCK)
+                        .withKey("baz")
+                        .withVersion(lock4.value())
+                        .build();
+
+        map.prepare(new TransactionLog<>(transactionId4, lock4.value(), Arrays.asList(update5)))
+                .thenAccept(result -> {
+                    assertTrue(result);
+                }).join();
+    }
+
     protected void transactionCommitTests() throws Throwable {
         final byte[] value1 = Tools.getBytesUtf8("value1");
         final byte[] value2 = Tools.getBytesUtf8("value2");
 
         AtomixConsistentMap map = createAtomixClient().getResource("testCommitTestsMap",
-                                                                   AtomixConsistentMap.class).join();
+                AtomixConsistentMap.class).join();
         TestMapEventListener listener = new TestMapEventListener();
 
         map.addListener(listener).join();
 
-        // PUT_IF_ABSENT
+        TransactionId transactionId = TransactionId.from("tx1");
+
+        // Begin the transaction.
+        Version lock = map.begin(transactionId).join();
+
+        // PUT_IF_VERSION_MATCH
         MapUpdate<String, byte[]> update1 =
-                MapUpdate.<String, byte[]>newBuilder().withType(MapUpdate.Type.PUT_IF_ABSENT)
-                .withKey("foo")
-                .withValue(value1)
-                .build();
+                MapUpdate.<String, byte[]>newBuilder().withType(MapUpdate.Type.PUT_IF_VERSION_MATCH)
+                        .withKey("foo")
+                        .withValue(value1)
+                        .withVersion(lock.value())
+                        .build();
 
-        TransactionLog<MapUpdate<String, byte[]>> tx =
-                new TransactionLog<>(TransactionId.from("tx1"), Arrays.asList(update1));
-
-        map.prepare(tx).thenAccept(result -> {
+        map.prepare(new TransactionLog<>(transactionId, lock.value(), Arrays.asList(update1))).thenAccept(result -> {
             assertEquals(true, result);
         }).join();
         // verify changes in Tx is not visible yet until commit
@@ -392,7 +462,7 @@
 
         assertFalse(listener.eventReceived());
 
-        map.commit(tx.transactionId()).join();
+        map.commit(transactionId).join();
         MapEvent<String, byte[]> event = listener.event();
         assertNotNull(event);
         assertEquals(MapEvent.Type.INSERT, event.type());
@@ -407,19 +477,21 @@
         assertEquals(MapEvent.Type.UPDATE, event.type());
         assertTrue(Arrays.equals(value2, event.newValue().value()));
 
-
         // REMOVE_IF_VERSION_MATCH
         byte[] currFoo = map.get("foo").get().value();
         long currFooVersion = map.get("foo").get().version();
         MapUpdate<String, byte[]> remove1 =
                 MapUpdate.<String, byte[]>newBuilder().withType(MapUpdate.Type.REMOVE_IF_VERSION_MATCH)
-                .withKey("foo")
-                .withCurrentVersion(currFooVersion)
-                .build();
+                        .withKey("foo")
+                        .withVersion(currFooVersion)
+                        .build();
 
-        tx = new TransactionLog<>(TransactionId.from("tx2"), Arrays.asList(remove1));
+        transactionId = TransactionId.from("tx2");
 
-        map.prepare(tx).thenAccept(result -> {
+        // Begin the transaction.
+        map.begin(transactionId).join();
+
+        map.prepare(new TransactionLog<>(transactionId, lock.value(), Arrays.asList(remove1))).thenAccept(result -> {
             assertTrue("prepare should succeed", result);
         }).join();
         // verify changes in Tx is not visible yet until commit
@@ -433,7 +505,7 @@
             assertThat(result.value(), is(currFoo));
         }).join();
 
-        map.commit(tx.transactionId()).join();
+        map.commit(transactionId).join();
         event = listener.event();
         assertNotNull(event);
         assertEquals(MapEvent.Type.REMOVE, event.type());
@@ -450,24 +522,28 @@
         final byte[] value2 = Tools.getBytesUtf8("value2");
 
         AtomixConsistentMap map = createAtomixClient().getResource("testTransactionRollbackTestsMap",
-                                                                   AtomixConsistentMap.class).join();
+                AtomixConsistentMap.class).join();
         TestMapEventListener listener = new TestMapEventListener();
 
         map.addListener(listener).join();
 
+        TransactionId transactionId = TransactionId.from("tx1");
+
+        Version lock = map.begin(transactionId).join();
+
         MapUpdate<String, byte[]> update1 =
-                MapUpdate.<String, byte[]>newBuilder().withType(MapUpdate.Type.PUT_IF_ABSENT)
-                .withKey("foo")
-                .withValue(value1)
-                .build();
-        TransactionLog<MapUpdate<String, byte[]>> tx =
-                new TransactionLog<>(TransactionId.from("tx1"), Arrays.asList(update1));
-        map.prepare(tx).thenAccept(result -> {
+                MapUpdate.<String, byte[]>newBuilder().withType(MapUpdate.Type.PUT_IF_VERSION_MATCH)
+                        .withKey("foo")
+                        .withValue(value1)
+                        .withVersion(lock.value())
+                        .build();
+
+        map.prepare(new TransactionLog<>(transactionId, lock.value(), Arrays.asList(update1))).thenAccept(result -> {
             assertEquals(true, result);
         }).join();
         assertFalse(listener.eventReceived());
 
-        map.rollback(tx.transactionId()).join();
+        map.rollback(transactionId).join();
         assertFalse(listener.eventReceived());
 
         map.get("foo").thenAccept(result -> {