diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java
index 00b185d..b09e000 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java
@@ -39,8 +39,9 @@
 import org.apache.asterix.common.config.DatasetConfig.DatasetType;
 import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable;
 import org.apache.asterix.common.ioopcallbacks.AbstractLSMIOOperationCallback;
-import org.apache.asterix.common.transactions.DatasetId;
 import org.apache.asterix.common.transactions.ITransactionContext;
+import org.apache.asterix.common.transactions.ITransactionManager;
+import org.apache.asterix.common.transactions.TransactionOptions;
 import org.apache.asterix.external.util.DataflowUtils;
 import org.apache.asterix.file.StorageComponentProvider;
 import org.apache.asterix.metadata.entities.Dataset;
@@ -146,7 +147,8 @@
         lsmBtree = (TestLsmBtree) indexDataflowHelper.getIndexInstance();
         indexDataflowHelper.close();
         nc.newJobId();
-        txnCtx = nc.getTransactionManager().getTransactionContext(nc.getTxnJobId(ctx), true);
+        txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
+                new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
         insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, KEY_INDEXES,
                 KEY_INDICATORS_LIST, storageManager).getLeft();
     }
@@ -188,7 +190,7 @@
                 tupleAppender.write(insertOp, true);
             }
             insertOp.close();
-            nc.getTransactionManager().completedTransaction(txnCtx, DatasetId.NULL, -1, true);
+            nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
 
             // get all components
             List<ILSMMemoryComponent> memComponents = lsmBtree.getMemoryComponents();
@@ -239,8 +241,7 @@
                 tupleAppender.write(insertOp, true);
             }
             insertOp.close();
-            nc.getTransactionManager().completedTransaction(txnCtx, DatasetId.NULL, -1, true);
-
+            nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
             // get all components
             List<ILSMMemoryComponent> memComponents = lsmBtree.getMemoryComponents();
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
@@ -255,7 +256,8 @@
 
             // insert again
             nc.newJobId();
-            txnCtx = nc.getTransactionManager().getTransactionContext(nc.getTxnJobId(ctx), true);
+            txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
+                    new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
             insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, KEY_INDEXES,
                     KEY_INDICATORS_LIST, storageManager).getLeft();
             insertOp.open();
@@ -267,7 +269,7 @@
                 tupleAppender.write(insertOp, true);
             }
             insertOp.close();
-            nc.getTransactionManager().completedTransaction(txnCtx, DatasetId.NULL, -1, true);
+            nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
             searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
             // rollback the last disk component
             lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
@@ -308,7 +310,7 @@
                 tupleAppender.write(insertOp, true);
             }
             insertOp.close();
-            nc.getTransactionManager().completedTransaction(txnCtx, DatasetId.NULL, -1, true);
+            nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
 
             // get all components
             List<ILSMMemoryComponent> memComponents = lsmBtree.getMemoryComponents();
@@ -378,7 +380,7 @@
                 tupleAppender.write(insertOp, true);
             }
             insertOp.close();
-            nc.getTransactionManager().completedTransaction(txnCtx, DatasetId.NULL, -1, true);
+            nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
             // get all components
             List<ILSMMemoryComponent> memComponents = lsmBtree.getMemoryComponents();
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
@@ -434,7 +436,7 @@
                 tupleAppender.write(insertOp, true);
             }
             insertOp.close();
-            nc.getTransactionManager().completedTransaction(txnCtx, DatasetId.NULL, -1, true);
+            nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
             // get all components
             List<ILSMMemoryComponent> memComponents = lsmBtree.getMemoryComponents();
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
@@ -498,7 +500,7 @@
                 tupleAppender.write(insertOp, true);
             }
             insertOp.close();
-            nc.getTransactionManager().completedTransaction(txnCtx, DatasetId.NULL, -1, true);
+            nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
             // get all components
             List<ILSMMemoryComponent> memComponents = lsmBtree.getMemoryComponents();
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
@@ -558,7 +560,7 @@
                 tupleAppender.write(insertOp, true);
             }
             insertOp.close();
-            nc.getTransactionManager().completedTransaction(txnCtx, DatasetId.NULL, -1, true);
+            nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
             // get all components
             List<ILSMMemoryComponent> memComponents = lsmBtree.getMemoryComponents();
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
@@ -620,7 +622,7 @@
                 tupleAppender.write(insertOp, true);
             }
             insertOp.close();
-            nc.getTransactionManager().completedTransaction(txnCtx, DatasetId.NULL, -1, true);
+            nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
             // get all components
             List<ILSMMemoryComponent> memComponents = lsmBtree.getMemoryComponents();
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
@@ -691,7 +693,7 @@
                 tupleAppender.write(insertOp, true);
             }
             insertOp.close();
-            nc.getTransactionManager().completedTransaction(txnCtx, DatasetId.NULL, -1, true);
+            nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
             // get all components
             List<ILSMMemoryComponent> memComponents = lsmBtree.getMemoryComponents();
             List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java
index 82eb16a..1907ed6 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java
@@ -31,9 +31,10 @@
 import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
 import org.apache.asterix.common.config.DatasetConfig.DatasetType;
 import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable;
-import org.apache.asterix.common.transactions.DatasetId;
 import org.apache.asterix.common.transactions.ILogRecord;
 import org.apache.asterix.common.transactions.ITransactionContext;
+import org.apache.asterix.common.transactions.ITransactionManager;
+import org.apache.asterix.common.transactions.TransactionOptions;
 import org.apache.asterix.external.util.DataflowUtils;
 import org.apache.asterix.file.StorageComponentProvider;
 import org.apache.asterix.metadata.entities.Dataset;
@@ -119,8 +120,8 @@
                         storageManager, KEY_INDEXES, KEY_INDICATORS_LIST);
                 IHyracksTaskContext ctx = nc.createTestContext(true);
                 nc.newJobId();
-                ITransactionContext txnCtx =
-                        nc.getTransactionManager().getTransactionContext(nc.getTxnJobId(ctx), true);
+                ITransactionContext txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
+                        new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
                 LSMInsertDeleteOperatorNodePushable insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES,
                         RECORD_TYPE, META_TYPE, null, KEY_INDEXES, KEY_INDICATORS_LIST, storageManager).getLeft();
                 insertOp.open();
@@ -147,7 +148,7 @@
                     tupleAppender.write(insertOp, true);
                 }
                 insertOp.close();
-                nc.getTransactionManager().completedTransaction(txnCtx, DatasetId.NULL, -1, true);
+                nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
                 IndexDataflowHelperFactory iHelperFactory =
                         new IndexDataflowHelperFactory(nc.getStorageManager(), indexInfo.getFileSplitProvider());
                 IIndexDataflowHelper dataflowHelper =
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java
index 5384c92..1cc24d5 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java
@@ -34,11 +34,12 @@
 import org.apache.asterix.common.configuration.Property;
 import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable;
 import org.apache.asterix.common.transactions.Checkpoint;
-import org.apache.asterix.common.transactions.DatasetId;
 import org.apache.asterix.common.transactions.ICheckpointManager;
 import org.apache.asterix.common.transactions.IRecoveryManager;
 import org.apache.asterix.common.transactions.ITransactionContext;
+import org.apache.asterix.common.transactions.ITransactionManager;
 import org.apache.asterix.common.transactions.ITransactionSubsystem;
+import org.apache.asterix.common.transactions.TransactionOptions;
 import org.apache.asterix.external.util.DataflowUtils;
 import org.apache.asterix.file.StorageComponentProvider;
 import org.apache.asterix.metadata.entities.Dataset;
@@ -128,8 +129,8 @@
                         KEY_INDICATOR_LIST);
                 IHyracksTaskContext ctx = nc.createTestContext(false);
                 nc.newJobId();
-                ITransactionContext txnCtx =
-                        nc.getTransactionManager().getTransactionContext(nc.getTxnJobId(ctx), true);
+                ITransactionContext txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
+                        new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
                 // Prepare insert operation
                 LSMInsertDeleteOperatorNodePushable insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES,
                         RECORD_TYPE, META_TYPE, null, KEY_INDEXES, KEY_INDICATOR_LIST, storageManager).getLeft();
@@ -202,7 +203,7 @@
                     tupleAppender.write(insertOp, true);
                 }
                 insertOp.close();
-                nc.getTransactionManager().completedTransaction(txnCtx, DatasetId.NULL, -1, true);
+                nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
             } finally {
                 nc.deInit();
             }
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/metadata/MetadataTxnTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/metadata/MetadataTxnTest.java
index 3969ec5..3e906b4 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/metadata/MetadataTxnTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/metadata/MetadataTxnTest.java
@@ -21,11 +21,20 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.logging.ConsoleHandler;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
 import org.apache.asterix.api.common.AsterixHyracksIntegrationUtil;
+import org.apache.asterix.common.api.IDatasetLifecycleManager;
+import org.apache.asterix.common.api.INcApplicationContext;
+import org.apache.asterix.common.config.DatasetConfig;
 import org.apache.asterix.common.config.GlobalConfig;
+import org.apache.asterix.common.context.PrimaryIndexOperationTracker;
 import org.apache.asterix.common.dataflow.ICcApplicationContext;
 import org.apache.asterix.metadata.MetadataManager;
 import org.apache.asterix.metadata.MetadataTransactionContext;
@@ -36,7 +45,9 @@
 import org.apache.asterix.metadata.utils.DatasetUtil;
 import org.apache.asterix.test.common.TestExecutor;
 import org.apache.asterix.testframework.context.TestCaseContext;
+import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory;
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -137,4 +148,88 @@
             MetadataManager.INSTANCE.commitTransaction(readMdTxn);
         }
     }
+
+    @Test
+    public void concurrentMetadataTxn() throws Exception {
+        // get create type and dataset
+        String datasetName = "dataset1";
+        final TestCaseContext.OutputFormat format = TestCaseContext.OutputFormat.CLEAN_JSON;
+        testExecutor.executeSqlppUpdateOrDdl("CREATE TYPE KeyType AS { id: int };", format);
+        testExecutor.executeSqlppUpdateOrDdl("CREATE DATASET " + datasetName + "(KeyType) PRIMARY KEY id;", format);
+
+        // get created dataset
+        ICcApplicationContext appCtx =
+                (ICcApplicationContext) integrationUtil.getClusterControllerService().getApplicationContext();
+        MetadataProvider metadataProvider = new MetadataProvider(appCtx, null);
+        final MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
+        metadataProvider.setMetadataTxnContext(mdTxnCtx);
+        Dataset sourceDataset;
+        try {
+            sourceDataset = metadataProvider.findDataset(MetadataBuiltinEntities.DEFAULT_DATAVERSE_NAME, datasetName);
+            MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
+        } finally {
+            metadataProvider.getLocks().unlock();
+        }
+
+        /*
+         * Concurrently insert copies of the created dataset with
+         * different names and either commit or abort the transaction.
+         */
+        final AtomicInteger failCount = new AtomicInteger(0);
+        Thread transactor1 = new Thread(() -> IntStream.range(1, 100).forEach(x -> {
+            try {
+                addDataset(appCtx, sourceDataset, x, x % 2 == 0);
+            } catch (Exception e) {
+                e.printStackTrace();
+                failCount.incrementAndGet();
+            }
+        }));
+
+        Thread transactor2 = new Thread(() -> IntStream.range(101, 200).forEach(x -> {
+            try {
+                addDataset(appCtx, sourceDataset, x, x % 3 == 0);
+            } catch (Exception e) {
+                e.printStackTrace();
+                failCount.incrementAndGet();
+            }
+        }));
+
+        transactor1.start();
+        transactor2.start();
+        transactor1.join();
+        transactor2.join();
+
+        Assert.assertEquals(0, failCount.get());
+
+        // make sure all metadata indexes have no pending operations after all txns committed/aborted
+        final IDatasetLifecycleManager datasetLifecycleManager =
+                ((INcApplicationContext) integrationUtil.ncs[0].getApplicationContext()).getDatasetLifecycleManager();
+        int maxMetadatasetId = 14;
+        for (int i = 1; i <= maxMetadatasetId; i++) {
+            if (datasetLifecycleManager.getIndex(i, i) != null) {
+                final PrimaryIndexOperationTracker opTracker = datasetLifecycleManager.getOperationTracker(i);
+                Assert.assertEquals(0, opTracker.getNumActiveOperations());
+            }
+        }
+    }
+
+    private void addDataset(ICcApplicationContext appCtx, Dataset source, int datasetPostfix, boolean abort)
+            throws Exception {
+        Dataset dataset = new Dataset(source.getDataverseName(), "ds_" + datasetPostfix, source.getDataverseName(),
+                source.getDatasetType().name(), source.getNodeGroupName(), NoMergePolicyFactory.NAME, null,
+                source.getDatasetDetails(), source.getHints(), DatasetConfig.DatasetType.INTERNAL, datasetPostfix, 0);
+        MetadataProvider metadataProvider = new MetadataProvider(appCtx, null);
+        final MetadataTransactionContext writeTxn = MetadataManager.INSTANCE.beginTransaction();
+        metadataProvider.setMetadataTxnContext(writeTxn);
+        try {
+            MetadataManager.INSTANCE.addDataset(writeTxn, dataset);
+            if (abort) {
+                MetadataManager.INSTANCE.abortTransaction(writeTxn);
+            } else {
+                MetadataManager.INSTANCE.commitTransaction(writeTxn);
+            }
+        } finally {
+            metadataProvider.getLocks().unlock();
+        }
+    }
 }
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java
index eb47248..cb83d56 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java
@@ -128,8 +128,6 @@
                         KEY_INDICATOR_LIST);
                 IHyracksTaskContext ctx = nc.createTestContext(false);
                 nc.newJobId();
