path: root/src/main/java/org/onap/music/service/impl/MusicZKCore.java

/*
 * ============LICENSE_START==========================================
 * org.onap.music
 * ===================================================================
 *  Copyright (c) 2017 AT&T Intellectual Property
 * ===================================================================
 *  Modifications Copyright (c) 2019 Samsung
 * ===================================================================
 *  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.List;
import java.util.Map;
import java.util.StringTokenizer;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.NoNodeException;
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.DataType;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.TableMetadata;

/**
 * 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);
        MusicUtil.writeBackToQuorum(selectQuery, primaryKeyName, updateQuery, keyspaceName, tableName,
            cqlFormattedPrimaryKeyValue);
    }




    /**
     *
     * @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) {
        return MusicUtil.validateLock(lockName);
    }

    @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);
    }




}