diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
index 6a2b4e0..e9c8cf7 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
@@ -31,9 +31,9 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.Map.Entry;
 import java.util.Properties;
 import java.util.Random;
+import java.util.Map.Entry;
 import java.util.concurrent.ExecutorService;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -52,15 +52,16 @@
 import org.apache.asterix.app.result.ResultHandle;
 import org.apache.asterix.app.result.ResultReader;
 import org.apache.asterix.common.config.ClusterProperties;
+import org.apache.asterix.common.config.ExternalProperties;
+import org.apache.asterix.common.config.GlobalConfig;
 import org.apache.asterix.common.config.DatasetConfig.DatasetType;
 import org.apache.asterix.common.config.DatasetConfig.ExternalFilePendingOp;
 import org.apache.asterix.common.config.DatasetConfig.IndexType;
 import org.apache.asterix.common.config.DatasetConfig.TransactionState;
-import org.apache.asterix.common.config.ExternalProperties;
-import org.apache.asterix.common.config.GlobalConfig;
 import org.apache.asterix.common.context.IStorageComponentProvider;
 import org.apache.asterix.common.dataflow.ICcApplicationContext;
 import org.apache.asterix.common.exceptions.ACIDException;
+import org.apache.asterix.common.exceptions.AsterixException;
 import org.apache.asterix.common.exceptions.CompilationException;
 import org.apache.asterix.common.exceptions.ErrorCode;
 import org.apache.asterix.common.functions.FunctionSignature;
@@ -153,16 +154,16 @@
 import org.apache.asterix.om.types.TypeSignature;
 import org.apache.asterix.transaction.management.service.transaction.DatasetIdFactory;
 import org.apache.asterix.translator.AbstractLangTranslator;
-import org.apache.asterix.translator.CompiledStatements.CompiledDeleteStatement;
-import org.apache.asterix.translator.CompiledStatements.CompiledInsertStatement;
-import org.apache.asterix.translator.CompiledStatements.CompiledLoadFromFileStatement;
-import org.apache.asterix.translator.CompiledStatements.CompiledUpsertStatement;
-import org.apache.asterix.translator.CompiledStatements.ICompiledDmlStatement;
 import org.apache.asterix.translator.IStatementExecutor;
 import org.apache.asterix.translator.IStatementExecutorContext;
 import org.apache.asterix.translator.SessionConfig;
 import org.apache.asterix.translator.SessionOutput;
 import org.apache.asterix.translator.TypeTranslator;
+import org.apache.asterix.translator.CompiledStatements.CompiledDeleteStatement;
+import org.apache.asterix.translator.CompiledStatements.CompiledInsertStatement;
+import org.apache.asterix.translator.CompiledStatements.CompiledLoadFromFileStatement;
+import org.apache.asterix.translator.CompiledStatements.CompiledUpsertStatement;
+import org.apache.asterix.translator.CompiledStatements.ICompiledDmlStatement;
 import org.apache.asterix.translator.util.ValidateUtil;
 import org.apache.asterix.utils.DataverseUtil;
 import org.apache.asterix.utils.FeedOperations;