-                ITransactionContext txnCtx =
-                        nc.getTransactionManager().getTransactionContext(nc.getTxnJobId(ctx), true);
                 // Prepare insert operation
                 LSMInsertDeleteOperatorNodePushable insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES,
                         RECORD_TYPE, META_TYPE, null, KEY_INDEXES, KEY_INDICATOR_LIST, storageManager).getLeft();
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
index 6f35a3d..ababe9c 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
@@ -174,7 +174,7 @@
         //modificationCallback can be NoOpOperationCallback when redo/undo operations are executed.
         if (modificationCallback != NoOpOperationCallback.INSTANCE) {
             numActiveOperations.incrementAndGet();
-            ((AbstractOperationCallback) modificationCallback).incrementLocalNumActiveOperations();
+            ((AbstractOperationCallback) modificationCallback).beforeOperation();
         }
     }
 
@@ -182,7 +182,7 @@
         //modificationCallback can be NoOpOperationCallback when redo/undo operations are executed.
         if (modificationCallback != NoOpOperationCallback.INSTANCE) {
             numActiveOperations.decrementAndGet();
-            ((AbstractOperationCallback) modificationCallback).decrementLocalNumActiveOperations();
+            ((AbstractOperationCallback) modificationCallback).afterOperation();
         }
     }
 
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ACIDException.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ACIDException.java
index 7b489f2..77634eb 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ACIDException.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ACIDException.java
@@ -27,7 +27,7 @@
  * ACIDException encountered during crash recovery shall not have a transaction
  * context as recovery does not happen as part of a transaction.
  */
