Deadlock detection by owner

Issue-ID: MUSIC-502
Signed-off-by: Martella, Arthur <arthur.martella.1@att.com>
Change-Id: Iec20cfeec96d7031c691055ffba2f65c34854adf

8 files changed, 378 insertions, 24 deletions

diff --git a/src/main/java/org/onap/music/datastore/PreparedQueryObject.java b/src/main/java/org/onap/music/datastore/PreparedQueryObject.java
index d65096a7..fdac50be 100644
--- a/src/main/java/org/onap/music/datastore/PreparedQueryObject.java
+++ b/src/main/java/org/onap/music/datastore/PreparedQueryObject.java
@@ -89,7 +89,28 @@ public class PreparedQueryObject {
     }
 
     public String getOperation() {
-        return operation;
+        if (operation!=null) return operation;
+        if (query.length()==0) return null;
+        String queryStr = query.toString().toLowerCase();
+        String firstOp = null;
+        int firstOpChar = query.length();
+        if (queryStr.indexOf("insert")>-1 && queryStr.indexOf("insert")<firstOpChar) {
+            firstOp = "insert";
+            firstOpChar = queryStr.indexOf("insert");
+        }
+        if (queryStr.indexOf("update")>-1 && queryStr.indexOf("update")<firstOpChar) {
+            firstOp = "update";
+            firstOpChar = queryStr.indexOf("update");
+        }
+        if (queryStr.indexOf("delete")>-1 && queryStr.indexOf("delete")<firstOpChar) {
+            firstOp = "delete";
+            firstOpChar = queryStr.indexOf("delete");
+        }
+        if (queryStr.indexOf("select")>-1 && queryStr.indexOf("select")<firstOpChar) {
+            firstOp = "select";
+            firstOpChar = queryStr.indexOf("select");
+        }
+        return firstOp;
     }
 
     public void setOperation(String operation) {
diff --git a/src/main/java/org/onap/music/exceptions/MusicDeadlockException.java b/src/main/java/org/onap/music/exceptions/MusicDeadlockException.java
new file mode 100644
index 00000000..f5478e78
--- /dev/null
+++ b/src/main/java/org/onap/music/exceptions/MusicDeadlockException.java
@@ -0,0 +1,75 @@
+/*
+ * ============LICENSE_START==========================================
+ * org.onap.music
+ * ===================================================================
+ *  Copyright (c) 2019 AT&T Intellectual Property
+ * ===================================================================
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ * ============LICENSE_END=============================================
+ * ====================================================================
+ */
+package org.onap.music.exceptions;
+
+public class MusicDeadlockException extends MusicLockingException {
+
+	public String owner = null;
+	public String keyspace = null;
+	public String table = null;
+	public String key = null;
+	
+	public MusicDeadlockException() {
+		super();
+	}
+
+	public MusicDeadlockException(String message) {
+		super(message);
+	}
+	
+	public MusicDeadlockException(Throwable cause) {
+		super(cause);
+	}
+
+	public MusicDeadlockException(String message, Throwable cause) {
+		super(message, cause);
+	}
+
+	public MusicDeadlockException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+		super(message, cause, enableSuppression, writableStackTrace);
+	}
+
+	public void setValues(String owner, String keyspace, String table, String key) {
+		this.owner = owner;
+		this.keyspace = keyspace;
+		this.table = table;
+		this.key = key;
+	}
+
+	public String getOwner() {
+		return owner;
+	}
+
+	public String getKeyspace() {
+		return keyspace;
+	}
+
+	public String getTable() {
+		return table;
+	}
+
+	public String getKey() {
+		return key;
+	}
+
+	
+}
diff --git a/src/main/java/org/onap/music/lockingservice/cassandra/CassaLockStore.java b/src/main/java/org/onap/music/lockingservice/cassandra/CassaLockStore.java
index 06a087ab..f57ecb6e 100644
--- a/src/main/java/org/onap/music/lockingservice/cassandra/CassaLockStore.java
+++ b/src/main/java/org/onap/music/lockingservice/cassandra/CassaLockStore.java
@@ -24,6 +24,7 @@
 package org.onap.music.lockingservice.cassandra;
 
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
 
 import org.onap.music.datastore.MusicDataStore;
