Push variuos changes

- Spring Boot
- Cadi
- Cassandra Locking

Change-Id: Ie9882f81f0ca141bdb7862cdabf978481fcd7c4a
Issue-ID: MUSIC-296,MUSIC-272
Signed-off-by: Nelson,Thomas(tn1381)(arthurdent3) <tn1381@att.com>

1 files changed, 937 insertions, 0 deletions

diff --git a/src/main/java/org/onap/music/service/impl/MusicZKCore.java b/src/main/java/org/onap/music/service/impl/MusicZKCore.java
new file mode 100644
index 00000000..1662b20c
--- /dev/null
+++ b/src/main/java/org/onap/music/service/impl/MusicZKCore.java
@@ -0,0 +1,937 @@
+/*
+ * ============LICENSE_START==========================================
+ * org.onap.music
+ * ===================================================================
+ *  Copyright (c) 2017 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.service.impl;
+
+
+import java.io.StringWriter;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import javax.ws.rs.core.MediaType;
+
+import org.apache.commons.jcs.access.CacheAccess;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.NoNodeException;
+import org.onap.music.datastore.MusicDataStore;
+import org.onap.music.datastore.PreparedQueryObject;
+import org.onap.music.datastore.jsonobjects.JsonKeySpace;
+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.MusicLockingException;
+import org.onap.music.exceptions.MusicQueryException;
+import org.onap.music.exceptions.MusicServiceException;
+import org.onap.music.lockingservice.cassandra.MusicLockState;
+import org.onap.music.lockingservice.cassandra.MusicLockState.LockStatus;
+import org.onap.music.lockingservice.zookeeper.MusicLockingService;
+import org.onap.music.main.MusicUtil;
+import org.onap.music.main.ResultType;
+import org.onap.music.main.ReturnType;
+import org.onap.music.service.MusicCoreService;
+import org.onap.music.datastore.*;
+
+import com.datastax.driver.core.ColumnDefinitions;
+import com.datastax.driver.core.ColumnDefinitions.Definition;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.TableMetadata;
+import com.sun.jersey.api.client.Client;
+import com.sun.jersey.api.client.ClientResponse;
+import com.sun.jersey.api.client.WebResource;
+
+/**
+ * This class .....
+ *
+ *
+ */
+public class MusicZKCore implements MusicCoreService {
+
+    public static MusicLockingService mLockHandle = null;
+    private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(MusicZKCore.class);
+    private static MusicZKCore musicZKCoreInstance = null;
+    
+    private MusicZKCore() {
+        
+    }
+    public static MusicZKCore getInstance() {
+        
+        if(musicZKCoreInstance == null) {
+            musicZKCoreInstance = new MusicZKCore();
+        }
+        return musicZKCoreInstance;
+    }
+    
+
+
+
+
+    public static MusicLockingService getLockingServiceHandle() throws MusicLockingException {
+        logger.info(EELFLoggerDelegate.applicationLogger,"Acquiring lock store handle");
+        long start = System.currentTimeMillis();
+
+        if (mLockHandle == null) {
+            try {
+                mLockHandle = new MusicLockingService();
+            } catch (Exception e) {
+                logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKHANDLE,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+                throw new MusicLockingException("Failed to aquire Locl store handle " + e);
+            }
+        }
+        long end = System.currentTimeMillis();
+        logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to acquire lock store handle:" + (end - start) + " ms");
+        return mLockHandle;
+    }
+
+
+
+    public  String createLockReference(String lockName) {
+        logger.info(EELFLoggerDelegate.applicationLogger,"Creating lock reference for lock name:" + lockName);
+        long start = System.currentTimeMillis();
+        String lockId = null;
+        try {
+            lockId = getLockingServiceHandle().createLockId("/" + lockName);
+        } catch (MusicLockingException e) {
+            logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.CREATELOCK+lockName,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+
+        }
+        long end = System.currentTimeMillis();
+        logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to create lock reference:" + (end - start) + " ms");
+        return lockId;
+    }
+
+    /**
+     *
+     * @param key
+     * @return
+     */
+    public static boolean isTableOrKeySpaceLock(String key) {
+        String[] splitString = key.split("\\.");
+        if (splitString.length > 2)
+            return false;
+        else
+            return true;
+    }
+
+    /**
+     *
+     * @param key
+     * @return
+     */
+    public static MusicLockState getMusicLockState(String key) {
+        long start = System.currentTimeMillis();
+        try {
+            String[] splitString = key.split("\\.");
+            String keyspaceName = splitString[0];
+            String tableName = splitString[1];
+            String primaryKey = splitString[2];
+            MusicLockState mls;
+            String lockName = keyspaceName + "." + tableName + "." + primaryKey;
+            mls = getLockingServiceHandle().getLockState(lockName);
+            long end = System.currentTimeMillis();
+            logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to get lock state:" + (end - start) + " ms");
+            return mls;
+        } catch (NullPointerException | MusicLockingException e) {
+            logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.INVALIDLOCK,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+        }
+        return null;
+    }
+
+    public  ReturnType acquireLockWithLease(String key, String lockId, long leasePeriod) {
+        try {
+            long start = System.currentTimeMillis();
+            /* check if the current lock has exceeded its lease and if yes, release that lock */
+            MusicLockState mls = getMusicLockState(key);
+            if (mls != null) {
+                if (mls.getLockStatus().equals(LockStatus.LOCKED)) {
+                    logger.info(EELFLoggerDelegate.applicationLogger,"The current lock holder for " + key + " is " + mls.getLockHolder()
+                                    + ". Checking if it has exceeded lease");
+                    long currentLockPeriod = System.currentTimeMillis() - mls.getLeaseStartTime();
+                    long currentLeasePeriod = mls.getLeasePeriod();
+                    if (currentLockPeriod > currentLeasePeriod) {
+                        logger.info(EELFLoggerDelegate.applicationLogger,"Lock period " + currentLockPeriod
+                                        + " has exceeded lease period " + currentLeasePeriod);
+                        boolean voluntaryRelease = false;
+                        String currentLockHolder = mls.getLockHolder();
+                        mls = releaseLock(currentLockHolder, voluntaryRelease);
+                    }
+                }
+            } else
+                logger.error(EELFLoggerDelegate.errorLogger,key, AppMessages.INVALIDLOCK,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+
+            /*
+             * call the traditional acquire lock now and if the result returned is true, set the
+             * begin time-stamp and lease period
+             */
+            if (acquireLock(key, lockId).getResult() == ResultType.SUCCESS) {
+                mls = getMusicLockState(key);// get latest state
+                if ( mls == null ) {
+                    logger.info(EELFLoggerDelegate.applicationLogger,"Music Lock State is null");
+                    return new ReturnType(ResultType.FAILURE, "Could not acquire lock, Lock State is null");
+                }
+                if (mls.getLeaseStartTime() == -1) {// set it again only if it is not set already
+                    mls.setLeaseStartTime(System.currentTimeMillis());
+                    mls.setLeasePeriod(leasePeriod);
+                    getLockingServiceHandle().setLockState(key, mls);
+                }
+                long end = System.currentTimeMillis();
+                logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to acquire leased lock:" + (end - start) + " ms");
+                return new ReturnType(ResultType.SUCCESS, "Accquired lock");
+            } else {
+                long end = System.currentTimeMillis();
+                logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to fail to acquire leased lock:" + (end - start) + " ms");
+                return new ReturnType(ResultType.FAILURE, "Could not acquire lock");
+            }
+        } catch (Exception e) {
+            StringWriter sw = new StringWriter();
+               logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), "[ERR506E] Failed to aquire lock ",ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+
+            String exceptionAsString = sw.toString();
+            return new ReturnType(ResultType.FAILURE,
+                            "Exception thrown in acquireLockWithLease:\n" + exceptionAsString);
+        }
+    }
+
+    public  ReturnType acquireLock(String key, String lockId) throws MusicLockingException {
+        /*
+         * first check if I am on top. Since ids are not reusable there is no need to check
+         * lockStatus If the status is unlocked, then the above call will automatically return
+         * false.
+         */
+        Boolean result = false;
+        try {
+            result = getLockingServiceHandle().isMyTurn(lockId);
+        } catch (MusicLockingException e2) {
+            logger.error(EELFLoggerDelegate.errorLogger,AppMessages.INVALIDLOCK + lockId + " " + e2);
+            throw new MusicLockingException();
+        }
+        if (!result) {
+            logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: Not your turn, someone else has the lock");
+            try {
+                if (!getLockingServiceHandle().lockIdExists(lockId)) {
+                    logger.info(EELFLoggerDelegate.applicationLogger, "In acquire lock: this lockId doesn't exist");
+                    return new ReturnType(ResultType.FAILURE, "Lockid doesn't exist");
+                }
+            } catch (MusicLockingException e) {
+                logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.INVALIDLOCK+lockId,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+                 throw new MusicLockingException();
+            }
+            logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: returning failure");
+            return new ReturnType(ResultType.FAILURE, "Not your turn, someone else has the lock");
+        }
+
+
+        // this is for backward compatibility where locks could also be acquired on just
+        // keyspaces or tables.
+        if (isTableOrKeySpaceLock(key)) {
+            logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: A table or keyspace lock so no need to perform sync...so returning true");
+            return new ReturnType(ResultType.SUCCESS, "A table or keyspace lock so no need to perform sync...so returning true");
+        }
+
+        // read the lock name corresponding to the key and if the status is locked or being locked,
+        // then return false
+        MusicLockState currentMls = null;
+        MusicLockState newMls = null;
+        try {
+            currentMls = getMusicLockState(key);
+            String currentLockHolder = null;
+            if(currentMls != null) { currentLockHolder = currentMls.getLockHolder(); };
+            if (lockId.equals(currentLockHolder)) {
+                logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: You already have the lock!");
+                return new ReturnType(ResultType.SUCCESS, "You already have the lock!");
+            }
+        } catch (NullPointerException e) {
+            logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.INVALIDLOCK+lockId,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+        }
+
+        // change status to "being locked". This state transition is necessary to ensure syncing
+        // before granting the lock
+        String lockHolder = null;
+        boolean needToSyncQuorum = false;
+        if (currentMls != null)
+            needToSyncQuorum = currentMls.isNeedToSyncQuorum();
+
+
+        newMls = new MusicLockState(MusicLockState.LockStatus.BEING_LOCKED, lockHolder,
+                        needToSyncQuorum);
+        try {
+            getLockingServiceHandle().setLockState(key, newMls);
+        } catch (MusicLockingException e1) {
+            logger.error(EELFLoggerDelegate.errorLogger,e1.getMessage(), AppMessages.LOCKSTATE+key,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+        }
+        logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: Set lock state to being_locked");
+
+        // do syncing if this was a forced lock release
+        if (needToSyncQuorum) {
+            logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: Since there was a forcible release, need to sync quorum!");
+            try {
+              syncQuorum(key);
+            } catch (Exception e) {
+              logger.error(EELFLoggerDelegate.errorLogger,"Failed to set Lock state " + e);
+            }
+        }
+
+        // change status to locked
+        lockHolder = lockId;
+        needToSyncQuorum = false;
+        newMls = new MusicLockState(MusicLockState.LockStatus.LOCKED, lockHolder, needToSyncQuorum);
+        try {
+            getLockingServiceHandle().setLockState(key, newMls);
+        } catch (MusicLockingException e) {
+            logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKSTATE+key,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+        }
+        logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: Set lock state to locked and assigned current lock ref "
+                        + lockId + " as holder");
+
+        return new ReturnType(result?ResultType.SUCCESS:ResultType.FAILURE, "Set lock state to locked and assigned a lock holder");
+    }
+
+
+
+    /**
+     *
+     * @param keyspaceName
+     * @param kspObject
+     * @return
+     * @throws Exception
+     */
+    public boolean createKeyspace(String keyspaceName, JsonKeySpace kspObject) throws Exception {
+        return true;
+    }
+
+
+    private static void syncQuorum(String key) throws Exception {
+        logger.info(EELFLoggerDelegate.applicationLogger,"Performing sync operation---");
+        String[] splitString = key.split("\\.");
+        String keyspaceName = splitString[0];
+        String tableName = splitString[1];
+        String primaryKeyValue = splitString[2];
+        PreparedQueryObject selectQuery = new PreparedQueryObject();
+        PreparedQueryObject updateQuery = new PreparedQueryObject();
+
+        // get the primary key d
+        TableMetadata tableInfo = MusicDataStoreHandle.returnColumnMetadata(keyspaceName, tableName);
+        String primaryKeyName = tableInfo.getPrimaryKey().get(0).getName();// we only support single
+                                                                           // primary key
+        DataType primaryKeyType = tableInfo.getPrimaryKey().get(0).getType();
+        Object cqlFormattedPrimaryKeyValue =
+                        MusicUtil.convertToActualDataType(primaryKeyType, primaryKeyValue);
+
+        // get the row of data from a quorum
+        selectQuery.appendQueryString("SELECT *  FROM " + keyspaceName + "." + tableName + " WHERE "
+                        + primaryKeyName + "= ?" + ";");
+        selectQuery.addValue(cqlFormattedPrimaryKeyValue);
+        ResultSet results = null;
+        try {
+            results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(selectQuery);
+            // write it back to a quorum
+            Row row = results.one();
+            ColumnDefinitions colInfo = row.getColumnDefinitions();
+            int totalColumns = colInfo.size();
+            int counter = 1;
+            StringBuilder fieldValueString = new StringBuilder("");
+            for (Definition definition : colInfo) {
+                String colName = definition.getName();
+                if (colName.equals(primaryKeyName))
+                    continue;
+                DataType colType = definition.getType();
+                Object valueObj = MusicDataStoreHandle.getDSHandle().getColValue(row, colName, colType);
+                Object valueString = MusicUtil.convertToActualDataType(colType, valueObj);
+                fieldValueString.append(colName + " = ?");
+                updateQuery.addValue(valueString);
+                if (counter != (totalColumns - 1))
+                    fieldValueString.append(",");
+                counter = counter + 1;
+            }
+            updateQuery.appendQueryString("UPDATE " + keyspaceName + "." + tableName + " SET "
+                            + fieldValueString + " WHERE " + primaryKeyName + "= ? " + ";");
+            updateQuery.addValue(cqlFormattedPrimaryKeyValue);
+
+            MusicDataStoreHandle.getDSHandle().executePut(updateQuery, "critical");
+        } catch (MusicServiceException | MusicQueryException e) {
+            logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.QUERYERROR +""+updateQuery ,ErrorSeverity.MAJOR, ErrorTypes.QUERYERROR);
+        }
+    }
+
+
+
+
+    /**
+     *
+     * @param query
+     * @return ResultSet
+     */
+    public  ResultSet quorumGet(PreparedQueryObject query) {
+        ResultSet results = null;
+        try {
+            results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(query);
+        } catch (MusicServiceException | MusicQueryException e) {
+            logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.UNKNOWNERROR ,ErrorSeverity.MAJOR, ErrorTypes.GENERALSERVICEERROR);
+
+        }
+        return results;
+
+    }
+
+
+
+    /**
+     *
+     * @param lockName
+     * @return
+     */
+    public  String whoseTurnIsIt(String lockName) {
+
+        try {
+            return getLockingServiceHandle().whoseTurnIsIt("/" + lockName) + "";
+        } catch (MusicLockingException e) {
+             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKINGERROR+lockName ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+        }
+        return null;
+
+
+    }
+
+    /**
+     *
+     * @param lockId
+     * @return
+     */
+    public static String getLockNameFromId(String lockId) {
+        StringTokenizer st = new StringTokenizer(lockId);
+        return st.nextToken("$");
+    }
+
+    public void destroyLockRef(String lockId) {
+        long start = System.currentTimeMillis();
+        try {
+            getLockingServiceHandle().unlockAndDeleteId(lockId);
+        } catch (MusicLockingException | NoNodeException e) {
+            logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.DESTROYLOCK+lockId  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+        }
+        long end = System.currentTimeMillis();
+        logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to destroy lock reference:" + (end - start) + " ms");
+    }
+
+    public MusicLockState releaseLock(String lockId, boolean voluntaryRelease) {
+        long start = System.currentTimeMillis();
+        try {
+            getLockingServiceHandle().unlockAndDeleteId(lockId);
+        } catch (MusicLockingException e1) {
+            logger.error(EELFLoggerDelegate.errorLogger,e1.getMessage(), AppMessages.RELEASELOCK+lockId  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+        } catch (KeeperException.NoNodeException nne) {
+            logger.error(EELFLoggerDelegate.errorLogger,"Failed to release Lock " + lockId + " " + nne);
+            MusicLockState mls = new MusicLockState("Lock doesn't exists. Release lock operation failed.");
+            return mls;
+        }
+        String lockName = getLockNameFromId(lockId);
+        MusicLockState mls;
+        String lockHolder = null;
+        if (voluntaryRelease) {
+            mls = new MusicLockState(MusicLockState.LockStatus.UNLOCKED, lockHolder);
+            logger.info(EELFLoggerDelegate.applicationLogger,"In unlock: lock voluntarily released for " + lockId);
+        } else {
+            boolean needToSyncQuorum = true;
+            mls = new MusicLockState(MusicLockState.LockStatus.UNLOCKED, lockHolder,
+                            needToSyncQuorum);
+            logger.info(EELFLoggerDelegate.applicationLogger,"In unlock: lock forcibly released for " + lockId);
+        }
+        try {
+            getLockingServiceHandle().setLockState(lockName, mls);
+        } catch (MusicLockingException e) {
+            logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.RELEASELOCK+lockId  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+        }
+        long end = System.currentTimeMillis();
+        logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to release lock:" + (end - start) + " ms");
+        return mls;
+    }
+
+    public static  void  voluntaryReleaseLock(String lockId) throws MusicLockingException{
+        try {
+            getLockingServiceHandle().unlockAndDeleteId(lockId);
+        } catch (KeeperException.NoNodeException e) {
+            // ??? No way
+        }
+    }
+
+    /**
+     *
+     * @param lockName
+     * @throws MusicLockingException
+     */
+    public  void deleteLock(String lockName) throws MusicLockingException {
+        long start = System.currentTimeMillis();
+        logger.info(EELFLoggerDelegate.applicationLogger,"Deleting lock for " + lockName);
+        try {
+            getLockingServiceHandle().deleteLock("/" + lockName);
+        } catch (MusicLockingException e) {
+             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.DELTELOCK+lockName  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+             throw new MusicLockingException(e.getMessage());
+        }
+        long end = System.currentTimeMillis();
+        logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to delete lock:" + (end - start) + " ms");
+    }
+
+
+    /**
+     *
+     * @param nodeName
+     */
+    public static void pureZkCreate(String nodeName) {
+        try {
+            getLockingServiceHandle().getzkLockHandle().createNode(nodeName);
+        } catch (MusicLockingException e) {
+             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), "[ERR512E] Failed to get ZK Lock Handle "  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+        }
+    }
+
+    /**
+     *
+     * @param nodeName
+     * @param data
+     */
+    public static void pureZkWrite(String nodeName, byte[] data) {
+        long start = System.currentTimeMillis();
+        logger.info(EELFLoggerDelegate.applicationLogger,"Performing zookeeper write to " + nodeName);
+        try {
+            getLockingServiceHandle().getzkLockHandle().setNodeData(nodeName, data);
+        } catch (MusicLockingException e) {
+            logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), "[ERR512E] Failed to get ZK Lock Handle "  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+        }
+        logger.info(EELFLoggerDelegate.applicationLogger,"Performed zookeeper write to " + nodeName);
+        long end = System.currentTimeMillis();
+        logger.info(EELFLoggerDelegate.applicationLogger,"Time taken for the actual zk put:" + (end - start) + " ms");
+    }
+
+    /**
+     *
+     * @param nodeName
+     * @return
+     */
+    public static byte[] pureZkRead(String nodeName) {
+        long start = System.currentTimeMillis();
+        byte[] data = null;
+        try {
+            data = getLockingServiceHandle().getzkLockHandle().getNodeData(nodeName);
+        } catch (MusicLockingException e) {
+            logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), "[ERR512E] Failed to get ZK Lock Handle "  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+        }
+        long end = System.currentTimeMillis();
+        logger.info(EELFLoggerDelegate.applicationLogger,"Time taken for the actual zk put:" + (end - start) + " ms");
+        return data;
+    }
+
+
+
+    // Prepared Query Additions.
+
+    /**
+     *
+     * @param keyspaceName
+     * @param tableName
+     * @param primaryKey
+     * @param queryObject
+     * @return ReturnType
+     * @throws MusicServiceException
+     */
+    public  ReturnType eventualPut(PreparedQueryObject queryObject) {
+        boolean result = false;
+        try {
+            result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, MusicUtil.EVENTUAL);
+        } catch (MusicServiceException | MusicQueryException ex) {
+            logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage(), "[ERR512E] Failed to get ZK Lock Handle "  ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
+            logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage() + "  " + ex.getCause() + " " + ex);
+            return new ReturnType(ResultType.FAILURE, ex.getMessage());
+        }
+        if (result) {
+            return new ReturnType(ResultType.SUCCESS, "Eventual Operation Successfully performed");
+        } else {
+            return new ReturnType(ResultType.FAILURE, "Eventual Operation failed to perform");
+        }
+    }
+
+    /**
+     *
+     * @param keyspaceName
+     * @param tableName
+     * @param primaryKey
+     * @param queryObject
+     * @param lockId
+     * @return
+     */
+    public  ReturnType criticalPut(String keyspaceName, String tableName, String primaryKey,
+                    PreparedQueryObject queryObject, String lockId, Condition conditionInfo) {
+        long start = System.currentTimeMillis();
+
+        try {
+            MusicLockState mls = getLockingServiceHandle()
+                            .getLockState(keyspaceName + "." + tableName + "." + primaryKey);
+            if (mls.getLockHolder().equals(lockId) == true) {
+                if (conditionInfo != null)
+                  try {
+                    if (conditionInfo.testCondition() == false)
+                        return new ReturnType(ResultType.FAILURE,
+                                        "Lock acquired but the condition is not true");
+                  } catch (Exception e) {
+                    return new ReturnType(ResultType.FAILURE,
+                            "Exception thrown while doing the critical put, check sanctity of the row/conditions:\n"
+                                            + e.getMessage());
+                  }
+                boolean result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, MusicUtil.CRITICAL);
+                long end = System.currentTimeMillis();
+                logger.info(EELFLoggerDelegate.applicationLogger,"Time taken for the critical put:" + (end - start) + " ms");
+                if (result) {
+                    return new ReturnType(ResultType.SUCCESS, "Update performed");
+                } else {
+                    return new ReturnType(ResultType.FAILURE, "Update failed to perform");
+                }
+            } else
+                return new ReturnType(ResultType.FAILURE,
+                                "Cannot perform operation since you are the not the lock holder");
+        } catch (MusicQueryException | MusicServiceException  e) {
+            logger.error(EELFLoggerDelegate.errorLogger,e.getMessage());
+            return new ReturnType(ResultType.FAILURE,
+                            "Exception thrown while doing the critical put, check sanctity of the row/conditions:\n"
+                                            + e.getMessage());
+        }catch(MusicLockingException ex){
+            return new ReturnType(ResultType.FAILURE,ex.getMessage());
+        }
+
+    }
+
+    /**
+     *
+     * @param queryObject
+     * @param consistency
+     * @return Boolean Indicates success or failure
+     * @throws MusicServiceException
+     *
+     *
+     */
+    public  ResultType nonKeyRelatedPut(PreparedQueryObject queryObject, String consistency) throws MusicServiceException {
+        // this is mainly for some functions like keyspace creation etc which does not
+        // really need the bells and whistles of Music locking.
+        boolean result = false;
+        try {
+            result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, consistency);
+        } catch (MusicQueryException | MusicServiceException ex) {
+            logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage(), AppMessages.UNKNOWNERROR  ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
+            throw new MusicServiceException(ex.getMessage());
+        }
+        return result?ResultType.SUCCESS:ResultType.FAILURE;
+    }
+
+    /**
+     * This method performs DDL operation on cassandra.
+     *
+     * @param queryObject query object containing prepared query and values
+     * @return ResultSet
+     * @throws MusicServiceException
+     */
+    public  ResultSet get(PreparedQueryObject queryObject) throws MusicServiceException {
+        ResultSet results = null;
+        try {
+            results = MusicDataStoreHandle.getDSHandle().executeOneConsistencyGet(queryObject);
+        } catch (MusicQueryException | MusicServiceException e) {
+            logger.error(EELFLoggerDelegate.errorLogger,e.getMessage());
+            throw new MusicServiceException(e.getMessage());
+        }
+        return results;
+    }
+
+    public static String getMyHostId() {
+        PreparedQueryObject pQuery = new PreparedQueryObject();
+        pQuery.appendQueryString("SELECT HOST_ID FROM SYSTEM.LOCAL");
+        ResultSet rs = null;
+        try {
+            rs = MusicDataStoreHandle.getDSHandle().executeOneConsistencyGet(pQuery);
+            Row row = rs.one();
+            return (row == null) ? "UNKNOWN" : row.getUUID("HOST_ID").toString();
+        } catch (Exception e) {
+            e.printStackTrace();
+            logger.error(EELFLoggerDelegate.errorLogger,e.getMessage());
+        }
+        logger.error(EELFLoggerDelegate.errorLogger, "Some issue during MusicCore.getMyHostId");
+        return "UNKNOW";
+    }
+
+    /**
+     * This method performs DDL operations on cassandra, if the the resource is available. Lock ID
+     * is used to check if the resource is free.
+     *
+     * @param keyspaceName name of the keyspace
+     * @param tableName name of the table
+     * @param primaryKey primary key value
+     * @param queryObject query object containing prepared query and values
+     * @param lockId lock ID to check if the resource is free to perform the operation.
+     * @return ResultSet
+     */
+    public  ResultSet criticalGet(String keyspaceName, String tableName, String primaryKey,
+                    PreparedQueryObject queryObject, String lockId) throws MusicServiceException {
+        ResultSet results = null;
+        try {
+            MusicLockState mls = getLockingServiceHandle()
+                            .getLockState(keyspaceName + "." + tableName + "." + primaryKey);
+            if (mls.getLockHolder().equals(lockId)) {
+                results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(queryObject);
+            } else
+                throw new MusicServiceException("YOU DO NOT HAVE THE LOCK");
+        } catch (MusicQueryException | MusicServiceException | MusicLockingException e) {
+            logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.UNKNOWNERROR  ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
+        }
+        return results;
+    }
+
+    /**
+     * This method performs DML operation on cassandra, when the lock of the dd is acquired.
+     *
+     * @param keyspaceName name of the keyspace
+     * @param tableName name of the table
+     * @param primaryKey primary key value
+     * @param queryObject query object containing prepared query and values
+     * @return ReturnType
+     * @throws MusicLockingException
+     */
+    public  ReturnType atomicPut(String keyspaceName, String tableName, String primaryKey,
+                    PreparedQueryObject queryObject, Condition conditionInfo) throws MusicLockingException {
+
+        long start = System.currentTimeMillis();
+        String key = keyspaceName + "." + tableName + "." + primaryKey;
+        String lockId = createLockReference(key);
+        long lockCreationTime = System.currentTimeMillis();
+        ReturnType lockAcqResult = acquireLock(key, lockId);
+        long lockAcqTime = System.currentTimeMillis();
+        if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
+            logger.info(EELFLoggerDelegate.applicationLogger,"acquired lock with id " + lockId);
+            ReturnType criticalPutResult = criticalPut(keyspaceName, tableName, primaryKey,
+                            queryObject, lockId, conditionInfo);
+            long criticalPutTime = System.currentTimeMillis();
+            voluntaryReleaseLock(lockId);
+            long lockDeleteTime = System.currentTimeMillis();
+            String timingInfo = "|lock creation time:" + (lockCreationTime - start)
+                            + "|lock accquire time:" + (lockAcqTime - lockCreationTime)
+                            + "|critical put time:" + (criticalPutTime - lockAcqTime)
+                            + "|lock delete time:" + (lockDeleteTime - criticalPutTime) + "|";
+            criticalPutResult.setTimingInfo(timingInfo);
+            return criticalPutResult;
+        } else {
+            logger.info(EELFLoggerDelegate.applicationLogger,"unable to acquire lock, id " + lockId);
+            destroyLockRef(lockId);
+            return lockAcqResult;
+        }
+    }
+
+    /**
+     * this function is mainly for the benchmarks to see the effect of lock deletion.
+     *
+     * @param keyspaceName
+     * @param tableName
+     * @param primaryKey
+     * @param queryObject
+     * @param conditionInfo
+     * @return
+     * @throws MusicLockingException
+     */
+    public  ReturnType atomicPutWithDeleteLock(String keyspaceName, String tableName,
+                    String primaryKey, PreparedQueryObject queryObject, Condition conditionInfo) throws MusicLockingException {
+
+        long start = System.currentTimeMillis();
+        String key = keyspaceName + "." + tableName + "." + primaryKey;
+        String lockId = createLockReference(key);
+        long lockCreationTime = System.currentTimeMillis();
+        long leasePeriod = MusicUtil.getDefaultLockLeasePeriod();
+        ReturnType lockAcqResult = acquireLock(key, lockId);
+        long lockAcqTime = System.currentTimeMillis();
+        if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
+            logger.info(EELFLoggerDelegate.applicationLogger,"acquired lock with id " + lockId);
+            ReturnType criticalPutResult = criticalPut(keyspaceName, tableName, primaryKey,
+                            queryObject, lockId, conditionInfo);
+            long criticalPutTime = System.currentTimeMillis();
+            deleteLock(key);
+            long lockDeleteTime = System.currentTimeMillis();
+            String timingInfo = "|lock creation time:" + (lockCreationTime - start)
+                            + "|lock accquire time:" + (lockAcqTime - lockCreationTime)
+                            + "|critical put time:" + (criticalPutTime - lockAcqTime)
+                            + "|lock delete time:" + (lockDeleteTime - criticalPutTime) + "|";
+            criticalPutResult.setTimingInfo(timingInfo);
+            return criticalPutResult;
+        } else {
+            logger.info(EELFLoggerDelegate.applicationLogger,"unable to acquire lock, id " + lockId);
+            deleteLock(key);
+            return lockAcqResult;
+        }
+    }
+
+
+
+
+    /**
+     * This method performs DDL operation on cassasndra, when the lock for the resource is acquired.
+     *
+     * @param keyspaceName name of the keyspace
+     * @param tableName name of the table
+     * @param primaryKey primary key value
+     * @param queryObject query object containing prepared query and values
+     * @return ResultSet
+     * @throws MusicServiceException
+     * @throws MusicLockingException
+     */
+    public  ResultSet atomicGet(String keyspaceName, String tableName, String primaryKey,
+                    PreparedQueryObject queryObject) throws MusicServiceException, MusicLockingException {
+        String key = keyspaceName + "." + tableName + "." + primaryKey;
+        String lockId = createLockReference(key);
+        long leasePeriod = MusicUtil.getDefaultLockLeasePeriod();
+        ReturnType lockAcqResult = acquireLock(key, lockId);
+        if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
+            logger.info(EELFLoggerDelegate.applicationLogger,"acquired lock with id " + lockId);
+            ResultSet result =
+                            criticalGet(keyspaceName, tableName, primaryKey, queryObject, lockId);
+            voluntaryReleaseLock(lockId);
+            return result;
+        } else {
+            destroyLockRef(lockId);
+            logger.info(EELFLoggerDelegate.applicationLogger,"unable to acquire lock, id " + lockId);
+            return null;
+        }
+    }
+
+    public  ResultSet atomicGetWithDeleteLock(String keyspaceName, String tableName, String primaryKey,
+            PreparedQueryObject queryObject) throws MusicServiceException, MusicLockingException {
+        String key = keyspaceName + "." + tableName + "." + primaryKey;
+        String lockId = createLockReference(key);
+        long leasePeriod = MusicUtil.getDefaultLockLeasePeriod();
+
+        ReturnType lockAcqResult = acquireLock(key, lockId);
+
+        if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
+            logger.info(EELFLoggerDelegate.applicationLogger, "acquired lock with id " + lockId);
+            ResultSet result = criticalGet(keyspaceName, tableName, primaryKey, queryObject, lockId);
+            deleteLock(key);
+            return result;
+        } else {
+            deleteLock(key);
+            logger.info(EELFLoggerDelegate.applicationLogger, "unable to acquire lock, id " + lockId);
+            return null;
+        }
+    }
+
+    /**
+     * @param lockName
+     * @return
+     */
+    public Map<String, Object> validateLock(String lockName) {
+        Map<String, Object> resultMap = new HashMap<>();
+        String[] locks = lockName.split("\\.");
+        if(locks.length < 3) {
+            resultMap.put("Error", "Invalid lock. Please make sure lock is of the type keyspaceName.tableName.primaryKey");
+            return resultMap;
+        }
+        String keyspace= locks[0];
+        if(keyspace.startsWith("$"))
+            keyspace = keyspace.substring(1);
+        resultMap.put("keyspace",keyspace);
+        return resultMap;
+    }
+
+
+
+    @Override
+    public ResultType createTable(String keyspace, String table, PreparedQueryObject tableQueryObject,
+            String consistency) throws MusicServiceException {
+        boolean result = false;
+        
+        try {
+            //create shadow locking table 
+            result = createLockQueue(keyspace, table);
+            if(result == false) 
+                return ResultType.FAILURE;
+
+            result = false;
+            
+            //create table to track unsynced_keys
+            table = "unsyncedKeys_"+table; 
+            
+            String tabQuery = "CREATE TABLE IF NOT EXISTS "+keyspace+"."+table
+                    + " ( key text,PRIMARY KEY (key) );";
+            System.out.println(tabQuery);
+            PreparedQueryObject queryObject = new PreparedQueryObject(); 
+            
+            queryObject.appendQueryString(tabQuery);
+            result = false;
+            result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, "eventual");
+
+        
+            //create actual table
+            result = MusicDataStoreHandle.getDSHandle().executePut(tableQueryObject, consistency);
+        } catch (MusicQueryException | MusicServiceException ex) {
+            logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage(), AppMessages.UNKNOWNERROR  ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
+            throw new MusicServiceException(ex.getMessage());
+        }
+        return result?ResultType.SUCCESS:ResultType.FAILURE;
+    }
+    
+    public static boolean createLockQueue(String keyspace, String table) throws MusicServiceException, MusicQueryException {
+        logger.info(EELFLoggerDelegate.applicationLogger,
+                "Create lock queue/table for " +  keyspace+"."+table);
+        table = "lockQ_"+table; 
+        String tabQuery = "CREATE TABLE IF NOT EXISTS "+keyspace+"."+table
+                + " ( key text, lockReference bigint, createTime text, acquireTime text, guard bigint static, PRIMARY KEY ((key), lockReference) ) "
+                + "WITH CLUSTERING ORDER BY (lockReference ASC);";
+        System.out.println(tabQuery);
+        PreparedQueryObject queryObject = new PreparedQueryObject(); 
+        
+        queryObject.appendQueryString(tabQuery);
+        boolean result;
+        result = MusicDataStoreHandle.mDstoreHandle.executePut(queryObject, "eventual");
+        return result;
+    }
+
+
+    @Override
+    public List<String> getLockQueue(String fullyQualifiedKey)
+            throws MusicServiceException, MusicQueryException, MusicLockingException {
+        // TODO Auto-generated method stub
+        return null;
+    }
+
+
+
+    @Override
+    public long getLockQueueSize(String fullyQualifiedKey)
+            throws MusicServiceException, MusicQueryException, MusicLockingException {
+        // TODO Auto-generated method stub
+        return 0;
+    }
+    @Override
+    public ReturnType eventualPut_nb(PreparedQueryObject queryObject, String keyspace, String tablename,
+            String primaryKey) {
+        return eventualPut(queryObject);
+    }
+
+
+
+
+}