-public class ACIDException extends Exception {
+public class ACIDException extends RuntimeException {
 
     private static final long serialVersionUID = -8855848112541877323L;
     private ITransactionContext txnContext;
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/AbstractOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/AbstractOperationCallback.java
index 9844344..098bbd4 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/AbstractOperationCallback.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/AbstractOperationCallback.java
@@ -31,14 +31,16 @@
     protected final ITransactionContext txnCtx;
     protected final ILockManager lockManager;
     protected final long[] longHashes;
+    protected final long resourceId;
 
-    public AbstractOperationCallback(DatasetId datasetId, int[] primaryKeyFields, ITransactionContext txnCtx,
-            ILockManager lockManager) {
+    public AbstractOperationCallback(DatasetId datasetId, long resourceId, int[] primaryKeyFields,
+            ITransactionContext txnCtx, ILockManager lockManager) {
         this.datasetId = datasetId;
+        this.resourceId = resourceId;
         this.primaryKeyFields = primaryKeyFields;
         this.txnCtx = txnCtx;
         this.lockManager = lockManager;
-        this.longHashes = new long[2];
+        longHashes = new long[2];
     }
 
     public int computePrimaryKeyHashValue(ITupleReference tuple, int[] primaryKeyFields) {
@@ -46,12 +48,11 @@
         return Math.abs((int) longHashes[0]);
     }
 
-    public void incrementLocalNumActiveOperations() {
-        txnCtx.incrementNumActiveOperations();
+    public void beforeOperation() {
+        txnCtx.beforeOperation(resourceId);
     }
 
-    public void decrementLocalNumActiveOperations() {
-        txnCtx.decrementNumActiveOperations();
+    public void afterOperation() {
+        txnCtx.afterOperation(resourceId);
     }
-
 }
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ITransactionContext.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ITransactionContext.java
index 3dda5d3..c4a2d03 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ITransactionContext.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ITransactionContext.java
@@ -19,43 +19,132 @@
 package org.apache.asterix.common.transactions;
 
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+import org.apache.hyracks.storage.common.IModificationOperationCallback;
 
+/**
+ * A typical transaction lifecycle goes through the following steps:
+ * 1. {@link ITransactionContext#register(long, ILSMIndex, IModificationOperationCallback, boolean)}
+ * 2. {@link ITransactionContext#beforeOperation(long)}
+ * 3. {@link ITransactionContext#notifyUpdateCommitted(long)}
+ * 4. {@link ITransactionContext#notifyEntityCommitted}
+ * 5. {@link ITransactionContext#afterOperation(long)}
+ * 6. {@link ITransactionContext#complete()}
+ */
 public interface ITransactionContext {
 
-    public void registerIndexAndCallback(long resourceId, ILSMIndex index, AbstractOperationCallback callback,
-            boolean isPrimaryIndex);
+    /**
+     * Registers {@link ILSMIndex} in the transaction. Registering an index
+     * must be done before any operation is performed on the index by this
+     * transaction.
+     *
+     * @param resourceId
+     * @param index
+     * @param callback
+     * @param primaryIndex
+     */
+    void register(long resourceId, ILSMIndex index, IModificationOperationCallback callback, boolean primaryIndex);
 
-    public TxnId getTxnId();
+    /**
+     * Gets the unique transaction id.
+     *
+     * @return the unique transaction id
+     */
+    TxnId getTxnId();
 
-    public void setTimeout(boolean isTimeout);
+    /**
+     * Sets a flag indicating that the transaction timed out.
+     *
+     * @param isTimeout
+     */
+    void setTimeout(boolean isTimeout);
 
-    public boolean isTimeout();
+    /**
+     * Tests if the transaction was timed out.
+     *
+     * @return true if this transaction timed out. Otherwise false.
+     */
+    boolean isTimeout();
 
-    public void setTxnState(int txnState);
+    /**
+     * Sets the state if this transaction.
+     *
+     * @param txnState
+     */
+    void setTxnState(int txnState);
 
-    public int getTxnState();
+    /**
+     * Gets the current state of this transaction.
+     *
+     * @return the current state of this transaction
+     */
+    int getTxnState();
 
-    public long getFirstLSN();
+    /**
+     * Gets the first log sequence number of this transaction.
+     *
+     * @return the first log sequence number
+     */
+    long getFirstLSN();
 
-    public long getLastLSN();
+    /**
+     * Gets the last log sequence number of this transactions.
+     *
+     * @return the last log sequence number
+     */
+    long getLastLSN();
 
-    public void setLastLSN(long LSN);
+    /**
+     * Sets the last log sequence number of this transactions.
+     *
+     * @param newValue
+     */
+    void setLastLSN(long newValue);
 
-    public boolean isWriteTxn();
+    /**
+     * Tests if this is a write transaction.
+     *
+     * @return true if this is a write transaction, otherwise false.
+     */
+    boolean isWriteTxn();
 
-    public void setWriteTxn(boolean isWriterTxn);
+    /**
+     * Sets a flag indication that this is a write transaction.
+     *
+     * @param isWriterTxn
+     */
+    void setWriteTxn(boolean isWriterTxn);
 
-    public String prettyPrint();
+    /**
+     * Called before an operation is performed on index
+     * with resource id {@code resourceId}.
+     *
+     * @param resourceId
+     */
+    void beforeOperation(long resourceId);
 
-    public void setMetadataTransaction(boolean isMetadataTxn);
+    /**
+     * Called to notify the transaction that an update log belonging
+     * to this transaction on index with {@code resourceId} has been
+     * flushed to disk.
+     *
+     * @param resourceId
+     */
+    void notifyUpdateCommitted(long resourceId);
 
-    public boolean isMetadataTransaction();
+    /**
+     * Called to notify the transaction that an entity commit
+     * log belonging to this transaction has been flushed to
+     * disk.
+     */
+    void notifyEntityCommitted();
 
-    public void notifyOptracker(boolean isJobLevelCommit);
-
-    public void incrementNumActiveOperations();
-
-    public void decrementNumActiveOperations();
+    /**
+     * Called after an operation is performed on index
+     * with resource id {@code resourceId}.
+     *
+     * @param resourceId
+     */
+    void afterOperation(long resourceId);
 
     /**
      * Called when no further operations will be performed by the transaction
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ITransactionManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ITransactionManager.java
index 77c6a9f..9603ce3 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ITransactionManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ITransactionManager.java
@@ -24,105 +24,77 @@
  * Provides APIs for managing life cycle of a transaction, that is beginning a
  * transaction and aborting/committing the transaction.
  */
-
 public interface ITransactionManager {
 
     /**
      * A transaction may be in any of the following states ACTIVE: The
-     * transaction is ongoing and has not yet committed/aborted. COMMITTD: The
+     * transaction is ongoing and has not yet committed/aborted. COMMITTED: The
      * transaction has committed. ABORTED: The transaction has aborted.
      * TIMED_OUT: The transaction has timed out waiting to acquire a lock.
      */
-    public static final int ACTIVE = 0;
-    public static final int COMMITTED = 1;
-    public static final int ABORTED = 2;
-    public static final int TIMED_OUT = 3;
+    int ACTIVE = 0;
+    int COMMITTED = 1;
+    int ABORTED = 2;
+    int TIMED_OUT = 3;
+
+    enum AtomicityLevel {
+        /**
+         * all records are committed or nothing
+         */
+        ATOMIC,
+        /**
+         * any record with entity commit log
+         */
+        ENTITY_LEVEL
+    }
+
+    enum TransactionMode {
+        /**
+         * Transaction performs only read operations
+         */
+        READ,
+        /**
+         * Transaction may perform read and write operations
+         */
+        READ_WRITE
+    }
 
     /**
      * Begins a transaction identified by a transaction id and returns the
      * associated transaction context.
      *
      * @param txnId
-     *            a unique value for the transaction id.
-     * @return the transaction context associated with the initiated transaction
-     * @see ITransactionContext
+     * @param options
+     * @return The transaction context
      * @throws ACIDException
      */
-    public ITransactionContext beginTransaction(TxnId txnId) throws ACIDException;
+    ITransactionContext beginTransaction(TxnId txnId, TransactionOptions options) throws ACIDException;
 
     /**
      * Returns the transaction context of an active transaction given the
      * transaction id.
      *
      * @param txnId
-     *            a unique value for the transaction id.
-     * @param createIfNotExist
-     *            TODO
-     * @return
+     * @return The transaction context
      * @throws ACIDException
      */
-    public ITransactionContext getTransactionContext(TxnId txnId, boolean createIfNotExist) throws ACIDException;
+    ITransactionContext getTransactionContext(TxnId txnId) throws ACIDException;
 
     /**
-     * Commits a transaction.
+     * Commit a transactions
      *
-     * @param txnContext
-     *            the transaction context associated with the transaction
-     * @param datasetId
-     *            TODO
-     * @param pkHash
-     *            TODO
+     * @param txnId
      * @throws ACIDException
-     * @see ITransactionContextimport org.apache.hyracks.api.job.TxnId;
-     * @see ACIDException
      */
-    public void commitTransaction(ITransactionContext txnContext, DatasetId datasetId, int pkHash)
-            throws ACIDException;
+    void commitTransaction(TxnId txnId) throws ACIDException;
 
     /**
      * Aborts a transaction.
      *
-     * @param txnContext
-     *            the transaction context associated with the transaction
-     * @param datasetId
-     *            TODO
-     * @param pkHash
-     *            TODO
-     * @throws ACIDException
-     * @see ITransactionContext
-     * @see ACIDException
-     */
-    public void abortTransaction(ITransactionContext txnContext, DatasetId datasetId, int pkHash)
-            throws ACIDException;
-
-    /**
-     * Indicates end of all activity for a transaction. In other words, all
-     * participating threads in the transaction have completed the intended
-     * task.
-     *
-     * @param txnContext
-     *            the transaction context associated with the transaction
-     * @param datasetId
-     *            TODO
-     * @param pkHash
-     *            TODO
-     * @param success
-     *            indicates the success or failure. The transaction is committed
-     *            or aborted accordingly.
+     * @param txnId
      * @throws ACIDException
      */
-    public void completedTransaction(ITransactionContext txnContext, DatasetId datasetId, int pkHash,
-            boolean success) throws ACIDException;
-
-    /**
-     * Returns the Transaction Provider for the transaction eco-system. A
-     * transaction eco-system consists of a Log Manager, a Recovery Manager, a
-     * Transaction Manager and a Lock Manager.
-     *
-     * @see ITransactionSubsystem
-     * @return TransactionProvider
-     */
-    public ITransactionSubsystem getTransactionSubsystem();
+    void abortTransaction(TxnId txnId) throws ACIDException;
 
     /**
      * @return The current max txn id.
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/TransactionOptions.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/TransactionOptions.java
new file mode 100644
index 0000000..48dc452
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/TransactionOptions.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.asterix.common.transactions;
+
+import static org.apache.asterix.common.transactions.ITransactionManager.AtomicityLevel;
+
+public class TransactionOptions {
+
+    private final AtomicityLevel atomicityLevel;
+
+    // TODO add TransactionMode(READ/WRITE) to options
+    public TransactionOptions(AtomicityLevel atomicityLevel) {
+        this.atomicityLevel = atomicityLevel;
+    }
+
+    public AtomicityLevel getAtomicityLevel() {
+        return atomicityLevel;
+    }
+}
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/MetadataNode.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/MetadataNode.java
index 87a5272..368fc2a 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/MetadataNode.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/MetadataNode.java
@@ -19,13 +19,14 @@
 
 package org.apache.asterix.metadata;
 
+import static org.apache.asterix.common.transactions.ITransactionManager.AtomicityLevel;
+
 import java.rmi.RemoteException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import org.apache.asterix.common.api.IDatasetLifecycleManager;
@@ -36,12 +37,12 @@
 import org.apache.asterix.common.exceptions.ACIDException;
 import org.apache.asterix.common.functions.FunctionSignature;
 import org.apache.asterix.common.metadata.MetadataIndexImmutableProperties;
-import org.apache.asterix.common.transactions.AbstractOperationCallback;
 import org.apache.asterix.common.transactions.DatasetId;
 import org.apache.asterix.common.transactions.IRecoveryManager.ResourceType;
 import org.apache.asterix.common.transactions.ITransactionContext;
 import org.apache.asterix.common.transactions.ITransactionSubsystem;
 import org.apache.asterix.common.transactions.ImmutableDatasetId;
+import org.apache.asterix.common.transactions.TransactionOptions;
 import org.apache.asterix.common.transactions.TxnId;
 import org.apache.asterix.external.indexing.ExternalFile;
 import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider;
@@ -99,7 +100,6 @@
 import org.apache.asterix.transaction.management.opcallbacks.SecondaryIndexModificationOperationCallback;
 import org.apache.asterix.transaction.management.opcallbacks.UpsertOperationCallback;
 import org.apache.asterix.transaction.management.service.transaction.DatasetIdFactory;
-import org.apache.asterix.transaction.management.service.transaction.TransactionContext;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
 import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
@@ -117,7 +117,6 @@
 import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
-import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
 import org.apache.hyracks.storage.common.IIndex;
 import org.apache.hyracks.storage.common.IIndexAccessParameters;
@@ -128,7 +127,6 @@
 
 public class MetadataNode implements IMetadataNode {
     private static final long serialVersionUID = 1L;
-    private static final Logger LOGGER = Logger.getLogger(MetadataNode.class.getName());
     private static final DatasetId METADATA_DATASET_ID =
             new ImmutableDatasetId(MetadataPrimaryIndexes.PROPERTIES_METADATA.getDatasetId());
 
@@ -165,37 +163,29 @@
 
     @Override
     public void beginTransaction(TxnId transactionId) throws ACIDException, RemoteException {
-        ITransactionContext txnCtx = transactionSubsystem.getTransactionManager().beginTransaction(transactionId);
-        txnCtx.setMetadataTransaction(true);
+        TransactionOptions options = new TransactionOptions(AtomicityLevel.ATOMIC);
+        transactionSubsystem.getTransactionManager().beginTransaction(transactionId, options);
     }
 
     @Override
     public void commitTransaction(TxnId txnId) throws RemoteException, ACIDException {
-        ITransactionContext txnCtx = transactionSubsystem.getTransactionManager().getTransactionContext(txnId, false);
-        transactionSubsystem.getTransactionManager().commitTransaction(txnCtx, DatasetId.NULL, -1);
+        transactionSubsystem.getTransactionManager().commitTransaction(txnId);
     }
 
     @Override
     public void abortTransaction(TxnId txnId) throws RemoteException, ACIDException {
-        try {
-            ITransactionContext txnCtx =
-                    transactionSubsystem.getTransactionManager().getTransactionContext(txnId, false);
-            transactionSubsystem.getTransactionManager().abortTransaction(txnCtx, DatasetId.NULL, -1);
-        } catch (ACIDException e) {
-            LOGGER.log(Level.WARNING, "Exception aborting transaction", e);
-            throw e;
-        }
+        transactionSubsystem.getTransactionManager().abortTransaction(txnId);
     }
 
     @Override
     public void lock(TxnId txnId, byte lockMode) throws ACIDException, RemoteException {
-        ITransactionContext txnCtx = transactionSubsystem.getTransactionManager().getTransactionContext(txnId, false);
+        ITransactionContext txnCtx = transactionSubsystem.getTransactionManager().getTransactionContext(txnId);
         transactionSubsystem.getLockManager().lock(METADATA_DATASET_ID, -1, lockMode, txnCtx);
     }
 
     @Override
     public void unlock(TxnId txnId, byte lockMode) throws ACIDException, RemoteException {
-        ITransactionContext txnCtx = transactionSubsystem.getTransactionManager().getTransactionContext(txnId, false);
+        ITransactionContext txnCtx = transactionSubsystem.getTransactionManager().getTransactionContext(txnId);
         transactionSubsystem.getLockManager().unlock(METADATA_DATASET_ID, -1, lockMode, txnCtx);
     }
 
@@ -472,96 +462,66 @@
 
     private void insertTupleIntoIndex(TxnId txnId, IMetadataIndex metadataIndex, ITupleReference tuple)
             throws ACIDException, HyracksDataException {
-        long resourceID = metadataIndex.getResourceId();
-        String resourceName = metadataIndex.getFile().getRelativePath();
-        ILSMIndex lsmIndex = (ILSMIndex) datasetLifecycleManager.get(resourceName);
-        try {
-            datasetLifecycleManager.open(resourceName);
-
-            // prepare a Callback for logging
-            IModificationOperationCallback modCallback =
-                    createIndexModificationCallback(txnId, resourceID, metadataIndex, lsmIndex, Operation.INSERT);
-
-            IIndexAccessParameters iap = new IndexAccessParameters(modCallback, NoOpOperationCallback.INSTANCE);
-            ILSMIndexAccessor indexAccessor = lsmIndex.createAccessor(iap);
-
-            ITransactionContext txnCtx =
-                    transactionSubsystem.getTransactionManager().getTransactionContext(txnId, false);
-            txnCtx.setWriteTxn(true);
-            txnCtx.registerIndexAndCallback(resourceID, lsmIndex, (AbstractOperationCallback) modCallback,
-                    metadataIndex.isPrimaryIndex());
-
-            LSMIndexUtil.checkAndSetFirstLSN((AbstractLSMIndex) lsmIndex, transactionSubsystem.getLogManager());
-
-            // TODO: fix exceptions once new BTree exception model is in hyracks.
-            indexAccessor.forceInsert(tuple);
-            // Manually complete the operation after the insert. This is to decrement the
-            // resource counters within the
-            // index that determine how many tuples are still 'in-flight' within the index.
-            // Normally the log flusher
-            // does this. The only exception is the index registered as the "primary" which
-            // we will let be decremented
-            // by the job commit log event
-            if (!((TransactionContext) txnCtx).getPrimaryIndexOpTracker().equals(lsmIndex.getOperationTracker())) {
-                lsmIndex.getOperationTracker().completeOperation(lsmIndex, LSMOperationType.FORCE_MODIFICATION, null,
-                        modCallback);
-            }
-        } finally {
-            datasetLifecycleManager.close(resourceName);
-        }
+        modifyMetadataIndex(Operation.INSERT, txnId, metadataIndex, tuple);
     }
 
     private void upsertTupleIntoIndex(TxnId txnId, IMetadataIndex metadataIndex, ITupleReference tuple)
             throws ACIDException, HyracksDataException {
-        long resourceId = metadataIndex.getResourceId();
+        modifyMetadataIndex(Operation.UPSERT, txnId, metadataIndex, tuple);
+    }
+
+    private void modifyMetadataIndex(Operation op, TxnId txnId, IMetadataIndex metadataIndex, ITupleReference tuple)
+            throws ACIDException, HyracksDataException {
         String resourceName = metadataIndex.getFile().getRelativePath();
         ILSMIndex lsmIndex = (ILSMIndex) datasetLifecycleManager.get(resourceName);
         datasetLifecycleManager.open(resourceName);
         try {
-            // prepare a Callback for logging
-            ITransactionContext txnCtx =
-                    transactionSubsystem.getTransactionManager().getTransactionContext(txnId, false);
-            IModificationOperationCallback modCallback =
-                    new UpsertOperationCallback(metadataIndex.getDatasetId(), metadataIndex.getPrimaryKeyIndexes(),
-                            txnCtx, transactionSubsystem.getLockManager(), transactionSubsystem, resourceId,
-                            metadataStoragePartition, ResourceType.LSM_BTREE, Operation.UPSERT);
+            ITransactionContext txnCtx = transactionSubsystem.getTransactionManager().getTransactionContext(txnId);
+            IModificationOperationCallback modCallback = createIndexModificationCallback(op, txnCtx, metadataIndex);
             IIndexAccessParameters iap = new IndexAccessParameters(modCallback, NoOpOperationCallback.INSTANCE);
             ILSMIndexAccessor indexAccessor = lsmIndex.createAccessor(iap);
             txnCtx.setWriteTxn(true);
-            txnCtx.registerIndexAndCallback(resourceId, lsmIndex, (AbstractOperationCallback) modCallback,
-                    metadataIndex.isPrimaryIndex());
+            txnCtx.register(metadataIndex.getResourceId(), lsmIndex, modCallback, metadataIndex.isPrimaryIndex());
             LSMIndexUtil.checkAndSetFirstLSN((AbstractLSMIndex) lsmIndex, transactionSubsystem.getLogManager());
-            indexAccessor.forceUpsert(tuple);
-            // Manually complete the operation after the insert. This is to decrement the
-            // resource counters within the
-            // index that determine how many tuples are still 'in-flight' within the index.
-            // Normally the log flusher
-            // does this. The only exception is the index registered as the "primary" which
-            // we will let be decremented
-            // by the job commit log event
-            if (!((TransactionContext) txnCtx).getPrimaryIndexOpTracker().equals(lsmIndex.getOperationTracker())) {
-                lsmIndex.getOperationTracker().completeOperation(lsmIndex, LSMOperationType.FORCE_MODIFICATION, null,
-                        modCallback);
+            switch (op) {
+                case INSERT:
+                    indexAccessor.insert(tuple);
+                    break;
+                case DELETE:
+                    indexAccessor.delete(tuple);
+                    break;
+                case UPSERT:
+                    indexAccessor.upsert(tuple);
+                    break;
+                default:
+                    throw new IllegalStateException("Unknown operation type: " + op);
             }
         } finally {
             datasetLifecycleManager.close(resourceName);
         }
     }
 
-    private IModificationOperationCallback createIndexModificationCallback(TxnId txnId, long resourceId,
-            IMetadataIndex metadataIndex, ILSMIndex lsmIndex, Operation indexOp) throws ACIDException {
-        ITransactionContext txnCtx = transactionSubsystem.getTransactionManager().getTransactionContext(txnId, false);
-
-        // Regardless of the index type (primary or secondary index), secondary index
-        // modification callback is given
-        // This is still correct since metadata index operation doesn't require any lock
-        // from ConcurrentLockMgr and
-        // The difference between primaryIndexModCallback and secondaryIndexModCallback
-        // is that primary index requires
-        // locks and secondary index doesn't.
-        return new SecondaryIndexModificationOperationCallback(metadataIndex.getDatasetId(),
-                metadataIndex.getPrimaryKeyIndexes(), txnCtx, transactionSubsystem.getLockManager(),
-                transactionSubsystem, resourceId, metadataStoragePartition, ResourceType.LSM_BTREE, indexOp);
+    private IModificationOperationCallback createIndexModificationCallback(Operation indexOp,
+            ITransactionContext txnCtx, IMetadataIndex metadataIndex) {
+        switch (indexOp) {
+            case INSERT:
+            case DELETE:
+                /*
+                 * Regardless of the index type (primary or secondary index), secondary index modification
+                 * callback is given. This is still correct since metadata index operation doesn't require
+                 * any lock from ConcurrentLockMgr.
+                 */
+                return new SecondaryIndexModificationOperationCallback(metadataIndex.getDatasetId(),
+                        metadataIndex.getPrimaryKeyIndexes(), txnCtx, transactionSubsystem.getLockManager(),
+                        transactionSubsystem, metadataIndex.getResourceId(), metadataStoragePartition,
+                        ResourceType.LSM_BTREE, indexOp);
+            case UPSERT:
+                return new UpsertOperationCallback(metadataIndex.getDatasetId(), metadataIndex.getPrimaryKeyIndexes(),
+                        txnCtx, transactionSubsystem.getLockManager(), transactionSubsystem,
+                        metadataIndex.getResourceId(), metadataStoragePartition, ResourceType.LSM_BTREE, indexOp);
+            default:
+                throw new IllegalStateException("Unknown operation type: " + indexOp);
+        }
     }
 
     @Override
@@ -822,40 +782,7 @@
 
     private void deleteTupleFromIndex(TxnId txnId, IMetadataIndex metadataIndex, ITupleReference tuple)
             throws ACIDException, HyracksDataException {
-        long resourceID = metadataIndex.getResourceId();
-        String resourceName = metadataIndex.getFile().getRelativePath();
-        ILSMIndex lsmIndex = (ILSMIndex) datasetLifecycleManager.get(resourceName);
-        try {
-            datasetLifecycleManager.open(resourceName);
-            // prepare a Callback for logging
-            IModificationOperationCallback modCallback =
-                    createIndexModificationCallback(txnId, resourceID, metadataIndex, lsmIndex, Operation.DELETE);
-            IIndexAccessParameters iap = new IndexAccessParameters(modCallback, NoOpOperationCallback.INSTANCE);
-            ILSMIndexAccessor indexAccessor = lsmIndex.createAccessor(iap);
-
-            ITransactionContext txnCtx =
-                    transactionSubsystem.getTransactionManager().getTransactionContext(txnId, false);
-            txnCtx.setWriteTxn(true);
-            txnCtx.registerIndexAndCallback(resourceID, lsmIndex, (AbstractOperationCallback) modCallback,
-                    metadataIndex.isPrimaryIndex());
-
-            LSMIndexUtil.checkAndSetFirstLSN((AbstractLSMIndex) lsmIndex, transactionSubsystem.getLogManager());
-
-            indexAccessor.forceDelete(tuple);
-            // Manually complete the operation after the insert. This is to decrement the
-            // resource counters within the
-            // index that determine how many tuples are still 'in-flight' within the index.
-            // Normally the log flusher
-            // does this. The only exception is the index registered as the "primary" which
-            // we will let be decremented
-            // by the job commit log event
-            if (!((TransactionContext) txnCtx).getPrimaryIndexOpTracker().equals(lsmIndex.getOperationTracker())) {
-                lsmIndex.getOperationTracker().completeOperation(lsmIndex, LSMOperationType.FORCE_MODIFICATION, null,
-                        modCallback);
-            }
-        } finally {
-            datasetLifecycleManager.close(resourceName);
-        }
+        modifyMetadataIndex(Operation.DELETE, txnId, metadataIndex, tuple);
     }
 
     @Override
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/job/listener/JobEventListenerFactory.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/job/listener/JobEventListenerFactory.java
index 23ad1e1..d3c3fe7 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/job/listener/JobEventListenerFactory.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/job/listener/JobEventListenerFactory.java
@@ -18,12 +18,14 @@
  */
 package org.apache.asterix.runtime.job.listener;
 
+import static org.apache.asterix.common.transactions.ITransactionManager.AtomicityLevel;
+
 import org.apache.asterix.common.api.IJobEventListenerFactory;
 import org.apache.asterix.common.api.INcApplicationContext;
 import org.apache.asterix.common.exceptions.ACIDException;
-import org.apache.asterix.common.transactions.DatasetId;
 import org.apache.asterix.common.transactions.ITransactionContext;
 import org.apache.asterix.common.transactions.ITransactionManager;
+import org.apache.asterix.common.transactions.TransactionOptions;
 import org.apache.asterix.common.transactions.TxnId;
 import org.apache.hyracks.api.context.IHyracksJobletContext;
 import org.apache.hyracks.api.job.IJobletEventListener;
@@ -79,10 +81,13 @@
                 try {
                     ITransactionManager txnManager = ((INcApplicationContext) jobletContext.getServiceContext()
                             .getApplicationContext()).getTransactionSubsystem().getTransactionManager();
-                    ITransactionContext txnContext = txnManager.getTransactionContext(txnId, false);
+                    ITransactionContext txnContext = txnManager.getTransactionContext(txnId);
                     txnContext.setWriteTxn(transactionalWrite);
-                    txnManager.completedTransaction(txnContext, DatasetId.NULL, -1,
-                            !(jobStatus == JobStatus.FAILURE));
+                    if (jobStatus != JobStatus.FAILURE) {
+                        txnManager.commitTransaction(txnId);
+                    } else {
+                        txnManager.abortTransaction(txnId);
+                    }
                 } catch (ACIDException e) {
                     throw new Error(e);
                 }
@@ -91,8 +96,9 @@
             @Override
             public void jobletStart() {
                 try {
+                    TransactionOptions options = new TransactionOptions(AtomicityLevel.ENTITY_LEVEL);
                     ((INcApplicationContext) jobletContext.getServiceContext().getApplicationContext())
-                            .getTransactionSubsystem().getTransactionManager().getTransactionContext(txnId, true);
+                            .getTransactionSubsystem().getTransactionManager().beginTransaction(txnId, options);
                 } catch (ACIDException e) {
                     throw new Error(e);
                 }
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/job/listener/MultiTransactionJobletEventListenerFactory.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/job/listener/MultiTransactionJobletEventListenerFactory.java
index 23c86f3..bfe1925 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/job/listener/MultiTransactionJobletEventListenerFactory.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/job/listener/MultiTransactionJobletEventListenerFactory.java
@@ -23,9 +23,9 @@
 import org.apache.asterix.common.api.IJobEventListenerFactory;
 import org.apache.asterix.common.api.INcApplicationContext;
 import org.apache.asterix.common.exceptions.ACIDException;
-import org.apache.asterix.common.transactions.DatasetId;
 import org.apache.asterix.common.transactions.ITransactionContext;
 import org.apache.asterix.common.transactions.ITransactionManager;
+import org.apache.asterix.common.transactions.TransactionOptions;
 import org.apache.asterix.common.transactions.TxnId;
 import org.apache.hyracks.api.context.IHyracksJobletContext;
 import org.apache.hyracks.api.job.IJobletEventListener;
@@ -75,10 +75,13 @@
                             ((INcApplicationContext) jobletContext.getServiceContext().getApplicationContext())
                                     .getTransactionSubsystem().getTransactionManager();
                     for (TxnId txnId : txnIds) {
-                        ITransactionContext txnContext = txnManager.getTransactionContext(txnId, false);
+                        ITransactionContext txnContext = txnManager.getTransactionContext(txnId);
                         txnContext.setWriteTxn(transactionalWrite);
-                        txnManager.completedTransaction(txnContext, DatasetId.NULL, -1,
-                                !(jobStatus == JobStatus.FAILURE));
+                        if (jobStatus != JobStatus.FAILURE) {
+                            txnManager.commitTransaction(txnId);
+                        } else {
+                            txnManager.abortTransaction(txnId);
+                        }
                     }
                 } catch (ACIDException e) {
                     throw new Error(e);
@@ -88,9 +91,11 @@
             @Override
             public void jobletStart() {
                 try {
+                    TransactionOptions options =
+                            new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL);
                     for (TxnId txnId : txnIds) {
                         ((INcApplicationContext) jobletContext.getServiceContext().getApplicationContext())
-                                .getTransactionSubsystem().getTransactionManager().getTransactionContext(txnId, true);
+                                .getTransactionSubsystem().getTransactionManager().beginTransaction(txnId, options);
                     }
                 } catch (ACIDException e) {
                     throw new Error(e);
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/std/FlushDatasetOperatorDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/std/FlushDatasetOperatorDescriptor.java
index 6f7287b..d61e9a0 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/std/FlushDatasetOperatorDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/std/FlushDatasetOperatorDescriptor.java
@@ -78,7 +78,7 @@
                     ILockManager lockManager = appCtx.getTransactionSubsystem().getLockManager();
                     ITransactionManager txnManager = appCtx.getTransactionSubsystem().getTransactionManager();
                     // get the local transaction
-                    ITransactionContext txnCtx = txnManager.getTransactionContext(txnId, false);
+                    ITransactionContext txnCtx = txnManager.getTransactionContext(txnId);
                     // lock the dataset granule
                     lockManager.lock(datasetId, -1, LockMode.S, txnCtx);
                     // flush the dataset synchronously
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/AbstractIndexModificationOperationCallback.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/AbstractIndexModificationOperationCallback.java
index 3a2f195..3da9e83 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/AbstractIndexModificationOperationCallback.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/AbstractIndexModificationOperationCallback.java
@@ -68,7 +68,6 @@
         }
     }
 
-    protected final long resourceId;
     protected final byte resourceType;
     protected final Operation indexOp;
     protected final ITransactionSubsystem txnSubsystem;
@@ -77,8 +76,7 @@
     protected AbstractIndexModificationOperationCallback(DatasetId datasetId, int[] primaryKeyFields,
             ITransactionContext txnCtx, ILockManager lockManager, ITransactionSubsystem txnSubsystem, long resourceId,
             int resourcePartition, byte resourceType, Operation indexOp) {
-        super(datasetId, primaryKeyFields, txnCtx, lockManager);
-        this.resourceId = resourceId;
+        super(datasetId, resourceId, primaryKeyFields, txnCtx, lockManager);
         this.resourceType = resourceType;
         this.indexOp = indexOp;
         this.txnSubsystem = txnSubsystem;
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/LockThenSearchOperationCallback.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/LockThenSearchOperationCallback.java
index c97fb1b..fe17b39 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/LockThenSearchOperationCallback.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/LockThenSearchOperationCallback.java
@@ -45,10 +45,10 @@
     private final ILogRecord logRecord;
     private int pkHash;
 
-    public LockThenSearchOperationCallback(DatasetId datasetId, int[] entityIdFields,
+    public LockThenSearchOperationCallback(DatasetId datasetId, long resourceId, int[] entityIdFields,
             ITransactionSubsystem txnSubsystem, ITransactionContext txnCtx,
             IOperatorNodePushable operatorNodePushable) {
-        super(datasetId, entityIdFields, txnCtx, txnSubsystem.getLockManager());
+        super(datasetId, resourceId, entityIdFields, txnCtx, txnSubsystem.getLockManager());
         this.operatorNodePushable = (LSMIndexInsertUpdateDeleteOperatorNodePushable) operatorNodePushable;
         this.logManager = txnSubsystem.getLogManager();
         this.logRecord = new LogRecord();
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/LockThenSearchOperationCallbackFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/LockThenSearchOperationCallbackFactory.java
index 3f3dbd9..9f96263 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/LockThenSearchOperationCallbackFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/LockThenSearchOperationCallbackFactory.java
@@ -49,9 +49,9 @@
         try {
             IJobletEventListenerFactory fact = ctx.getJobletContext().getJobletEventListenerFactory();
             ITransactionContext txnCtx = txnSubsystem.getTransactionManager()
-                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId), false);
-            return new LockThenSearchOperationCallback(new DatasetId(datasetId), primaryKeyFields, txnSubsystem, txnCtx,
-                    operatorNodePushable);
+                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId));
+            return new LockThenSearchOperationCallback(new DatasetId(datasetId), resourceId, primaryKeyFields,
+                    txnSubsystem, txnCtx, operatorNodePushable);
         } catch (ACIDException e) {
             throw new HyracksDataException(e);
         }
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexInstantSearchOperationCallback.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexInstantSearchOperationCallback.java
index b13a08e..ec776a5 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexInstantSearchOperationCallback.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexInstantSearchOperationCallback.java
@@ -35,9 +35,9 @@
 public class PrimaryIndexInstantSearchOperationCallback extends AbstractOperationCallback
         implements ISearchOperationCallback {
 
-    public PrimaryIndexInstantSearchOperationCallback(DatasetId datasetId, int[] entityIdFields,
+    public PrimaryIndexInstantSearchOperationCallback(DatasetId datasetId, long resourceId, int[] entityIdFields,
             ILockManager lockManager, ITransactionContext txnCtx) {
-        super(datasetId, entityIdFields, txnCtx, lockManager);
+        super(datasetId, resourceId, entityIdFields, txnCtx, lockManager);
     }
 
     @Override
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexInstantSearchOperationCallbackFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexInstantSearchOperationCallbackFactory.java
index 93108f9..f9c8e3c 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexInstantSearchOperationCallbackFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexInstantSearchOperationCallbackFactory.java
@@ -34,8 +34,8 @@
 import org.apache.hyracks.storage.am.common.api.ISearchOperationCallbackFactory;
 import org.apache.hyracks.storage.common.ISearchOperationCallback;
 
-public class PrimaryIndexInstantSearchOperationCallbackFactory extends AbstractOperationCallbackFactory implements
-        ISearchOperationCallbackFactory {
+public class PrimaryIndexInstantSearchOperationCallbackFactory extends AbstractOperationCallbackFactory
+        implements ISearchOperationCallbackFactory {
 
     private static final long serialVersionUID = 1L;
 
@@ -45,15 +45,15 @@
     }
 
     @Override
-    public ISearchOperationCallback createSearchOperationCallback(long resourceId, IHyracksTaskContext ctx, IOperatorNodePushable operatorNodePushable)
-            throws HyracksDataException {
+    public ISearchOperationCallback createSearchOperationCallback(long resourceId, IHyracksTaskContext ctx,
+            IOperatorNodePushable operatorNodePushable) throws HyracksDataException {
         ITransactionSubsystem txnSubsystem = txnSubsystemProvider.getTransactionSubsystem(ctx);
         try {
             IJobletEventListenerFactory fact = ctx.getJobletContext().getJobletEventListenerFactory();
             ITransactionContext txnCtx = txnSubsystem.getTransactionManager()
-                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId), false);
-            return new PrimaryIndexInstantSearchOperationCallback(new DatasetId(datasetId), primaryKeyFields,
-                    txnSubsystem.getLockManager(), txnCtx);
+                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId));
+            return new PrimaryIndexInstantSearchOperationCallback(new DatasetId(datasetId), resourceId,
+                    primaryKeyFields, txnSubsystem.getLockManager(), txnCtx);
         } catch (ACIDException e) {
             throw new HyracksDataException(e);
         }
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexModificationOperationCallbackFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexModificationOperationCallbackFactory.java
index fb01952..8f5e386 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexModificationOperationCallbackFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexModificationOperationCallbackFactory.java
@@ -23,7 +23,6 @@
 import org.apache.asterix.common.context.ITransactionSubsystemProvider;
 import org.apache.asterix.common.dataflow.DatasetLocalResource;
 import org.apache.asterix.common.exceptions.ACIDException;
-import org.apache.asterix.common.transactions.AbstractOperationCallback;
 import org.apache.asterix.common.transactions.AbstractOperationCallbackFactory;
 import org.apache.asterix.common.transactions.DatasetId;
 import org.apache.asterix.common.transactions.ITransactionContext;
@@ -70,12 +69,12 @@
         try {
             IJobletEventListenerFactory fact = ctx.getJobletContext().getJobletEventListenerFactory();
             ITransactionContext txnCtx = txnSubsystem.getTransactionManager()
-                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId), false);
+                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId));
             DatasetLocalResource aResource = (DatasetLocalResource) resource.getResource();
             IModificationOperationCallback modCallback = new PrimaryIndexModificationOperationCallback(
                     new DatasetId(datasetId), primaryKeyFields, txnCtx, txnSubsystem.getLockManager(), txnSubsystem,
                     resource.getId(), aResource.getPartition(), resourceType, indexOp, operatorNodePushable);
-            txnCtx.registerIndexAndCallback(resource.getId(), index, (AbstractOperationCallback) modCallback, true);
+            txnCtx.register(resource.getId(), index, modCallback, true);
             return modCallback;
         } catch (ACIDException e) {
             throw HyracksDataException.create(e);
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexSearchOperationCallback.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexSearchOperationCallback.java
index a9075d0..961f799 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexSearchOperationCallback.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexSearchOperationCallback.java
@@ -34,9 +34,9 @@
  */
 public class PrimaryIndexSearchOperationCallback extends AbstractOperationCallback implements ISearchOperationCallback {
 
-    public PrimaryIndexSearchOperationCallback(DatasetId datasetId, int[] entityIdFields, ILockManager lockManager,
-            ITransactionContext txnCtx) {
-        super(datasetId, entityIdFields, txnCtx, lockManager);
+    public PrimaryIndexSearchOperationCallback(DatasetId datasetId, long resourceId, int[] entityIdFields,
+            ILockManager lockManager, ITransactionContext txnCtx) {
+        super(datasetId, resourceId, entityIdFields, txnCtx, lockManager);
     }
 
     @Override
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexSearchOperationCallbackFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexSearchOperationCallbackFactory.java
index 076b0d9..64cbbc9 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexSearchOperationCallbackFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexSearchOperationCallbackFactory.java
@@ -51,8 +51,8 @@
         try {
             IJobletEventListenerFactory fact = ctx.getJobletContext().getJobletEventListenerFactory();
             ITransactionContext txnCtx = txnSubsystem.getTransactionManager()
-                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId), false);
-            return new PrimaryIndexSearchOperationCallback(new DatasetId(datasetId), primaryKeyFields,
+                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId));
+            return new PrimaryIndexSearchOperationCallback(new DatasetId(datasetId), resourceId, primaryKeyFields,
                     txnSubsystem.getLockManager(), txnCtx);
         } catch (ACIDException e) {
             throw new HyracksDataException(e);
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexModificationOperationCallbackFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexModificationOperationCallbackFactory.java
index 5882046..3fc42c9 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexModificationOperationCallbackFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexModificationOperationCallbackFactory.java
@@ -23,7 +23,6 @@
 import org.apache.asterix.common.context.ITransactionSubsystemProvider;
 import org.apache.asterix.common.dataflow.DatasetLocalResource;
 import org.apache.asterix.common.exceptions.ACIDException;
-import org.apache.asterix.common.transactions.AbstractOperationCallback;
 import org.apache.asterix.common.transactions.AbstractOperationCallbackFactory;
 import org.apache.asterix.common.transactions.DatasetId;
 import org.apache.asterix.common.transactions.ITransactionContext;
@@ -66,12 +65,12 @@
         try {
             IJobletEventListenerFactory fact = ctx.getJobletContext().getJobletEventListenerFactory();
             ITransactionContext txnCtx = txnSubsystem.getTransactionManager()
-                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId), false);
+                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId));
             DatasetLocalResource aResource = (DatasetLocalResource) resource.getResource();
             IModificationOperationCallback modCallback = new SecondaryIndexModificationOperationCallback(
                     new DatasetId(datasetId), primaryKeyFields, txnCtx, txnSubsystem.getLockManager(), txnSubsystem,
                     resource.getId(), aResource.getPartition(), resourceType, indexOp);