@@ -32,6 +33,8 @@ import org.onap.music.eelf.logging.EELFLoggerDelegate;
 import org.onap.music.exceptions.MusicLockingException;
 import org.onap.music.exceptions.MusicQueryException;
 import org.onap.music.exceptions.MusicServiceException;
+import org.onap.music.main.DeadlockDetectionUtil;
+import org.onap.music.main.DeadlockDetectionUtil.OwnershipType;
 import org.onap.music.main.MusicUtil;
 
 import com.datastax.driver.core.ResultSet;
@@ -60,12 +63,15 @@ public class CassaLockStore {
         private String createTime;
         private String acquireTime;
         private LockType locktype;
-        public LockObject(boolean isLockOwner, String lockRef, String createTime, String acquireTime, LockType locktype) {
+        // Owner is the self-declared client which "owns" this row. It is used for deadlock detection.  It is not (directly) related to isLockOwner.
+        private String owner;
+        public LockObject(boolean isLockOwner, String lockRef, String createTime, String acquireTime, LockType locktype, String owner) {
             this.setIsLockOwner(isLockOwner);
             this.setLockRef(lockRef);
             this.setAcquireTime(acquireTime);
             this.setCreateTime(createTime);
             this.setLocktype(locktype);
+            this.setOwner(owner);
         }
         public boolean getIsLockOwner() {
             return isLockOwner;
@@ -97,6 +103,12 @@ public class CassaLockStore {
         public void setLocktype(LockType locktype) {
             this.locktype = locktype;
         }
+        public String getOwner() {
+            return owner;
+        }
+        public void setOwner(String owner) {
+            this.owner = owner;
+        }
     }
     
     /**
@@ -114,7 +126,7 @@ public class CassaLockStore {
         table = table_prepend_name+table;
         String tabQuery = "CREATE TABLE IF NOT EXISTS "+keyspace+"."+table
                 + " ( key text, lockReference bigint, createTime text, acquireTime text, guard bigint static, "
-                + "writeLock boolean, PRIMARY KEY ((key), lockReference) ) "
+                + "writeLock boolean, owner text, PRIMARY KEY ((key), lockReference) ) "
                 + "WITH CLUSTERING ORDER BY (lockReference ASC);";
         PreparedQueryObject queryObject = new PreparedQueryObject();
         
@@ -129,22 +141,22 @@ public class CassaLockStore {
      * @param keyspace of the locks.
      * @param table of the locks.
      * @param lockName is the primary key of the lock table
+     * @param lockType is the type of lock (read/write)
+     * @param owner is the owner of the lock (optional, for deadlock detection)
      * @return the UUID lock reference.
      * @throws MusicServiceException
      * @throws MusicQueryException
      */
-    public String genLockRefandEnQueue(String keyspace, String table, String lockName, LockType locktype) throws MusicServiceException, MusicQueryException, MusicLockingException {
-        return genLockRefandEnQueue(keyspace, table, lockName, locktype, 0);
+    public String genLockRefandEnQueue(String keyspace, String table, String lockName, LockType locktype, String owner) throws MusicServiceException, MusicQueryException, MusicLockingException {
+        return genLockRefandEnQueue(keyspace, table, lockName, locktype, owner, 0);
     }
     
-    private String genLockRefandEnQueue(String keyspace, String table, String lockName, LockType locktype, int count) throws MusicServiceException, MusicQueryException, MusicLockingException {
+    private String genLockRefandEnQueue(String keyspace, String table, String lockName, LockType locktype, String owner, int count) throws MusicServiceException, MusicQueryException, MusicLockingException {
         logger.info(EELFLoggerDelegate.applicationLogger,
                 "Create " + locktype + " lock reference for " +  keyspace + "." + table + "." + lockName);
         String lockTable ="";
         lockTable = table_prepend_name+table;
     
-
-
         PreparedQueryObject queryObject = new PreparedQueryObject();
         String selectQuery = "SELECT guard FROM " + keyspace + "." + lockTable + " WHERE key=?;";
 
@@ -170,7 +182,7 @@ public class CassaLockStore {
                 " UPDATE " + keyspace + "." + lockTable +
                 " SET guard=? WHERE key=? IF guard = " + (prevGuard == 0 ? "NULL" : "?") +";" +
                 " INSERT INTO " + keyspace + "." + lockTable +
-                "(key, lockReference, createTime, acquireTime, writeLock) VALUES (?,?,?,?,?) IF NOT EXISTS; APPLY BATCH;";
+                "(key, lockReference, createTime, acquireTime, writeLock, owner) VALUES (?,?,?,?,?,?) IF NOT EXISTS; APPLY BATCH;";
 
         queryObject.addValue(lockRef);
         queryObject.addValue(lockName);
@@ -182,6 +194,7 @@ public class CassaLockStore {
         queryObject.addValue(String.valueOf(lockEpochMillis));
         queryObject.addValue("0");
         queryObject.addValue(locktype==LockType.WRITE ? true : false );
+        queryObject.addValue(owner);
         queryObject.appendQueryString(insQuery);
         boolean pResult = dsHandle.executePut(queryObject, "critical");
         if (!pResult) {// couldn't create lock ref, retry
@@ -190,14 +203,12 @@ public class CassaLockStore {
                 logger.warn(EELFLoggerDelegate.applicationLogger, "Unable to create lock reference");
                 throw new MusicLockingException("Unable to create lock reference");
             }
-            return genLockRefandEnQueue(keyspace, table, lockName, locktype, count);
+            return genLockRefandEnQueue(keyspace, table, lockName, locktype, owner, count);
         }
         return "$" + keyspace + "." + table + "." + lockName + "$" + String.valueOf(lockRef);
     }
-    
-    
 
-    /**
+	/**
      * Returns a result set containing the list of clients waiting for a particular lock
      * 
      * @param keyspace
@@ -269,14 +280,15 @@ public class CassaLockStore {
         ResultSet results = dsHandle.executeOneConsistencyGet(queryObject);
         Row row = results.one();
         if (row == null || row.isNull("lockReference")) {
-            return new LockObject(false, null, null, null, null);
+            return new LockObject(false, null, null, null, null, null);
         }
         String lockReference = "" + row.getLong("lockReference");
         String createTime = row.getString("createTime");
         String acquireTime = row.getString("acquireTime");
         LockType locktype = row.isNull("writeLock") || row.getBool("writeLock") ? LockType.WRITE : LockType.READ;
+        String owner = row.getString("owner");
 
-        return new LockObject(true, lockReference, createTime, acquireTime, locktype);
+        return new LockObject(true, lockReference, createTime, acquireTime, locktype, owner);
     }
 
     public List<String> getCurrentLockHolders(String keyspace, String table, String key)
@@ -394,8 +406,9 @@ public class CassaLockStore {
         String acquireTime = row.getString("acquireTime");
         LockType locktype = row.isNull("writeLock") || row.getBool("writeLock") ? LockType.WRITE : LockType.READ;
         boolean isLockOwner = isLockOwner(keyspace, table, key, lockRef);
+        String owner = row.getString("owner");
 
-        return new LockObject(isLockOwner, lockReference, createTime, acquireTime, locktype);
+        return new LockObject(isLockOwner, lockReference, createTime, acquireTime, locktype, owner);
     }
 
 
@@ -454,4 +467,45 @@ public class CassaLockStore {
         dsHandle.getSession().execute(updateQuery);
     }  
 
+    public boolean checkForDeadlock(String keyspace, String table, String lockName, LockType locktype, String owner, boolean forAcquire) throws MusicServiceException, MusicQueryException {
+        if (locktype.equals(LockType.READ)) return false;
+        if (owner==null || owner.length()==0) return false;
+
+        String lockTable = table_prepend_name + table;
+        PreparedQueryObject queryObject = new PreparedQueryObject();
+        queryObject.appendQueryString("SELECT key, acquiretime, owner FROM " + keyspace + "." + lockTable);
+        queryObject.appendQueryString(" WHERE writelock = True ALLOW FILTERING");
+
+        DeadlockDetectionUtil ddu = new DeadlockDetectionUtil();
+
+        ResultSet rs = dsHandle.executeQuorumConsistencyGet(queryObject);
+        logger.debug("rs has " + rs.getAvailableWithoutFetching() + (rs.isFullyFetched()?"":" (or more!)") );
+        Iterator<Row> it = rs.iterator();
+        while (it.hasNext()) {
+            Row row = it.next();
+            logger.debug("key = " + row.getString("key") + ", time = " + row.getString("acquiretime") + ", owner = " + row.getString("owner") );
+            ddu.setExisting(row.getString("key"), row.getString("owner"), ("0".equals(row.getString("acquiretime")))?OwnershipType.CREATED:OwnershipType.ACQUIRED);
+        }
+        boolean deadlock = ddu.checkForDeadlock(lockName, owner, forAcquire?OwnershipType.ACQUIRED:OwnershipType.CREATED);
+        if (deadlock) logger.warn("Deadlock detected when " + owner + " tried to create lock on " + keyspace + "." + lockTable + "." + lockName);
+        return deadlock;
+    }
+
+    public List<String> getAllLocksForOwner(String ownerId, String keyspace, String table) throws MusicServiceException, MusicQueryException {
+        List<String> toRet = new ArrayList<String>();
+        String lockTable = table_prepend_name + table;
+        PreparedQueryObject queryObject = new PreparedQueryObject();
+        queryObject.appendQueryString("SELECT key, lockreference FROM " + keyspace + "." + lockTable);
+        queryObject.appendQueryString(" WHERE owner = '" + ownerId + "' ALLOW FILTERING");
+
+        ResultSet rs = dsHandle.executeQuorumConsistencyGet(queryObject);
+        Iterator<Row> it = rs.iterator();
+        while (it.hasNext()) {
+            Row row = it.next();
+            toRet.add(row.getString("key") + "$" + row.getLong("lockreference"));
+        }
+        return toRet;
+    }
+
+
 }
diff --git a/src/main/java/org/onap/music/main/DeadlockDetectionUtil.java b/src/main/java/org/onap/music/main/DeadlockDetectionUtil.java
new file mode 100644
index 00000000..4c9a74b7
--- /dev/null
+++ b/src/main/java/org/onap/music/main/DeadlockDetectionUtil.java
@@ -0,0 +1,148 @@
+/*
+ * ============LICENSE_START==========================================
+ * org.onap.music
+ * ===================================================================
+ *  Copyright (c) 2019 AT&T Intellectual Property
+ * ===================================================================
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ * ============LICENSE_END=============================================
+ * ====================================================================
+ */
+package org.onap.music.main;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class DeadlockDetectionUtil {
+	private Map<String, Node> nodeList = null;
+	public enum OwnershipType {NONE, CREATED, ACQUIRED};
+
+	private class Node implements Comparable<Node> {
+		private String id;
+		private List<Node> links;
+		private boolean visited = false;
+		private boolean onStack = false;
+		
+		@Override
+		public String toString() {
+			StringBuffer sb = new StringBuffer();
+			for (Node link : links) sb.append(link.id);
+			return "Node [id=" + id + ", links=" + sb.toString() + ", visited=" + visited + ", onStack=" + onStack + "]";
+		}
+
+		public Node(String id) {
+			super();
+			this.id = id;
+			this.links = new ArrayList<Node>();
+		}
+
+		public List<Node> getLinks() {
+			return links;
+		}
+
+		public void addLink(Node link) {
+			this.links.add(link);
+		}
+		
+		public void removeLink(Node link) {
+			this.links.remove(link);
+		}
+
+		public boolean isVisited() {
+			return visited;
+		}
+
+		public boolean isOnStack() {
+			return onStack;
+		}
+
+		public void setVisited(boolean visited) {
+			this.visited = visited;
+		}
+
+		public void setOnStack(boolean onStack) {
+			this.onStack = onStack;
+		}
+
+		@Override
+		public int compareTo(Node arg0) {
+			return id.compareTo(arg0.id);
+		}
+	}
+	
+	public DeadlockDetectionUtil() {
+		this.nodeList = new HashMap<String, Node>();
+	}
+
+	public void listAllNodes() {
+		System.out.println("In DeadlockDetectionUtil: ");
+		for (String key : nodeList.keySet()) {
+			System.out.println("    " + key + " : " + nodeList.get(key));
+		}
+	}
+
+	public boolean checkForDeadlock(String resource, String owner, OwnershipType operation) {
+		setExisting(resource, owner, operation);
+
+		Node currentNode = null;
+		if (operation.equals(OwnershipType.ACQUIRED)) {
+			currentNode = nodeList.get("r" + resource);
+		} else if (operation.equals(OwnershipType.CREATED)) {
+			currentNode = nodeList.get("o" + owner);
+		}
+
+		boolean cycle = findCycle(currentNode);
+		return cycle;
+	}
+
+	private boolean findCycle(Node currentNode) {
+		if (currentNode==null) return false;
+		if (currentNode.isOnStack()) return true;
+		if (currentNode.isVisited()) return false;
+		currentNode.setOnStack(true);
+		currentNode.setVisited(true);
+		for (Node childNode : currentNode.getLinks()) {
+			if (findCycle(childNode)) return true;
+		}
+		currentNode.setOnStack(false);
+		return false;
+	}
+
+	public void setExisting(String resource, String owner, OwnershipType operation) {
+		String resourceKey = "r" + resource;
+		Node resourceNode = nodeList.get(resourceKey);
+		if (resourceNode==null) {
+			resourceNode = new Node(resourceKey);
+			nodeList.put(resourceKey, resourceNode);
+		}
+		
+		String ownerKey = "o" + owner;
+		Node ownerNode = nodeList.get(ownerKey);
+		if (ownerNode==null) {
+			ownerNode = new Node(ownerKey);
+			nodeList.put(ownerKey, ownerNode);
+		}
+		
+		if (operation.equals(OwnershipType.ACQUIRED)) {
+			resourceNode.addLink(ownerNode);
+			ownerNode.removeLink(resourceNode);
+		} else if (operation.equals(OwnershipType.CREATED)) {
+			ownerNode.addLink(resourceNode);
+			resourceNode.removeLink(ownerNode);
+		}
+	}
+
+}
diff --git a/src/main/java/org/onap/music/main/MusicCore.java b/src/main/java/org/onap/music/main/MusicCore.java
index da94e1a6..e889e180 100644
--- a/src/main/java/org/onap/music/main/MusicCore.java
+++ b/src/main/java/org/onap/music/main/MusicCore.java
@@ -89,6 +89,10 @@ public class MusicCore {
         return musicCore.createLockReference(fullyQualifiedKey, locktype);
     }
 
+    public static String createLockReference(String fullyQualifiedKey, LockType locktype, String owner) throws MusicLockingException {
+        return musicCore.createLockReference(fullyQualifiedKey, locktype, owner);
+    }
+
     public static ResultType createTable(String keyspace, String table, PreparedQueryObject tableQueryObject,
             String consistency) throws MusicServiceException {
         return musicCore.createTable(keyspace, table, tableQueryObject, consistency);
@@ -193,6 +197,10 @@ public class MusicCore {
         return musicCore.releaseLock(lockId, voluntaryRelease);
     }
     
+    public static List<String> releaseAllLocksForOwner(String ownerId, String keyspace, String table) throws MusicLockingException, MusicServiceException, MusicQueryException {
+        return musicCore.releaseAllLocksForOwner(ownerId, keyspace, table);
+	}
+
     //Added changes for orm implementation.
     
     public static ResultType createKeyspace(JsonKeySpace jsonKeySpaceObject, String consistencyInfo) 
@@ -245,6 +253,4 @@ public class MusicCore {
         return musicCore.deleteFromTable(jsonDeleteObj,rowParams);
     }
 
-
-
 }
diff --git a/src/main/java/org/onap/music/rest/RestMusicLocksAPI.java b/src/main/java/org/onap/music/rest/RestMusicLocksAPI.java
index 35f03e60..c08fc5d8 100644
--- a/src/main/java/org/onap/music/rest/RestMusicLocksAPI.java
+++ b/src/main/java/org/onap/music/rest/RestMusicLocksAPI.java
@@ -116,6 +116,7 @@ public class RestMusicLocksAPI {
             @ApiParam(value = "Authorization", required = true) @HeaderParam(MusicUtil.AUTHORIZATION) String authorization,
             @ApiParam(value = "AID", required = false, hidden = true) @HeaderParam("aid") String aid,
             JsonLock lockObject,
+            @ApiParam(value = "Lock Owner", required = false) @HeaderParam("owner") String owner,
             @ApiParam(value = "Application namespace",
                             required = false, hidden = true) @HeaderParam("ns") String ns) throws Exception{
         try {
@@ -136,7 +137,7 @@ public class RestMusicLocksAPI {
             }
             String lockId;
             try {
-                lockId= MusicCore.createLockReference(lockName, locktype);
+                lockId= MusicCore.createLockReference(lockName, locktype, owner);
             } catch (MusicLockingException e) {
                 return response.status(Status.BAD_REQUEST).entity(new JsonResponse(ResultType.FAILURE).setError(e.getMessage()).toMap()).build();
             }
diff --git a/src/main/java/org/onap/music/service/MusicCoreService.java b/src/main/java/org/onap/music/service/MusicCoreService.java
index c96d2b30..1ecb2ee1 100644
--- a/src/main/java/org/onap/music/service/MusicCoreService.java
+++ b/src/main/java/org/onap/music/service/MusicCoreService.java
@@ -98,6 +98,14 @@ public interface MusicCoreService {
      */
     public String createLockReference(String fullyQualifiedKey, LockType locktype) throws MusicLockingException;
     
+    /**
+     * Create a lock ref in the music lock store
+     * @param fullyQualifiedKey the key to create a lock on
+     * @param locktype the type of lock create, see {@link LockType}
+     * @param owner the owner of the lock, for deadlock prevention
+     */
+    public String createLockReference(String fullyQualifiedKey, LockType locktype, String owner) throws MusicLockingException;
+
     public ReturnType acquireLockWithLease(String key, String lockReference, long leasePeriod)
         throws MusicLockingException, MusicQueryException, MusicServiceException; // key,lock id,time
 
@@ -143,7 +151,9 @@ public interface MusicCoreService {
     public Map<String, Object> validateLock(String lockName);
 
     public MusicLockState releaseLock(String lockId, boolean voluntaryRelease) throws MusicLockingException;
-    
+
+	public List<String> releaseAllLocksForOwner(String ownerId, String keyspace, String table) throws MusicLockingException, MusicServiceException, MusicQueryException;
+
     
     //Methods added for orm
     
diff --git a/src/main/java/org/onap/music/service/impl/MusicCassaCore.java b/src/main/java/org/onap/music/service/impl/MusicCassaCore.java
index 5c02d34d..0c30cc74 100644
--- a/src/main/java/org/onap/music/service/impl/MusicCassaCore.java
+++ b/src/main/java/org/onap/music/service/impl/MusicCassaCore.java
@@ -47,6 +47,7 @@ import org.onap.music.eelf.logging.EELFLoggerDelegate;
 import org.onap.music.eelf.logging.format.AppMessages;
 import org.onap.music.eelf.logging.format.ErrorSeverity;
 import org.onap.music.eelf.logging.format.ErrorTypes;
+import org.onap.music.exceptions.MusicDeadlockException;
 import org.onap.music.exceptions.MusicLockingException;
 import org.onap.music.exceptions.MusicQueryException;
 import org.onap.music.exceptions.MusicServiceException;
@@ -116,6 +117,10 @@ public class MusicCassaCore implements MusicCoreService {
     }
 
     public String createLockReference(String fullyQualifiedKey, LockType locktype) throws MusicLockingException {
+        return createLockReference(fullyQualifiedKey, locktype, null);
+    }
+
+    public String createLockReference(String fullyQualifiedKey, LockType locktype, String owner) throws MusicLockingException {
         String[] splitString = fullyQualifiedKey.split("\\.");
         String keyspace = splitString[0];
         String table = splitString[1];
@@ -124,15 +129,30 @@ public class MusicCassaCore implements MusicCoreService {
         logger.info(EELFLoggerDelegate.applicationLogger,"Creating lock reference for lock name:" + lockName);
         long start = System.currentTimeMillis();
         String lockReference = null;
+
+        try {
+            boolean deadlock = getLockingServiceHandle().checkForDeadlock(keyspace, table, lockName, locktype, owner, false);
+            if (deadlock) {
+                MusicDeadlockException e = new MusicDeadlockException("Deadlock detected when " + owner + " tried to create lock on " + keyspace + "." + table + "." + lockName);
+                e.setValues(owner, keyspace, table, lockName);
+                throw e;
+            }
+        } catch (MusicDeadlockException e) {
+            //just threw this, no need to wrap it
+            throw e;
+        } catch (MusicServiceException | MusicQueryException e) {
+            logger.error(EELFLoggerDelegate.applicationLogger, e);
+            throw new MusicLockingException("Unable to check for deadlock. " + e.getMessage(), e);
+        }
         
         try {
-            lockReference = "" + getLockingServiceHandle().genLockRefandEnQueue(keyspace, table, lockName, locktype);
+            lockReference = "" + getLockingServiceHandle().genLockRefandEnQueue(keyspace, table, lockName, locktype, owner);
         } catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
             logger.error(EELFLoggerDelegate.applicationLogger, e);
-            throw new MusicLockingException("Unable to create lock reference. " + e.getMessage());
+            throw new MusicLockingException("Unable to create lock reference. " + e.getMessage(), e);
         } catch (Exception e) {
             logger.error(EELFLoggerDelegate.applicationLogger, e);
-            throw new MusicLockingException("Unable to create lock reference. " + e.getMessage());
+            throw new MusicLockingException("Unable to create lock reference. " + e.getMessage(), e);
         }
         long end = System.currentTimeMillis();
         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to create lock reference:" + (end - start) + " ms");
@@ -185,6 +205,12 @@ public class MusicCassaCore implements MusicCoreService {
             return new ReturnType(ResultType.FAILURE, lockId + " is not a lock holder");//not top of the lock store q
         }
         
+        if (getLockingServiceHandle().checkForDeadlock(keyspace, table, primaryKeyValue, lockInfo.getLocktype(), lockInfo.getOwner(), true)) {
+            MusicDeadlockException e = new MusicDeadlockException("Deadlock detected when " + lockInfo.getOwner()  + " tried to create lock on " + keyspace + "." + table + "." + primaryKeyValue);
+            e.setValues(lockInfo.getOwner(), keyspace, table, primaryKeyValue);
+            throw e;
+        }
+
         //check to see if the value of the key has to be synced in case there was a forceful release
         String syncTable = keyspace+".unsyncedKeys_"+table;
         String query = "select * from "+syncTable+" where key='"+localFullyQualifiedKey+"';";
@@ -319,7 +345,7 @@ public class MusicCassaCore implements MusicCoreService {
         String table = splitString[1];
         String primaryKeyValue = splitString[2];
         try {
-        	return getLockingServiceHandle().getCurrentLockHolders(keyspace, table, primaryKeyValue);
+            return getLockingServiceHandle().getCurrentLockHolders(keyspace, table, primaryKeyValue);
         } catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKINGERROR+fullyQualifiedKey ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
         }
@@ -421,6 +447,19 @@ public class MusicCassaCore implements MusicCoreService {
         return destroyLockRef(fullyQualifiedKey, lockReference);
     }
 
+    @Override
+    public List<String> releaseAllLocksForOwner(String ownerId, String keyspace, String table) throws MusicLockingException, MusicServiceException, MusicQueryException {
+//        System.out.println("IN RELEASEALLLOCKSFOROWNER, ");
+
+        List<String> lockIds = getLockingServiceHandle().getAllLocksForOwner(ownerId, keyspace, table);
+        for (String lockId : lockIds) {
+//            System.out.println(" LOCKID = " + lockId);
+            //return "$" + keyspace + "." + table + "." + lockName + "$" + String.valueOf(lockRef);
+            releaseLock("$" + keyspace + "." + table + "." + lockId, true);
+        }
+        return lockIds;
+    }
+
     /**
      *
      * @param lockName