@@ -1719,7 +1720,7 @@
         String dataverseName = getActiveDataverse(stmtInsertUpsert.getDataverseName());
         final IMetadataLocker locker = new IMetadataLocker() {
             @Override
-            public void lock() {
+            public void lock() throws AsterixException {
                 MetadataLockManager.INSTANCE.insertDeleteUpsertBegin(metadataProvider.getLocks(),
                         dataverseName + "." + stmtInsertUpsert.getDatasetName());
             }
@@ -2277,9 +2278,9 @@
     }
 
     private interface IMetadataLocker {
-        void lock();
+        void lock() throws AsterixException;
 
-        void unlock();
+        void unlock() throws AsterixException;
     }
 
     private interface IResultPrinter {
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
index b7eef11..f06a0b7 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
@@ -99,6 +99,7 @@
     public static final int UNKNOWN_SEARCH_MODIFIER = 1036;
     public static final int COMPILATION_BAD_QUERY_PARAMETER_VALUE = 1037;
     public static final int COMPILATION_ILLEGAL_STATE = 1038;
+    public static final int COMPILATION_TWO_PHASE_LOCKING_VIOLATION = 1039;
 
     // Feed errors
     public static final int DATAFLOW_ILLEGAL_STATE = 3001;
diff --git a/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties b/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
index c1f5cc7..bba3a43 100644
--- a/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
+++ b/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
@@ -85,6 +85,7 @@
 1036 = Unknown search modifier type %1$s
 1037 = Invalid query parameter %1$s -- value has to be greater than or equal to %2$s bytes
 1038 = Illegal state. %1$s
+1039 = Two-phase locking violation -- locks can not be acquired after unlocking
 
 # Feed Errors
 3001 = Illegal state.
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/LockList.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/LockList.java
index 6e6f086..bb708db 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/LockList.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/LockList.java
@@ -21,25 +21,51 @@
 import java.util.ArrayList;
 import java.util.List;
 
-import org.apache.asterix.metadata.lock.IMetadataLock.Mode;
+import org.apache.asterix.common.exceptions.AsterixException;
+import org.apache.asterix.common.exceptions.ErrorCode;
 import org.apache.commons.lang3.tuple.Pair;
 
+/**
+ * The LockList is used for two phase locking.
+ */
 public class LockList {
-    List<Pair<IMetadataLock.Mode, IMetadataLock>> locks = new ArrayList<>();
+    private List<Pair<IMetadataLock.Mode, IMetadataLock>> locks = new ArrayList<>();
+    private boolean lockPhase = true;
 
-    public void add(IMetadataLock.Mode mode, IMetadataLock lock) {
+    /**
+     * Acquires a lock.
+     *
+     * @param mode
+     *            the lock mode.
+     * @param lock
+     *            the lock object.
+     */
+    public void add(IMetadataLock.Mode mode, IMetadataLock lock) throws AsterixException {
+        if (!lockPhase) {
+            throw new AsterixException(ErrorCode.COMPILATION_TWO_PHASE_LOCKING_VIOLATION);
+        }
         lock.acquire(mode);
         locks.add(Pair.of(mode, lock));
     }
 
+    /**
+     * Once unlock() is called, no caller can call add(IMetadataLock.Mode mode, IMetadataLock lock),
+     * except that reset() is called.
+     */
     public void unlock() {
         for (int i = locks.size() - 1; i >= 0; i--) {
             Pair<IMetadataLock.Mode, IMetadataLock> pair = locks.get(i);
             pair.getRight().release(pair.getLeft());
         }
+        locks.clear();
+        lockPhase = false;
     }
 
+    /**
+     * Clears the state and starts another pass of two phase locking again.
+     */
     public void reset() {
-        locks.clear();
+        unlock();
+        lockPhase = true;
     }
 }
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/MetadataLockManager.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/MetadataLockManager.java
index 5ae3aa4..6c8999a 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/MetadataLockManager.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/MetadataLockManager.java
@@ -22,6 +22,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
 
+import org.apache.asterix.common.exceptions.AsterixException;
 import org.apache.asterix.metadata.entities.FeedConnection;
 import org.apache.asterix.metadata.utils.DatasetUtil;
 
@@ -54,115 +55,104 @@
     }
 
     // Dataverse
-    public void acquireDataverseReadLock(LockList locks, String dataverseName) {
+    public void acquireDataverseReadLock(LockList locks, String dataverseName) throws AsterixException {
         MetadataLock lock = dataversesLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
-    public void acquireDataverseWriteLock(LockList locks, String dataverseName) {
+    public void acquireDataverseWriteLock(LockList locks, String dataverseName) throws AsterixException {
         MetadataLock lock = dataversesLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     // Dataset
-    public void acquireDatasetReadLock(LockList locks, String datasetName) {
+    public void acquireDatasetReadLock(LockList locks, String datasetName) throws AsterixException {
         DatasetLock lock = datasetsLocks.computeIfAbsent(datasetName, DATASET_LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
-    public void acquireDatasetWriteLock(LockList locks, String datasetName) {
+    public void acquireDatasetWriteLock(LockList locks, String datasetName) throws AsterixException {
         DatasetLock lock = datasetsLocks.computeIfAbsent(datasetName, DATASET_LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
-    public void acquireDatasetModifyLock(LockList locks, String datasetName) {
+    public void acquireDatasetModifyLock(LockList locks, String datasetName) throws AsterixException {
         DatasetLock lock = datasetsLocks.computeIfAbsent(datasetName, DATASET_LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.MODIFY, lock);
     }
 
-    public void acquireDatasetCreateIndexLock(LockList locks, String datasetName) {
+    public void acquireDatasetCreateIndexLock(LockList locks, String datasetName) throws AsterixException {
         DatasetLock lock = datasetsLocks.computeIfAbsent(datasetName, DATASET_LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.INDEX_BUILD, lock);
     }
 
-    public void acquireExternalDatasetRefreshLock(LockList locks, String datasetName) {
+    public void acquireExternalDatasetRefreshLock(LockList locks, String datasetName) throws AsterixException {
         DatasetLock lock = datasetsLocks.computeIfAbsent(datasetName, DATASET_LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.INDEX_BUILD, lock);
     }
 
     // Function
-    public void acquireFunctionReadLock(LockList locks, String dataverseName) {
+    public void acquireFunctionReadLock(LockList locks, String dataverseName) throws AsterixException {
         MetadataLock lock = functionsLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
-    public void acquireFunctionWriteLock(LockList locks, String dataverseName) {
+    public void acquireFunctionWriteLock(LockList locks, String dataverseName) throws AsterixException {
         MetadataLock lock = functionsLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     // Node Group
-    public void acquireNodeGroupReadLock(LockList locks, String dataverseName) {
+    public void acquireNodeGroupReadLock(LockList locks, String dataverseName) throws AsterixException {
         MetadataLock lock = nodeGroupsLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
-    public void acquireNodeGroupWriteLock(LockList locks, String dataverseName) {
+    public void acquireNodeGroupWriteLock(LockList locks, String dataverseName) throws AsterixException {
         MetadataLock lock = nodeGroupsLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     // Feeds
-    public void acquireFeedReadLock(LockList locks, String dataverseName) {
+    public void acquireFeedReadLock(LockList locks, String dataverseName) throws AsterixException {
         MetadataLock lock = feedsLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
-    public void acquireFeedWriteLock(LockList locks, String dataverseName) {
+    public void acquireFeedWriteLock(LockList locks, String dataverseName) throws AsterixException {
         MetadataLock lock = feedsLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
-    // Feed Policies
-    public void acquireFeedPolicyReadLock(LockList locks, String dataverseName) {
-        MetadataLock lock = feedPolicyLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
-        locks.add(IMetadataLock.Mode.READ, lock);
-    }
-
-    public void acquireFeedPolicyWriteLock(LockList locks, String dataverseName) {
+    public void acquireFeedPolicyWriteLock(LockList locks, String dataverseName) throws AsterixException {
         MetadataLock lock = feedPolicyLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     // CompactionPolicy
-    public void acquireCompactionPolicyReadLock(LockList locks, String dataverseName) {
+    public void acquireCompactionPolicyReadLock(LockList locks, String dataverseName) throws AsterixException {
         MetadataLock lock = compactionPolicyLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
-    public void acquireCompactionPolicyWriteLock(LockList locks, String dataverseName) {
-        MetadataLock lock = compactionPolicyLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
-        locks.add(IMetadataLock.Mode.WRITE, lock);
-    }
-
     // DataType
-    public void acquireDataTypeReadLock(LockList locks, String dataverseName) {
+    public void acquireDataTypeReadLock(LockList locks, String dataverseName) throws AsterixException {
         MetadataLock lock = dataTypeLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
-    public void acquireDataTypeWriteLock(LockList locks, String dataverseName) {
+    public void acquireDataTypeWriteLock(LockList locks, String dataverseName) throws AsterixException {
         MetadataLock lock = dataTypeLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     // Extensions
-    public void acquireExtensionReadLock(LockList locks, String dataverseName) {
+    public void acquireExtensionReadLock(LockList locks, String dataverseName) throws AsterixException {
         MetadataLock lock = extensionLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
-    public void acquireExtensionWriteLock(LockList locks, String dataverseName) {
+    public void acquireExtensionWriteLock(LockList locks, String dataverseName) throws AsterixException {
         MetadataLock lock = extensionLocks.computeIfAbsent(dataverseName, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
@@ -170,7 +160,7 @@
     public void createDatasetBegin(LockList locks, String dataverseName, String itemTypeDataverseName,
             String itemTypeFullyQualifiedName, String metaItemTypeDataverseName, String metaItemTypeFullyQualifiedName,
             String nodeGroupName, String compactionPolicyName, String datasetFullyQualifiedName,
-            boolean isDefaultCompactionPolicy) {
+            boolean isDefaultCompactionPolicy) throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         if (!dataverseName.equals(itemTypeDataverseName)) {
             acquireDataverseReadLock(locks, itemTypeDataverseName);
@@ -191,58 +181,66 @@
         acquireDatasetWriteLock(locks, datasetFullyQualifiedName);
     }
 
-    public void createIndexBegin(LockList locks, String dataverseName, String datasetFullyQualifiedName) {
+    public void createIndexBegin(LockList locks, String dataverseName, String datasetFullyQualifiedName)
+            throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         acquireDatasetCreateIndexLock(locks, datasetFullyQualifiedName);
     }
 
-    public void createTypeBegin(LockList locks, String dataverseName, String itemTypeFullyQualifiedName) {
+    public void createTypeBegin(LockList locks, String dataverseName, String itemTypeFullyQualifiedName)
+            throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         acquireDataTypeWriteLock(locks, itemTypeFullyQualifiedName);
     }
 
-    public void dropDatasetBegin(LockList locks, String dataverseName, String datasetFullyQualifiedName) {
+    public void dropDatasetBegin(LockList locks, String dataverseName, String datasetFullyQualifiedName)
+            throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         acquireDatasetWriteLock(locks, datasetFullyQualifiedName);
     }
 
-    public void dropIndexBegin(LockList locks, String dataverseName, String datasetFullyQualifiedName) {
+    public void dropIndexBegin(LockList locks, String dataverseName, String datasetFullyQualifiedName)
+            throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         acquireDatasetWriteLock(locks, datasetFullyQualifiedName);
     }
 
-    public void dropTypeBegin(LockList locks, String dataverseName, String dataTypeFullyQualifiedName) {
+    public void dropTypeBegin(LockList locks, String dataverseName, String dataTypeFullyQualifiedName)
+            throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         acquireDataTypeWriteLock(locks, dataTypeFullyQualifiedName);
     }
 
-    public void functionStatementBegin(LockList locks, String dataverseName, String functionFullyQualifiedName) {
+    public void functionStatementBegin(LockList locks, String dataverseName, String functionFullyQualifiedName)
+            throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         acquireFunctionWriteLock(locks, functionFullyQualifiedName);
     }
 
-    public void modifyDatasetBegin(LockList locks, String dataverseName, String datasetFullyQualifiedName) {
+    public void modifyDatasetBegin(LockList locks, String dataverseName, String datasetFullyQualifiedName)
+            throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         acquireDatasetModifyLock(locks, datasetFullyQualifiedName);
     }
 
-    public void insertDeleteUpsertBegin(LockList locks, String datasetFullyQualifiedName) {
+    public void insertDeleteUpsertBegin(LockList locks, String datasetFullyQualifiedName) throws AsterixException {
         acquireDataverseReadLock(locks, DatasetUtil.getDataverseFromFullyQualifiedName(datasetFullyQualifiedName));
         acquireDatasetModifyLock(locks, datasetFullyQualifiedName);
     }
 
-    public void dropFeedBegin(LockList locks, String dataverseName, String feedFullyQualifiedName) {
+    public void dropFeedBegin(LockList locks, String dataverseName, String feedFullyQualifiedName)
+            throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         acquireFeedWriteLock(locks, feedFullyQualifiedName);
     }
 
-    public void dropFeedPolicyBegin(LockList locks, String dataverseName, String policyName) {
+    public void dropFeedPolicyBegin(LockList locks, String dataverseName, String policyName) throws AsterixException {
         acquireFeedWriteLock(locks, policyName);
         acquireDataverseReadLock(locks, dataverseName);
     }
 
     public void startFeedBegin(LockList locks, String dataverseName, String feedName,
-            List<FeedConnection> feedConnections) {
+            List<FeedConnection> feedConnections) throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         acquireFeedReadLock(locks, feedName);
         for (FeedConnection feedConnection : feedConnections) {
@@ -252,43 +250,46 @@
         }
     }
 
-    public void stopFeedBegin(LockList locks, String dataverseName, String feedName) {
+    public void stopFeedBegin(LockList locks, String dataverseName, String feedName) throws AsterixException {
         // TODO: dataset lock?
         // Dataset locks are not required here since datasets are protected by the active event listener
         acquireDataverseReadLock(locks, dataverseName);
         acquireFeedReadLock(locks, feedName);
     }
 
-    public void createFeedBegin(LockList locks, String dataverseName, String feedFullyQualifiedName) {
+    public void createFeedBegin(LockList locks, String dataverseName, String feedFullyQualifiedName)
+            throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         acquireFeedWriteLock(locks, feedFullyQualifiedName);
     }
 
     public void connectFeedBegin(LockList locks, String dataverseName, String datasetFullyQualifiedName,
-            String feedFullyQualifiedName) {
+            String feedFullyQualifiedName) throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         acquireDatasetReadLock(locks, datasetFullyQualifiedName);
         acquireFeedReadLock(locks, feedFullyQualifiedName);
     }
 
-    public void createFeedPolicyBegin(LockList locks, String dataverseName, String policyName) {
+    public void createFeedPolicyBegin(LockList locks, String dataverseName, String policyName) throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         acquireFeedPolicyWriteLock(locks, policyName);
     }
 
     public void disconnectFeedBegin(LockList locks, String dataverseName, String datasetFullyQualifiedName,
-            String feedFullyQualifiedName) {
+            String feedFullyQualifiedName) throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         acquireDatasetReadLock(locks, datasetFullyQualifiedName);
         acquireFeedReadLock(locks, feedFullyQualifiedName);
     }
 
-    public void compactBegin(LockList locks, String dataverseName, String datasetFullyQualifiedName) {
+    public void compactBegin(LockList locks, String dataverseName, String datasetFullyQualifiedName)
+            throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         acquireDatasetReadLock(locks, datasetFullyQualifiedName);
     }
 
-    public void refreshDatasetBegin(LockList locks, String dataverseName, String datasetFullyQualifiedName) {
+    public void refreshDatasetBegin(LockList locks, String dataverseName, String datasetFullyQualifiedName)
+            throws AsterixException {
         acquireDataverseReadLock(locks, dataverseName);
         acquireExternalDatasetRefreshLock(locks, datasetFullyQualifiedName);
     }