-            txnCtx.registerIndexAndCallback(resource.getId(), index, (AbstractOperationCallback) modCallback, false);
+            txnCtx.register(resource.getId(), index, modCallback, false);
             return modCallback;
         } catch (ACIDException e) {
             throw HyracksDataException.create(e);
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexSearchOperationCallback.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexSearchOperationCallback.java
index 108a77e..1b87376 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexSearchOperationCallback.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexSearchOperationCallback.java
@@ -31,8 +31,8 @@
 public class SecondaryIndexSearchOperationCallback extends AbstractOperationCallback
         implements ISearchOperationCallback {
 
-    public SecondaryIndexSearchOperationCallback() {
-        super(DatasetId.NULL, null, null, null);
+    public SecondaryIndexSearchOperationCallback(long resourceId) {
+        super(DatasetId.NULL, resourceId, null, null, null);
     }
 
     @Override
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexSearchOperationCallbackFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexSearchOperationCallbackFactory.java
index 0b96164..972668a 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexSearchOperationCallbackFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexSearchOperationCallbackFactory.java
@@ -32,6 +32,6 @@
     @Override
     public ISearchOperationCallback createSearchOperationCallback(long resourceId, IHyracksTaskContext ctx, IOperatorNodePushable operatorNodePushable)
             throws HyracksDataException {
-        return new SecondaryIndexSearchOperationCallback();
+        return new SecondaryIndexSearchOperationCallback(resourceId);
     }
 }
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetPrimaryIndexModificationOperationCallbackFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetPrimaryIndexModificationOperationCallbackFactory.java
index 79ce788..735d7ea 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetPrimaryIndexModificationOperationCallbackFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetPrimaryIndexModificationOperationCallbackFactory.java
@@ -23,7 +23,6 @@
 import org.apache.asterix.common.context.ITransactionSubsystemProvider;
 import org.apache.asterix.common.dataflow.DatasetLocalResource;
 import org.apache.asterix.common.exceptions.ACIDException;
-import org.apache.asterix.common.transactions.AbstractOperationCallback;
 import org.apache.asterix.common.transactions.AbstractOperationCallbackFactory;
 import org.apache.asterix.common.transactions.DatasetId;
 import org.apache.asterix.common.transactions.ITransactionContext;
@@ -67,12 +66,12 @@
         try {
             IJobletEventListenerFactory fact = ctx.getJobletContext().getJobletEventListenerFactory();
             ITransactionContext txnCtx = txnSubsystem.getTransactionManager()
-                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId), false);
+                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId));
             DatasetLocalResource aResource = (DatasetLocalResource) resource.getResource();
             IModificationOperationCallback modCallback = new TempDatasetIndexModificationOperationCallback(
                     new DatasetId(datasetId), primaryKeyFields, txnCtx, txnSubsystem.getLockManager(), txnSubsystem,
                     resource.getId(), aResource.getPartition(), resourceType, indexOp);
-            txnCtx.registerIndexAndCallback(resource.getId(), index, (AbstractOperationCallback) modCallback, true);
+            txnCtx.register(resource.getId(), index, modCallback, true);
             return modCallback;
         } catch (ACIDException e) {
             throw new HyracksDataException(e);
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetSecondaryIndexModificationOperationCallbackFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetSecondaryIndexModificationOperationCallbackFactory.java
index 8a27914..b744606 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetSecondaryIndexModificationOperationCallbackFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetSecondaryIndexModificationOperationCallbackFactory.java
@@ -23,7 +23,6 @@
 import org.apache.asterix.common.context.ITransactionSubsystemProvider;
 import org.apache.asterix.common.dataflow.DatasetLocalResource;
 import org.apache.asterix.common.exceptions.ACIDException;
-import org.apache.asterix.common.transactions.AbstractOperationCallback;
 import org.apache.asterix.common.transactions.AbstractOperationCallbackFactory;
 import org.apache.asterix.common.transactions.DatasetId;
 import org.apache.asterix.common.transactions.ITransactionContext;
@@ -69,11 +68,11 @@
         try {
             IJobletEventListenerFactory fact = ctx.getJobletContext().getJobletEventListenerFactory();
             ITransactionContext txnCtx = txnSubsystem.getTransactionManager()
-                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId), false);
+                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId));
             IModificationOperationCallback modCallback = new TempDatasetIndexModificationOperationCallback(
                     new DatasetId(datasetId), primaryKeyFields, txnCtx, txnSubsystem.getLockManager(), txnSubsystem,
                     resource.getId(), aResource.getPartition(), resourceType, indexOp);
-            txnCtx.registerIndexAndCallback(resource.getId(), index, (AbstractOperationCallback) modCallback, false);
+            txnCtx.register(resource.getId(), index, modCallback, false);
             return modCallback;
         } catch (ACIDException e) {
             throw new HyracksDataException(e);
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/UpsertOperationCallbackFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/UpsertOperationCallbackFactory.java
index dfd3eb1..da4aab8 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/UpsertOperationCallbackFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/UpsertOperationCallbackFactory.java
@@ -22,7 +22,6 @@
 import org.apache.asterix.common.context.ITransactionSubsystemProvider;
 import org.apache.asterix.common.dataflow.DatasetLocalResource;
 import org.apache.asterix.common.exceptions.ACIDException;
-import org.apache.asterix.common.transactions.AbstractOperationCallback;
 import org.apache.asterix.common.transactions.AbstractOperationCallbackFactory;
 import org.apache.asterix.common.transactions.DatasetId;
 import org.apache.asterix.common.transactions.ITransactionContext;
@@ -66,11 +65,11 @@
         try {
             IJobletEventListenerFactory fact = ctx.getJobletContext().getJobletEventListenerFactory();
             ITransactionContext txnCtx = txnSubsystem.getTransactionManager()
-                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId), false);
+                    .getTransactionContext(((IJobEventListenerFactory) fact).getTxnId(txnId));
             IModificationOperationCallback modCallback = new UpsertOperationCallback(new DatasetId(datasetId),
                     primaryKeyFields, txnCtx, txnSubsystem.getLockManager(), txnSubsystem, resource.getId(),
                     aResource.getPartition(), resourceType, indexOp);
-            txnCtx.registerIndexAndCallback(resource.getId(), index, (AbstractOperationCallback) modCallback, true);
+            txnCtx.register(resource.getId(), index, modCallback, true);
             return modCallback;
         } catch (ACIDException e) {
             throw new HyracksDataException(e);
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/runtime/CommitRuntime.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/runtime/CommitRuntime.java
index fe758e1..672e881 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/runtime/CommitRuntime.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/runtime/CommitRuntime.java
@@ -83,7 +83,7 @@
     @Override
     public void open() throws HyracksDataException {
         try {
-            transactionContext = transactionManager.getTransactionContext(txnId, false);
+            transactionContext = transactionManager.getTransactionContext(txnId);
             transactionContext.setWriteTxn(isWriteTransaction);
             ILogMarkerCallback callback = TaskUtil.get(ILogMarkerCallback.KEY_MARKER_CALLBACK, ctx);
             logRecord = new LogRecord(callback);
@@ -111,9 +111,7 @@
                  * active operation count of PrimaryIndexOptracker. By maintaining the count correctly and only allowing
                  * flushing when the count is 0, it can guarantee the no-steal policy for temporary datasets, too.
                  */
-                // TODO: Fix this for upserts. an upsert tuple right now expect to notify the opTracker twice (one for
-                // delete and one for insert)
-                transactionContext.notifyOptracker(false);
+                transactionContext.notifyEntityCommitted();
             } else {
                 tRef.reset(tAccess, t);
                 try {
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogBuffer.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogBuffer.java
index 3d78ad9..6ebf52c 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogBuffer.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogBuffer.java
@@ -234,27 +234,26 @@
     private void batchUnlock(int beginOffset, int endOffset) throws ACIDException {
         if (endOffset > beginOffset) {
             logBufferTailReader.initializeScan(beginOffset, endOffset);
-
             ITransactionContext txnCtx;
-
             LogRecord logRecord = logBufferTailReader.next();
             while (logRecord != null) {
                 if (logRecord.getLogSource() == LogSource.LOCAL) {
                     if (logRecord.getLogType() == LogType.ENTITY_COMMIT) {
                         reusableTxnId.setId(logRecord.getTxnId());
                         reusableDatasetId.setId(logRecord.getDatasetId());
-                        txnCtx = txnSubsystem.getTransactionManager().getTransactionContext(reusableTxnId, false);
+                        txnCtx = txnSubsystem.getTransactionManager().getTransactionContext(reusableTxnId);
                         txnSubsystem.getLockManager().unlock(reusableDatasetId, logRecord.getPKHashValue(),
                                 LockMode.ANY, txnCtx);
-                        txnCtx.notifyOptracker(false);
+                        txnCtx.notifyEntityCommitted();
                         if (txnSubsystem.getTransactionProperties().isCommitProfilerEnabled()) {
                             txnSubsystem.incrementEntityCommitCount();
                         }
+                    } else if (logRecord.getLogType() == LogType.UPDATE) {
+                        reusableTxnId.setId(logRecord.getTxnId());
+                        txnCtx = txnSubsystem.getTransactionManager().getTransactionContext(reusableTxnId);
+                        txnCtx.notifyUpdateCommitted(logRecord.getResourceId());
                     } else if (logRecord.getLogType() == LogType.JOB_COMMIT
                             || logRecord.getLogType() == LogType.ABORT) {
-                        reusableTxnId.setId(logRecord.getTxnId());
-                        txnCtx = txnSubsystem.getTransactionManager().getTransactionContext(reusableTxnId, false);
-                        txnCtx.notifyOptracker(true);
                         notifyJobTermination();
                     } else if (logRecord.getLogType() == LogType.FLUSH) {
                         notifyFlushTermination();
@@ -266,7 +265,6 @@
                         notifyReplicationTermination();
                     }
                 }
-
                 logRecord = logBufferTailReader.next();
             }
         }
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/AbstractTransactionContext.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/AbstractTransactionContext.java
new file mode 100644
index 0000000..43fe266
--- /dev/null
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/AbstractTransactionContext.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.asterix.transaction.management.service.transaction;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.asterix.common.context.ITransactionOperationTracker;
+import org.apache.asterix.common.transactions.ITransactionContext;
+import org.apache.asterix.common.transactions.ITransactionManager;
+import org.apache.asterix.common.transactions.TxnId;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+import org.apache.hyracks.storage.common.IModificationOperationCallback;
+import org.apache.hyracks.util.annotations.ThreadSafe;
+
+@ThreadSafe
+public abstract class AbstractTransactionContext implements ITransactionContext {
+
+    protected final TxnId txnId;
+    protected final Map<Long, ITransactionOperationTracker> txnOpTrackers;
+    private final AtomicLong firstLSN;
+    private final AtomicLong lastLSN;
+    private final AtomicInteger txnState;
+    private final AtomicBoolean isWriteTxn;
+    private boolean isTimeout = false;
+
+    protected AbstractTransactionContext(TxnId txnId) {
+        this.txnId = txnId;
+        firstLSN = new AtomicLong(-1);
+        lastLSN = new AtomicLong(-1);
+        txnState = new AtomicInteger(ITransactionManager.ACTIVE);
+        isTimeout = false;
+        isWriteTxn = new AtomicBoolean();
+        txnOpTrackers = new HashMap<>();
+    }
+
+    @Override
+    public long getFirstLSN() {
+        return firstLSN.get();
+    }
+
+    @Override
+    public void setLastLSN(long newValue) {
+        firstLSN.compareAndSet(-1, newValue);
+        lastLSN.set(Math.max(lastLSN.get(), newValue));
+    }
+
+    @Override
+    public void setTxnState(int txnState) {
+        this.txnState.set(txnState);
+    }
+
+    @Override
+    public int getTxnState() {
+        return txnState.get();
+    }
+
+    @Override
+    public TxnId getTxnId() {
+        return txnId;
+    }
+
+    @Override
+    public synchronized void setTimeout(boolean isTimeout) {
+        this.isTimeout = isTimeout;
+    }
+
+    @Override
+    public synchronized boolean isTimeout() {
+        return isTimeout;
+    }
+
+    @Override
+    public void setWriteTxn(boolean isWriteTxn) {
+        this.isWriteTxn.set(isWriteTxn);
+    }
+
+    @Override
+    public boolean isWriteTxn() {
+        return isWriteTxn.get();
+    }
+
+    @Override
+    public long getLastLSN() {
+        return lastLSN.get();
+    }
+
+    @Override
+    public void complete() {
+        try {
+            if (txnState.get() == ITransactionManager.ABORTED) {
+                cleanupForAbort();
+            }
+        } finally {
+            synchronized (txnOpTrackers) {
+                txnOpTrackers.forEach((resource, opTracker) -> opTracker.afterTransaction(resource));
+            }
+        }
+    }
+
+    @Override
+    public void register(long resourceId, ILSMIndex index, IModificationOperationCallback callback,
+            boolean primaryIndex) {
+        synchronized (txnOpTrackers) {
+            if (!txnOpTrackers.containsKey(resourceId)) {
+                final ITransactionOperationTracker txnOpTracker =
+                        (ITransactionOperationTracker) index.getOperationTracker();
+                txnOpTrackers.put(resourceId, txnOpTracker);
+                txnOpTracker.beforeTransaction(resourceId);
+            }
+        }
+    }
+
+    public String prettyPrint() {
+        StringBuilder sb = new StringBuilder();
+        sb.append("\n" + txnId + "\n");
+        sb.append("isWriteTxn: " + isWriteTxn + "\n");
+        sb.append("firstLSN: " + firstLSN.get() + "\n");
+        sb.append("lastLSN: " + lastLSN.get() + "\n");
+        sb.append("TransactionState: " + txnState + "\n");
+        sb.append("isTimeout: " + isTimeout + "\n");
+        return sb.toString();
+    }
+
+    protected abstract void cleanupForAbort();
+}
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/AtomicTransactionContext.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/AtomicTransactionContext.java
new file mode 100644
index 0000000..1d132a8
--- /dev/null
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/AtomicTransactionContext.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.asterix.transaction.management.service.transaction;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.asterix.common.context.PrimaryIndexOperationTracker;
+import org.apache.asterix.common.exceptions.ACIDException;
+import org.apache.asterix.common.transactions.TxnId;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
+import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType;
+import org.apache.hyracks.storage.common.IModificationOperationCallback;
+import org.apache.hyracks.util.annotations.ThreadSafe;
+
+@ThreadSafe
+public class AtomicTransactionContext extends AbstractTransactionContext {
+
+    private final Map<Long, ILSMOperationTracker> opTrackers = new HashMap<>();
+    private final Map<Long, AtomicInteger> indexPendingOps = new HashMap<>();
+    private final Map<Long, IModificationOperationCallback> callbacks = new HashMap<>();
+
+    public AtomicTransactionContext(TxnId txnId) {
+        super(txnId);
+    }
+
+    @Override
+    public void register(long resourceId, ILSMIndex index, IModificationOperationCallback callback,
+            boolean primaryIndex) {
+        super.register(resourceId, index, callback, primaryIndex);
+        synchronized (txnOpTrackers) {
+            if (primaryIndex && !opTrackers.containsKey(resourceId)) {
+                opTrackers.put(resourceId, index.getOperationTracker());
+                callbacks.put(resourceId, callback);
+                indexPendingOps.put(resourceId, new AtomicInteger(0));
+            }
+        }
+    }
+
+    @Override
+    public void notifyUpdateCommitted(long resourceId) {
+        try {
+            opTrackers.get(resourceId).completeOperation(null, LSMOperationType.MODIFICATION, null,
+                    callbacks.get(resourceId));
+        } catch (HyracksDataException e) {
+            throw new ACIDException(e);
+        }
+    }
+
+    @Override
+    public void notifyEntityCommitted() {
+        throw new IllegalStateException("Unexpected entity commit in atomic transaction");
+    }
+
+    @Override
+    public void beforeOperation(long resourceId) {
+        indexPendingOps.get(resourceId).incrementAndGet();
+    }
+
+    @Override
+    public void afterOperation(long resourceId) {
+        indexPendingOps.get(resourceId).decrementAndGet();
+    }
+
+    @Override
+    public void cleanupForAbort() {
+        // each opTracker should be cleaned
+        opTrackers.forEach((resId, opTracker) -> ((PrimaryIndexOperationTracker) opTracker)
+                .cleanupNumActiveOperationsForAbortedJob(indexPendingOps.get(resId).get()));
+    }
+
+    @Override
+    public int hashCode() {
+        return Long.hashCode(txnId.getId());
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        AtomicTransactionContext that = (AtomicTransactionContext) o;
+        return this.txnId.equals(that.txnId);
+    }
+}
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/EntityLevelTransactionContext.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/EntityLevelTransactionContext.java
new file mode 100644
index 0000000..e195451
--- /dev/null
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/EntityLevelTransactionContext.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.asterix.transaction.management.service.transaction;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.asterix.common.context.PrimaryIndexOperationTracker;
+import org.apache.asterix.common.exceptions.ACIDException;
+import org.apache.asterix.common.transactions.TxnId;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType;
+import org.apache.hyracks.storage.common.IModificationOperationCallback;
+import org.apache.hyracks.util.annotations.ThreadSafe;
+
+@ThreadSafe
+public class EntityLevelTransactionContext extends AbstractTransactionContext {
+
+    private PrimaryIndexOperationTracker primaryIndexOpTracker;
+    private IModificationOperationCallback primaryIndexCallback;
+    private final AtomicInteger pendingOps;
+
+    public EntityLevelTransactionContext(TxnId txnId) {
+        super(txnId);
+        pendingOps = new AtomicInteger(0);
+    }
+
+    @Override
+    public void register(long resourceId, ILSMIndex index, IModificationOperationCallback callback,
+            boolean primaryIndex) {
+        super.register(resourceId, index, callback, primaryIndex);
+        synchronized (txnOpTrackers) {
+            if (primaryIndex && primaryIndexOpTracker == null) {
+                primaryIndexCallback = callback;
+                primaryIndexOpTracker = (PrimaryIndexOperationTracker) index.getOperationTracker();
+            }
+        }
+    }
+
+    @Override
+    public void beforeOperation(long resourceId) {
+        pendingOps.incrementAndGet();
+    }
+
+    @Override
+    public void notifyUpdateCommitted(long resourceId) {
+        // no op
+    }
+
+    @Override
+    public void notifyEntityCommitted() {
+        try {
+            primaryIndexOpTracker.completeOperation(null, LSMOperationType.MODIFICATION, null, primaryIndexCallback);
+        } catch (HyracksDataException e) {
+            throw new ACIDException(e);
+        }
+    }
+
+    @Override
+    public void afterOperation(long resourceId) {
+        pendingOps.decrementAndGet();
+    }
+
+    @Override
+    protected void cleanupForAbort() {
+        if (primaryIndexOpTracker != null) {
+            primaryIndexOpTracker.cleanupNumActiveOperationsForAbortedJob(pendingOps.get());
+        }
+    }
+
+    @Override
+    public int hashCode() {
+        return Long.hashCode(txnId.getId());
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        EntityLevelTransactionContext that = (EntityLevelTransactionContext) o;
+        return this.txnId.equals(that.txnId);
+    }
+}
\ No newline at end of file
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/FieldsHashValueGenerator.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/FieldsHashValueGenerator.java
deleted file mode 100644
index c408f1d..0000000
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/FieldsHashValueGenerator.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you 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.apache.asterix.transaction.management.service.transaction;
-
-import org.apache.hyracks.api.dataflow.value.IBinaryHashFunction;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
-
-public class FieldsHashValueGenerator {
-    public static int computeFieldsHashValue(ITupleReference tuple, int[] fieldIndexes,
-            IBinaryHashFunction[] fieldHashFunctions) throws HyracksDataException {
-        int h = 0;
-        for (int i = 0; i < fieldIndexes.length; i++) {
-            int primaryKeyFieldIdx = fieldIndexes[i];
-            int fh = fieldHashFunctions[i].hash(tuple.getFieldData(primaryKeyFieldIdx),
-                    tuple.getFieldStart(primaryKeyFieldIdx), tuple.getFieldLength(primaryKeyFieldIdx));
-            h = h * 31 + fh;
-            if (h < 0) {
-                h = h * (-1);
-            }
-        }
-        return h;
-    }
-}
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/MutableResourceId.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/MutableResourceId.java
deleted file mode 100644
index c8e134c..0000000
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/MutableResourceId.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you 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.apache.asterix.transaction.management.service.transaction;
-
-public class MutableResourceId {
-    long id;
-
-    public MutableResourceId(long id) {
-        this.id = id;
-    }
-
-    public void setId(long id) {
-        this.id = id;
-    }
-
-    public long getId() {
-        return id;
-    }
-
-    @Override
-    public int hashCode() {
-        return (int) id;
-    }
-
-    @Override
-    public boolean equals(Object o) {
-        if ((o == null) || !(o instanceof MutableResourceId)) {
-            return false;
-        }
-        return ((MutableResourceId) o).id == this.id;
-    }
-}
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/TransactionContext.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/TransactionContext.java
deleted file mode 100644
index 1681a27..0000000
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/TransactionContext.java
+++ /dev/null
@@ -1,271 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you 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.apache.asterix.transaction.management.service.transaction;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.apache.asterix.common.context.ITransactionOperationTracker;
-import org.apache.asterix.common.context.PrimaryIndexOperationTracker;
-import org.apache.asterix.common.exceptions.ACIDException;
-import org.apache.asterix.common.transactions.AbstractOperationCallback;
-import org.apache.asterix.common.transactions.ITransactionContext;
-import org.apache.asterix.common.transactions.ITransactionManager;
-import org.apache.asterix.common.transactions.LogRecord;
-import org.apache.asterix.common.transactions.TxnId;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
-import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType;
-import org.apache.hyracks.storage.common.IModificationOperationCallback;
-
-/*
- * An object of TransactionContext is created and accessed(read/written) by multiple threads which work for
- * a single job identified by a txnId. Thus, the member variables in the object can be read/written
- * concurrently. Please see each variable declaration to know which one is accessed concurrently and
- * which one is not.
- */
-public class TransactionContext implements ITransactionContext {
-
-    private static final long serialVersionUID = -6105616785783310111L;
-
-    // txnId is set once and read concurrently.
-    private final TxnId txnId;
-
-    // There are no concurrent writers on both firstLSN and lastLSN
-    // since both values are updated by serialized log appenders.
-    // But readers and writers can be different threads,
-    // so both LSNs are atomic variables in order to be read and written
-    // atomically.
-    private final AtomicLong firstLSN;
-    private final AtomicLong lastLSN;
-
-    // txnState is read and written concurrently.
-    private final AtomicInteger txnState;
-
-    // isTimeout is read and written under the lockMgr's tableLatch
-    // Thus, no other synchronization is required separately.
-    private boolean isTimeout;
-
-    // isWriteTxn can be set concurrently by multiple threads.
-    private final AtomicBoolean isWriteTxn;
-
-    // isMetadataTxn is accessed by a single thread since the metadata is not
-    // partitioned
-    private boolean isMetadataTxn;
-
-    // indexMap is concurrently accessed by multiple threads,
-    // so those threads are synchronized on indexMap object itself
-    private final Map<Long, ITransactionOperationTracker> indexMap;
-
-    // TODO: fix ComponentLSNs' issues.
-    // primaryIndex, primaryIndexCallback, and primaryIndexOptracker will be
-    // modified accordingly
-    // when the issues of componentLSNs are fixed.
-    private ILSMIndex primaryIndex;
-    private AbstractOperationCallback primaryIndexCallback;
-    private PrimaryIndexOperationTracker primaryIndexOpTracker;
-
-    // The following three variables are used as temporary variables in order to
-    // avoid object creations.
-    // Those are used in synchronized methods.
-    private final LogRecord logRecord;
-
-    private final AtomicInteger transactorNumActiveOperations;
-
-    // TODO: implement transactionContext pool in order to avoid object
-    // creations.
-    // also, the pool can throttle the number of concurrent active jobs at every
-    // moment.
-    public TransactionContext(TxnId txnId) throws ACIDException {
-        this.txnId = txnId;
-        firstLSN = new AtomicLong(-1);
-        lastLSN = new AtomicLong(-1);
-        txnState = new AtomicInteger(ITransactionManager.ACTIVE);
-        isTimeout = false;
-        isWriteTxn = new AtomicBoolean(false);
-        isMetadataTxn = false;
-        indexMap = new HashMap<>();
-        primaryIndex = null;
-        logRecord = new LogRecord();
-        transactorNumActiveOperations = new AtomicInteger(0);
-    }
-
-    @Override
-    public void registerIndexAndCallback(long resourceId, ILSMIndex index, AbstractOperationCallback callback,
-            boolean isPrimaryIndex) {
-        synchronized (indexMap) {
-            if (isPrimaryIndex && primaryIndex == null) {
-                primaryIndex = index;
-                primaryIndexCallback = callback;
-                primaryIndexOpTracker = (PrimaryIndexOperationTracker) index.getOperationTracker();
-            }
-            if (!indexMap.containsKey(resourceId)) {
-                final ITransactionOperationTracker txnOpTracker =
-                        (ITransactionOperationTracker) index.getOperationTracker();
-                indexMap.put(resourceId, txnOpTracker);
-                txnOpTracker.beforeTransaction(resourceId);
-            }
-        }
-    }
-
-    public PrimaryIndexOperationTracker getPrimaryIndexOpTracker() {
-        synchronized (indexMap) {
-            return primaryIndexOpTracker;
-        }
-    }
-
-    // [Notice]
-    // This method is called sequentially by the LogAppender threads.
-    @Override
-    public void setLastLSN(long LSN) {
-        firstLSN.compareAndSet(-1, LSN);
-        lastLSN.set(Math.max(lastLSN.get(), LSN));
-    }
-
-    @Override
-    public void notifyOptracker(boolean isJobLevelCommit) {
-        try {
-            /**
-             * in case of transaction abort {@link TransactionContext#cleanupForAbort()} will
-             * clean the primaryIndexOpTracker state.
-             */
-            if (isJobLevelCommit && isMetadataTxn && txnState.get() != ITransactionManager.ABORTED) {
-                primaryIndexOpTracker.exclusiveJobCommitted();
-            } else if (!isJobLevelCommit) {
-                primaryIndexOpTracker.completeOperation(null, LSMOperationType.MODIFICATION, null,
-                        (IModificationOperationCallback) primaryIndexCallback);
-            }
-        } catch (HyracksDataException e) {
-            throw new IllegalStateException(e);
-        }
-    }
-
-    @Override
-    public void setWriteTxn(boolean isWriteTxn) {
-        this.isWriteTxn.set(isWriteTxn);
-    }
-
-    @Override
-    public boolean isWriteTxn() {
-        return isWriteTxn.get();
-    }
-
-    @Override
-    public long getFirstLSN() {
-        return firstLSN.get();
-    }
-
-    @Override
-    public long getLastLSN() {
-        return lastLSN.get();
-    }
-
-    @Override
-    public TxnId getTxnId() {
-        return txnId;
-    }
-
-    @Override
-    public void setTimeout(boolean isTimeout) {
-        this.isTimeout = isTimeout;
-    }
-
-    @Override
-    public boolean isTimeout() {
-        return isTimeout;
-    }
-
-    @Override
-    public void setTxnState(int txnState) {
-        this.txnState.set(txnState);
-    }
-
-    @Override
-    public int getTxnState() {
-        return txnState.get();
-    }
-
-    @Override
-    public int hashCode() {
-        return Long.hashCode(txnId.getId());
-    }
-
-    @Override
-    public boolean equals(Object o) {
-        return (o == this);
-    }
-
-    @Override
-    public void setMetadataTransaction(boolean isMetadataTxn) {
-        this.isMetadataTxn = isMetadataTxn;
-    }
-
-    @Override
-    public boolean isMetadataTransaction() {
-        return isMetadataTxn;
-    }
-
-    @Override
-    public String prettyPrint() {
-        StringBuilder sb = new StringBuilder();
-        sb.append("\n" + txnId + "\n");
-        sb.append("isWriteTxn: " + isWriteTxn + "\n");
-        sb.append("firstLSN: " + firstLSN.get() + "\n");
-        sb.append("lastLSN: " + lastLSN.get() + "\n");
-        sb.append("TransactionState: " + txnState + "\n");
-        sb.append("isTimeout: " + isTimeout + "\n");
-        return sb.toString();
-    }
-
-    public LogRecord getLogRecord() {
-        return logRecord;
-    }
-
-    @Override
-    public void incrementNumActiveOperations() {
-        transactorNumActiveOperations.incrementAndGet();
-    }
-
-    @Override
-    public void decrementNumActiveOperations() {
-        transactorNumActiveOperations.decrementAndGet();
-    }
-
-    @Override
-    public void complete() {
-        try {
-            if (txnState.get() == ITransactionManager.ABORTED) {
-                cleanupForAbort();
-            }
-        } finally {
-            synchronized (indexMap) {
-                indexMap.forEach((resource, opTracker) -> opTracker.afterTransaction(resource));
-            }
-        }
-    }
-
-    private void cleanupForAbort() {
-        if (primaryIndexOpTracker != null) {
-            primaryIndexOpTracker.cleanupNumActiveOperationsForAbortedJob(transactorNumActiveOperations.get());
-        }
-    }
-}
\ No newline at end of file
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/TransactionContextFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/TransactionContextFactory.java
new file mode 100644
index 0000000..4a465a4
--- /dev/null
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/TransactionContextFactory.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.asterix.transaction.management.service.transaction;
+
+import static org.apache.asterix.common.transactions.ITransactionManager.AtomicityLevel;
+
+import org.apache.asterix.common.transactions.ITransactionContext;
+import org.apache.asterix.common.transactions.TransactionOptions;
+import org.apache.asterix.common.transactions.TxnId;
+
+public class TransactionContextFactory {
+
+    private TransactionContextFactory() {
+    }
+
+    public static ITransactionContext create(TxnId txnId, TransactionOptions options) {
+        final AtomicityLevel atomicityLevel = options.getAtomicityLevel();
+        switch (atomicityLevel) {
+            case ATOMIC:
+                return new AtomicTransactionContext(txnId);
+            case ENTITY_LEVEL:
+                return new EntityLevelTransactionContext(txnId);
+            default:
+                throw new IllegalStateException("Unknown transaction context type: " + atomicityLevel);
+        }
+    }
+}
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/TransactionManager.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/TransactionManager.java
index 1799ea1..c03369b 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/TransactionManager.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/TransactionManager.java
@@ -18,6 +18,7 @@
  */
 package org.apache.asterix.transaction.management.service.transaction;
 
+import java.io.IOException;
 import java.io.OutputStream;
 import java.util.Map;
 import java.util.Set;
@@ -27,124 +28,93 @@
 import java.util.logging.Logger;
 
 import org.apache.asterix.common.exceptions.ACIDException;
-import org.apache.asterix.common.transactions.DatasetId;
 import org.apache.asterix.common.transactions.ITransactionContext;
 import org.apache.asterix.common.transactions.ITransactionManager;
 import org.apache.asterix.common.transactions.ITransactionSubsystem;
-import org.apache.asterix.common.transactions.TxnId;
 import org.apache.asterix.common.transactions.LogRecord;
+import org.apache.asterix.common.transactions.TransactionOptions;
+import org.apache.asterix.common.transactions.TxnId;
 import org.apache.asterix.common.utils.TransactionUtil;
 import org.apache.hyracks.api.lifecycle.ILifeCycleComponent;
+import org.apache.hyracks.util.annotations.ThreadSafe;
 
-/**
- * An implementation of the @see ITransactionManager interface that provides
- * implementation of APIs for governing the lifecycle of a transaction.
- */
+@ThreadSafe
 public class TransactionManager implements ITransactionManager, ILifeCycleComponent {
 
-    public static final boolean IS_DEBUG_MODE = false;//true
     private static final Logger LOGGER = Logger.getLogger(TransactionManager.class.getName());
     private final ITransactionSubsystem txnSubsystem;
-    private Map<TxnId, ITransactionContext> transactionContextRepository = new ConcurrentHashMap<>();
-    private AtomicLong maxTxnId = new AtomicLong(0);
+    private final Map<TxnId, ITransactionContext> txnCtxRepository = new ConcurrentHashMap<>();
+    private final AtomicLong maxTxnId = new AtomicLong(0);
 
     public TransactionManager(ITransactionSubsystem provider) {
         this.txnSubsystem = provider;
     }
 
     @Override
-    public void abortTransaction(ITransactionContext txnCtx, DatasetId datasetId, int PKHashVal) throws ACIDException {
-        if (txnCtx.getTxnState() != ITransactionManager.ABORTED) {
-            txnCtx.setTxnState(ITransactionManager.ABORTED);
+    public synchronized ITransactionContext beginTransaction(TxnId txnId, TransactionOptions options)
+            throws ACIDException {
+        ITransactionContext txnCtx = txnCtxRepository.get(txnId);
+        if (txnCtx != null) {
+            throw new ACIDException("Transaction with the same (" + txnId + ") already exists");
         }
-        try {
-            if (txnCtx.isWriteTxn()) {
-                LogRecord logRecord = ((TransactionContext) txnCtx).getLogRecord();
-                TransactionUtil.formJobTerminateLogRecord(txnCtx, logRecord, false);
-                txnSubsystem.getLogManager().log(logRecord);
-                txnSubsystem.getRecoveryManager().rollbackTransaction(txnCtx);
-            }
-        } catch (Exception ae) {
-            String msg = "Could not complete rollback! System is in an inconsistent state";
-            if (LOGGER.isLoggable(Level.SEVERE)) {
-                LOGGER.severe(msg);
-            }
-            ae.printStackTrace();
-            throw new ACIDException(msg, ae);
-        } finally {
-            txnCtx.complete();
-            txnSubsystem.getLockManager().releaseLocks(txnCtx);
-            transactionContextRepository.remove(txnCtx.getTxnId());
-        }
+        txnCtx = TransactionContextFactory.create(txnId, options);
+        txnCtxRepository.put(txnId, txnCtx);
+        ensureMaxTxnId(txnId.getId());
+        return txnCtx;
     }
 
     @Override
-    public ITransactionContext beginTransaction(TxnId txnId) throws ACIDException {
-        return getTransactionContext(txnId, true);
-    }
-
-    @Override
-    public ITransactionContext getTransactionContext(TxnId txnId, boolean createIfNotExist) throws ACIDException {
-        setMaxTxnId(txnId.getId());
-        ITransactionContext txnCtx = transactionContextRepository.get(txnId);
+    public ITransactionContext getTransactionContext(TxnId txnId) throws ACIDException {
+        ITransactionContext txnCtx = txnCtxRepository.get(txnId);
         if (txnCtx == null) {
-            if (createIfNotExist) {
-                synchronized (this) {
-                    txnCtx = transactionContextRepository.get(txnId);
-                    if (txnCtx == null) {
-                        txnCtx = new TransactionContext(txnId);
-                        transactionContextRepository.put(txnId, txnCtx);
-                    }
-                }
-            } else {
-                throw new ACIDException("TransactionContext of " + txnId + " doesn't exist.");
-            }
+            throw new ACIDException("Transaction " + txnId + " doesn't exist.");
         }
         return txnCtx;
     }
 
     @Override
-    public void commitTransaction(ITransactionContext txnCtx, DatasetId datasetId, int PKHashVal)
-            throws ACIDException {
-        //Only job-level commits call this method.
+    public void commitTransaction(TxnId txnId) throws ACIDException {
+        final ITransactionContext txnCtx = getTransactionContext(txnId);
         try {
             if (txnCtx.isWriteTxn()) {
-                LogRecord logRecord = ((TransactionContext) txnCtx).getLogRecord();
+                LogRecord logRecord = new LogRecord();
                 TransactionUtil.formJobTerminateLogRecord(txnCtx, logRecord, true);
                 txnSubsystem.getLogManager().log(logRecord);
+                txnCtx.setTxnState(ITransactionManager.COMMITTED);
             }
-        } catch (Exception ae) {
+        } catch (Exception e) {
             if (LOGGER.isLoggable(Level.SEVERE)) {
                 LOGGER.severe(" caused exception in commit !" + txnCtx.getTxnId());
             }
-            throw ae;
+            throw e;
         } finally {
             txnCtx.complete();
             txnSubsystem.getLockManager().releaseLocks(txnCtx);
-            transactionContextRepository.remove(txnCtx.getTxnId());
-            txnCtx.setTxnState(ITransactionManager.COMMITTED);
+            txnCtxRepository.remove(txnCtx.getTxnId());
         }
     }
 
     @Override
-    public void completedTransaction(ITransactionContext txnContext, DatasetId datasetId, int PKHashVal,
-            boolean success) throws ACIDException {
-        if (!success) {
-            abortTransaction(txnContext, datasetId, PKHashVal);
-        } else {
-            commitTransaction(txnContext, datasetId, PKHashVal);
-        }
-    }
-
-    @Override
-    public ITransactionSubsystem getTransactionSubsystem() {
-        return txnSubsystem;
-    }
-
-    public void setMaxTxnId(long txnId) {
-        long maxId = maxTxnId.get();
-        if (txnId > maxId) {
-            maxTxnId.compareAndSet(maxId, txnId);
+    public void abortTransaction(TxnId txnId) throws ACIDException {
+        final ITransactionContext txnCtx = getTransactionContext(txnId);
+        try {
+            if (txnCtx.isWriteTxn()) {
+                LogRecord logRecord = new LogRecord();
+                TransactionUtil.formJobTerminateLogRecord(txnCtx, logRecord, false);
+                txnSubsystem.getLogManager().log(logRecord);
+                txnSubsystem.getRecoveryManager().rollbackTransaction(txnCtx);
+                txnCtx.setTxnState(ITransactionManager.ABORTED);
+            }
+        } catch (ACIDException e) {
+            String msg = "Could not complete rollback! System is in an inconsistent state";
+            if (LOGGER.isLoggable(Level.SEVERE)) {
+                LOGGER.log(Level.SEVERE, msg, e);
+            }
+            throw new ACIDException(msg, e);
+        } finally {
+            txnCtx.complete();
+            txnSubsystem.getLockManager().releaseLocks(txnCtx);
+            txnCtxRepository.remove(txnCtx.getTxnId());
         }
     }
 
@@ -167,45 +137,41 @@
 
     @Override
     public void dumpState(OutputStream os) {
-        //#. dump TxnContext
         dumpTxnContext(os);
     }
 
+    private void ensureMaxTxnId(long txnId) {
+        maxTxnId.updateAndGet(current -> Math.max(current, txnId));
+    }
+
     private void dumpTxnContext(OutputStream os) {
         TxnId txnId;
         ITransactionContext txnCtx;
         StringBuilder sb = new StringBuilder();
-
         try {
             sb.append("\n>>dump_begin\t>>----- [ConfVars] -----");
-            Set<Map.Entry<TxnId, ITransactionContext>> entrySet = transactionContextRepository.entrySet();
-            if (entrySet != null) {
-                for (Map.Entry<TxnId, ITransactionContext> entry : entrySet) {
-                    if (entry != null) {
-                        txnId = entry.getKey();
-                        if (txnId != null) {
-                            sb.append("\n" + txnId);
-                        } else {
-                            sb.append("\nJID:null");
-                        }
+            Set<Map.Entry<TxnId, ITransactionContext>> entrySet = txnCtxRepository.entrySet();
+            for (Map.Entry<TxnId, ITransactionContext> entry : entrySet) {
+                if (entry != null) {
+                    txnId = entry.getKey();
+                    if (txnId != null) {
+                        sb.append("\n" + txnId);
+                    } else {
+                        sb.append("\nJID:null");
+                    }
 
-                        txnCtx = entry.getValue();
-                        if (txnCtx != null) {
-                            sb.append(txnCtx.prettyPrint());
-                        } else {
-                            sb.append("\nTxnCtx:null");
-                        }
+                    txnCtx = entry.getValue();
+                    if (txnCtx != null) {
+                        sb.append(((AbstractTransactionContext) txnCtx).prettyPrint());
+                    } else {
+                        sb.append("\nTxnCtx:null");
                     }
                 }
             }
-
             sb.append("\n>>dump_end\t>>----- [ConfVars] -----\n");
             os.write(sb.toString().getBytes());
-        } catch (Exception e) {
-            //ignore exception and continue dumping as much as possible.
-            if (IS_DEBUG_MODE) {
-                e.printStackTrace();
-            }
+        } catch (IOException e) {
+            LOGGER.log(Level.WARNING, "exception while dumping state", e);
         }
     }
 }