music-core/src/main/java/org/onap/music/service/impl/MusicCassaCore.java

   1 /*
   2  * ============LICENSE_START==========================================
   3  * org.onap.music
   4  * ===================================================================
   5  *  Copyright (c) 2017 AT&T Intellectual Property
   6  *  Modifications Copyright (c) 2018 IBM.
   7  * ===================================================================
   8  *  Modifications Copyright (c) 2019 Samsung
   9  * ===================================================================
  10  *  Licensed under the Apache License, Version 2.0 (the "License");
  11  *  you may not use this file except in compliance with the License.
  12  *  You may obtain a copy of the License at
  13  *
  14  *     http://www.apache.org/licenses/LICENSE-2.0
  15  *
  16  *  Unless required by applicable law or agreed to in writing, software
  17  *  distributed under the License is distributed on an "AS IS" BASIS,
  18  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  19  *  See the License for the specific language governing permissions and
  20  *  limitations under the License.
  21  *
  22  * ============LICENSE_END=============================================
  23  * ====================================================================
  24  */
  25
  26 package org.onap.music.service.impl;
  27
  28 import java.io.StringWriter;
  29 import java.util.Collections;
  30 import java.util.HashSet;
  31 import java.util.List;
  32 import java.util.Map;
  33 import java.util.Set;
  34 import java.util.StringTokenizer;
  35
  36 import javax.ws.rs.core.MultivaluedMap;
  37
  38 import org.onap.music.datastore.Condition;
  39 import org.onap.music.datastore.MusicDataStore;
  40 import org.onap.music.datastore.MusicDataStoreHandle;
  41 import org.onap.music.datastore.PreparedQueryObject;
  42 import org.onap.music.datastore.jsonobjects.JsonDelete;
  43 import org.onap.music.datastore.jsonobjects.JsonIndex;
  44 import org.onap.music.datastore.jsonobjects.JsonInsert;
  45 import org.onap.music.datastore.jsonobjects.JsonKeySpace;
  46 import org.onap.music.datastore.jsonobjects.JsonSelect;
  47 import org.onap.music.datastore.jsonobjects.JsonTable;
  48 import org.onap.music.datastore.jsonobjects.JsonUpdate;
  49 import org.onap.music.eelf.logging.EELFLoggerDelegate;
  50 import org.onap.music.eelf.logging.format.AppMessages;
  51 import org.onap.music.eelf.logging.format.ErrorSeverity;
  52 import org.onap.music.eelf.logging.format.ErrorTypes;
  53 import org.onap.music.exceptions.MusicDeadlockException;
  54 import org.onap.music.exceptions.MusicLockingException;
  55 import org.onap.music.exceptions.MusicQueryException;
  56 import org.onap.music.exceptions.MusicServiceException;
  57 import org.onap.music.lockingservice.cassandra.CassaLockStore;
  58 import org.onap.music.lockingservice.cassandra.CassaLockStore.LockObject;
  59 import org.onap.music.lockingservice.cassandra.LockType;
  60 import org.onap.music.lockingservice.cassandra.MusicLockState;
  61 import org.onap.music.lockingservice.cassandra.MusicLockState.LockStatus;
  62 import org.onap.music.main.MusicUtil;
  63 import org.onap.music.main.ResultType;
  64 import org.onap.music.main.ReturnType;
  65 import org.onap.music.service.MusicCoreService;
  66
  67 import com.datastax.driver.core.DataType;
  68 import com.datastax.driver.core.ResultSet;
  69 import com.datastax.driver.core.Row;
  70 import com.datastax.driver.core.TableMetadata;
  71
  72 public class MusicCassaCore implements MusicCoreService {
  73
  74     private static CassaLockStore mLockHandle = null;
  75     private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(MusicCassaCore.class);
  76     private static MusicCassaCore musicCassaCoreInstance = null;
  77     private static Set<String> set = Collections.synchronizedSet(new HashSet<String>());
  78
  79     private MusicCassaCore() {
  80         // not going to happen
  81     }
  82
  83     public static CassaLockStore getmLockHandle() {
  84         return mLockHandle;
  85     }
  86
  87     public static void setmLockHandle(CassaLockStore mLockHandle) {
  88         MusicCassaCore.mLockHandle = mLockHandle;
  89     }
  90
  91     public static MusicCassaCore getInstance() {
  92
  93         if(musicCassaCoreInstance == null) {
  94             musicCassaCoreInstance = new MusicCassaCore();
  95         }
  96         return musicCassaCoreInstance;
  97     }
  98
  99     public static CassaLockStore getLockingServiceHandle() throws MusicLockingException {
 100         logger.info(EELFLoggerDelegate.applicationLogger,"Acquiring lock store handle");
 101         long start = System.currentTimeMillis();
 102
 103         if (mLockHandle == null) {
 104             try {
 105                 mLockHandle = new CassaLockStore(MusicDataStoreHandle.getDSHandle());
 106             } catch (Exception e) {
 107                 logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKHANDLE,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 108                 throw new MusicLockingException("Failed to aquire Locl store handle " + e);
 109             }
 110         }
 111         long end = System.currentTimeMillis();
 112         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to acquire lock store handle:" + (end - start) + " ms");
 113         return mLockHandle;
 114     }
 115
 116     public String createLockReferenceAtomic(String fullyQualifiedKey) throws MusicLockingException {
 117         return createLockReferenceAtomic(fullyQualifiedKey, LockType.WRITE);
 118     }
 119
 120     /**
 121      * This will be called for Atomic calls
 122      *
 123      */
 124     public String createLockReferenceAtomic(String fullyQualifiedKey, LockType locktype) throws MusicLockingException {
 125         String[] splitString = fullyQualifiedKey.split("\\.");
 126         if (splitString.length < 3) {
 127             throw new MusicLockingException("Missing or incorrect lock details. Check table or key name.");
 128         }
 129         String keyspace = splitString[0];
 130         String table = splitString[1];
 131         String lockName = splitString[2];
 132
 133         logger.info(EELFLoggerDelegate.applicationLogger,"Creating lock reference for lock name:" + lockName);
 134         long start = 0L;
 135         long end = 0L;
 136         String lockReference = null;
 137         LockObject peek = null;
 138
 139         /** Lets check for an existing lock.
 140          * This will allow us to limit the amount of requests going forward.
 141          */
 142         start = System.currentTimeMillis();
 143         try {
 144             peek = getLockingServiceHandle().peekLockQueue(keyspace, table, lockName);
 145         } catch (MusicServiceException | MusicQueryException e) {
 146             //logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(),e);
 147             throw new MusicLockingException("Error getting lockholder info for key [" + lockName +"]:" + e.getMessage());
 148         }
 149
 150         if(peek!=null && (peek.getLocktype()!=null && peek.getLocktype().equals(LockType.WRITE)) && peek.getAcquireTime()!=null && peek.getLockRef()!=null) {
 151             long currentTime = System.currentTimeMillis();
 152             if((currentTime-Long.parseLong(peek.getAcquireTime()))<MusicUtil.getDefaultLockLeasePeriod()){
 153                 //logger.info(EELFLoggerDelegate.applicationLogger,"Lock holder exists and lease not expired. Please try again for key="+lockName);
 154                 throw new MusicLockingException("Unable to create lock reference for key [" + lockName + "]. Please try again.");
 155             }
 156         }
 157         end = System.currentTimeMillis();
 158         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to check for lock reference for key [" + lockName + "]:" + (end - start) + " ms");
 159
 160         start = System.currentTimeMillis();
 161         /* We are Creating a Thread safe set and adding the key to the set.
 162         * if a key exists then it wil be passed over and not go to the lock creation.
 163         * If a key doesn't exist then it will set the value in the set and continue to create a lock.
 164         *
 165         * This will ensure that no 2 threads using the same key will be able to try to create a lock
 166         * This wil in turn squash the amout of LWT Chatter in Cassandra an reduce the amount of
 167         * WriteTimeoutExceptions being experiences on single keys.
 168         */
 169         if ( set.add(fullyQualifiedKey)) {
 170             try {
 171                 lockReference = "" + getLockingServiceHandle().genLockRefandEnQueue(keyspace, table, lockName, locktype,null);
 172                 set.remove(fullyQualifiedKey);
 173             } catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
 174                 set.remove(fullyQualifiedKey);
 175                 throw new MusicLockingException(e.getMessage());
 176             } catch (Exception e) {
 177                 set.remove(fullyQualifiedKey);
 178                 e.printStackTrace();
 179                 logger.error(EELFLoggerDelegate.applicationLogger,"Exception in creatLockEnforced:"+ e.getMessage(),e);
 180                 throw new MusicLockingException("Unable to create lock reference for key [" + lockName + "]. " + e.getMessage());
 181             }
 182         } else {
 183             throw new MusicLockingException("Unable to create lock reference for key [" + lockName + "]. Please try again.");
 184         }
 185         end = System.currentTimeMillis();
 186         logger.info(EELFLoggerDelegate.debugLogger,"### Set = " + set);
 187         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to create lock reference  for key [" + lockName + "]:" + (end - start) + " ms");
 188         return lockReference;
 189
 190         //return createLockReference(fullyQualifiedKey, locktype, null);
 191     }
 192
 193     public String createLockReference(String fullyQualifiedKey, LockType locktype, String owner) throws MusicLockingException {
 194         String[] splitString = fullyQualifiedKey.split("\\.");
 195         if (splitString.length < 3) {
 196             throw new MusicLockingException("Missing or incorrect lock details. Check table or key name.");
 197         }
 198         String keyspace = splitString[0];
 199         String table = splitString[1];
 200         String lockName = splitString[2];
 201
 202         logger.info(EELFLoggerDelegate.applicationLogger,"Creating lock reference for lock name:" + lockName);
 203         long start = 0L;
 204         long end = 0L;
 205         String lockReference = null;
 206
 207         /* Check for a Deadlock */
 208         try {
 209             boolean deadlock = getLockingServiceHandle().checkForDeadlock(keyspace, table, lockName, locktype, owner, false);
 210             if (deadlock) {
 211                 MusicDeadlockException e = new MusicDeadlockException("Deadlock detected when " + owner + " tried to create lock on " + keyspace + "." + table + "." + lockName);
 212                 e.setValues(owner, keyspace, table, lockName);
 213                 throw e;
 214             }
 215         } catch (MusicDeadlockException e) {
 216             //just threw this, no need to wrap it
 217             throw e;
 218         } catch (MusicServiceException | MusicQueryException e) {
 219             logger.error(EELFLoggerDelegate.applicationLogger, e);
 220             throw new MusicLockingException("Unable to check for deadlock. " + e.getMessage(), e);
 221         }
 222         end = System.currentTimeMillis();
 223         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to check for deadlock for key [" + lockName + "]:" + (end - start) + " ms");
 224
 225         start = System.currentTimeMillis();
 226         try {
 227             lockReference = "" + getLockingServiceHandle().genLockRefandEnQueue(keyspace, table, lockName, locktype, owner);
 228         } catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
 229             logger.info(EELFLoggerDelegate.applicationLogger,e.getMessage(),e);
 230             throw new MusicLockingException("Unable to create lock reference for key [" + lockName + "]. Please try again: " + e.getMessage());
 231         } catch (Exception e) {
 232             logger.error(EELFLoggerDelegate.applicationLogger,e.getMessage(),e);
 233             throw new MusicLockingException("Unable to create lock reference. " + e.getMessage(), e);
 234         }
 235         end = System.currentTimeMillis();
 236         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to create lock reference  for key [" + lockName + "]:" + (end - start) + " ms");
 237         return lockReference;
 238     }
 239
 240     public ReturnType promoteLock(String lockId) throws MusicLockingException {
 241         String[] splitString = lockId.split("\\.");
 242         String keyspace = splitString[0].substring(1);//remove '$'
 243         String table = splitString[1];
 244         String primaryKeyValue = splitString[2].substring(0, splitString[2].lastIndexOf("$"));
 245         String lockRef = lockId.substring(lockId.lastIndexOf("$")+1); //lockRef is "$" to end
 246
 247         logger.info(EELFLoggerDelegate.applicationLogger,"Attempting to promote lock " + lockId);
 248
 249         try {
 250             return getLockingServiceHandle().promoteLock(keyspace, table, primaryKeyValue, lockRef);
 251         } catch (MusicServiceException e) {
 252             throw new MusicLockingException("Unable to promote lock. ", e);
 253         } catch (MusicQueryException e) {
 254             throw new MusicLockingException("Unable to promote lock. ", e);
 255         }
 256
 257     }
 258
 259
 260     public ReturnType acquireLockWithLease(String fullyQualifiedKey, String lockReference, long leasePeriod)
 261             throws MusicLockingException, MusicQueryException, MusicServiceException  {
 262         evictExpiredLockHolder(fullyQualifiedKey,leasePeriod);
 263         return acquireLock(fullyQualifiedKey, lockReference);
 264     }
 265
 266     private void evictExpiredLockHolder(String fullyQualifiedKey, long leasePeriod)
 267             throws MusicLockingException, MusicQueryException, MusicServiceException {
 268         String[] splitString = fullyQualifiedKey.split("\\.");
 269         String keyspace = splitString[0];
 270         String table = splitString[1];
 271         String primaryKeyValue = splitString[2];
 272
 273         LockObject currentLockHolderObject = getLockingServiceHandle().peekLockQueue(keyspace, table, primaryKeyValue);
 274
 275         if (!currentLockHolderObject.getIsLockOwner()) { // no lock holder
 276             return;
 277         }
 278         /*
 279          * Release the lock of the previous holder if it has expired. if the update to the acquire time has
 280          * not reached due to network delays, simply use the create time as the reference
 281          */
 282         long referenceTime = Math.max(Long.parseLong(currentLockHolderObject.getAcquireTime()),
 283                 Long.parseLong(currentLockHolderObject.getCreateTime()));
 284         if ((System.currentTimeMillis() - referenceTime) > leasePeriod) {
 285             forciblyReleaseLock(fullyQualifiedKey, currentLockHolderObject.getLockRef() + "");
 286             logger.info(EELFLoggerDelegate.applicationLogger, currentLockHolderObject.getLockRef() + " forcibly released");
 287         }
 288     }
 289
 290     public ReturnType acquireLock(String fullyQualifiedKey, String lockId)
 291             throws MusicLockingException, MusicQueryException, MusicServiceException {
 292         String[] splitString = lockId.split("\\.");
 293         String keyspace = splitString[0].substring(1);//remove '$'
 294         String table = splitString[1];
 295         String primaryKeyValue = splitString[2].substring(0, splitString[2].lastIndexOf("$"));
 296         String localFullyQualifiedKey = lockId.substring(1, lockId.lastIndexOf("$"));
 297         String lockRef = lockId.substring(lockId.lastIndexOf("$")+1); //lockRef is "$" to end
 298
 299         LockObject lockInfo = getLockingServiceHandle().getLockInfo(keyspace, table, primaryKeyValue, lockRef);
 300
 301         if (!lockInfo.getIsLockOwner()) {
 302             return new ReturnType(ResultType.FAILURE, lockId + " is not a lock holder");//not top of the lock store q
 303         }
 304
 305         if (getLockingServiceHandle().checkForDeadlock(keyspace, table, primaryKeyValue, lockInfo.getLocktype(), lockInfo.getOwner(), true)) {
 306             MusicDeadlockException e = new MusicDeadlockException("Deadlock detected when " + lockInfo.getOwner()  + " tried to create lock on " + keyspace + "." + table + "." + primaryKeyValue);
 307             e.setValues(lockInfo.getOwner(), keyspace, table, primaryKeyValue);
 308             throw e;
 309         }
 310
 311         //check to see if the value of the key has to be synced in case there was a forceful release
 312         String syncTable = keyspace+".unsyncedKeys_"+table;
 313         String query = "select * from "+syncTable+" where key='"+localFullyQualifiedKey+"';";
 314         PreparedQueryObject readQueryObject = new PreparedQueryObject();
 315         readQueryObject.appendQueryString(query);
 316         ResultSet results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(readQueryObject);
 317         if (!results.all().isEmpty()) {
 318             logger.info("In acquire lock: Since there was a forcible release, need to sync quorum!");
 319             try {
 320                 syncQuorum(keyspace, table, primaryKeyValue);
 321             } catch (Exception e) {
 322                 StringWriter sw = new StringWriter();
 323                     logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), "[ERR506E] Failed to aquire lock ",
 324                         ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR, e);
 325                 String exceptionAsString = sw.toString();
 326                 return new ReturnType(ResultType.FAILURE, "Exception thrown while syncing key:\n" + exceptionAsString);
 327             }
 328             String cleanQuery = "delete from " + syncTable + " where key='"+localFullyQualifiedKey+"';";
 329             PreparedQueryObject deleteQueryObject = new PreparedQueryObject();
 330             deleteQueryObject.appendQueryString(cleanQuery);
 331             MusicDataStoreHandle.getDSHandle().executePut(deleteQueryObject, "critical");
 332         }
 333
 334         getLockingServiceHandle().updateLockAcquireTime(keyspace, table, primaryKeyValue, lockRef);
 335
 336         return new ReturnType(ResultType.SUCCESS, lockRef+" is the lock holder for the key");
 337     }
 338
 339
 340
 341     /**
 342      *
 343      * @param tableQueryObject
 344      * @param consistency
 345      * @return Boolean Indicates success or failure
 346      * @throws MusicServiceException
 347      *
 348      *
 349      */
 350     public ResultType createTable(String keyspace, String table, PreparedQueryObject tableQueryObject,
 351             String consistency) throws MusicServiceException {
 352         boolean result = false;
 353
 354         try {
 355             // create shadow locking table
 356             result = getLockingServiceHandle().createLockQueue(keyspace, table);
 357             if (result == false)
 358                 return ResultType.FAILURE;
 359
 360             result = false;
 361
 362             // create table to track unsynced_keys
 363             table = "unsyncedKeys_" + table;
 364
 365             String tabQuery =
 366                     "CREATE TABLE IF NOT EXISTS " + keyspace + "." + table + " ( key text,PRIMARY KEY (key) );";
 367             PreparedQueryObject queryObject = new PreparedQueryObject();
 368
 369             queryObject.appendQueryString(tabQuery);
 370             result = false;
 371             result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, "eventual");
 372
 373             // create actual table
 374             result = MusicDataStoreHandle.getDSHandle().executePut(tableQueryObject, consistency);
 375         } catch (MusicQueryException | MusicServiceException | MusicLockingException ex) {
 376             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(), AppMessages.UNKNOWNERROR, ErrorSeverity.WARN,
 377                     ErrorTypes.MUSICSERVICEERROR);
 378             throw new MusicServiceException(ex.getMessage());
 379         }
 380         return result ? ResultType.SUCCESS : ResultType.FAILURE;
 381     }
 382
 383     private static void syncQuorum(String keyspace, String table, String primaryKeyValue) throws Exception {
 384         logger.info(EELFLoggerDelegate.applicationLogger,"Performing sync operation---");
 385         PreparedQueryObject selectQuery = new PreparedQueryObject();
 386         PreparedQueryObject updateQuery = new PreparedQueryObject();
 387
 388         // get the primary key d
 389         TableMetadata tableInfo = MusicDataStoreHandle.returnColumnMetadata(keyspace, table);
 390         String primaryKeyName = tableInfo.getPrimaryKey().get(0).getName(); // we only support single
 391                                                                             // primary key
 392         DataType primaryKeyType = tableInfo.getPrimaryKey().get(0).getType();
 393         Object cqlFormattedPrimaryKeyValue =
 394                         MusicUtil.convertToActualDataType(primaryKeyType, primaryKeyValue);
 395
 396         // get the row of data from a quorum
 397         selectQuery.appendQueryString("SELECT *  FROM " + keyspace + "." + table + " WHERE "
 398                         + primaryKeyName + "= ?" + ";");
 399         selectQuery.addValue(cqlFormattedPrimaryKeyValue);
 400         MusicUtil.writeBackToQuorum(selectQuery, primaryKeyName, updateQuery, keyspace, table,
 401             cqlFormattedPrimaryKeyValue);
 402     }
 403
 404     /**
 405      *
 406      * @param query
 407      * @return ResultSet
 408      */
 409     public ResultSet quorumGet(PreparedQueryObject query) {
 410         ResultSet results = null;
 411         try {
 412             results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(query);
 413         } catch (MusicServiceException | MusicQueryException e) {
 414             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.UNKNOWNERROR,
 415                 ErrorSeverity.MAJOR, ErrorTypes.GENERALSERVICEERROR, e);
 416
 417         }
 418         return results;
 419     }
 420
 421     public String whoseTurnIsIt(String fullyQualifiedKey) {
 422         String[] splitString = fullyQualifiedKey.split("\\.");
 423         String keyspace = splitString[0];
 424         String table = splitString[1];
 425         String primaryKeyValue = splitString[2];
 426         try {
 427             LockObject lockOwner = getLockingServiceHandle().peekLockQueue(keyspace, table, primaryKeyValue);
 428             if (!lockOwner.getIsLockOwner()) {
 429                 return null;
 430             }
 431             return "$" + fullyQualifiedKey + "$" + lockOwner.getLockRef();
 432         } catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
 433             logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), AppMessages.LOCKINGERROR + fullyQualifiedKey,
 434                     ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 435         }
 436         return null;
 437     }
 438
 439     public List<String> getCurrentLockHolders(String fullyQualifiedKey) {
 440         String[] splitString = fullyQualifiedKey.split("\\.");
 441         String keyspace = splitString[0];
 442         String table = splitString[1];
 443         String primaryKeyValue = splitString[2];
 444         try {
 445             return getLockingServiceHandle().getCurrentLockHolders(keyspace, table, primaryKeyValue);
 446         } catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
 447             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKINGERROR+fullyQualifiedKey ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 448         }
 449         return null;
 450     }
 451
 452     /**
 453      *
 454      * @param lockReference
 455      * @return
 456      */
 457     public static String getLockNameFromId(String lockReference) {
 458         StringTokenizer st = new StringTokenizer(lockReference);
 459         return st.nextToken("$");
 460     }
 461
 462     @Override
 463     public void destroyLockRef(String lockId) throws MusicLockingException {
 464         long start = System.currentTimeMillis();
 465         String fullyQualifiedKey = lockId.substring(1, lockId.lastIndexOf("$"));
 466         String lockRef = lockId.substring(lockId.lastIndexOf('$')+1);
 467         String[] splitString = fullyQualifiedKey.split("\\.");
 468         String keyspace = splitString[0];
 469         String table = splitString[1];
 470         String primaryKeyValue = splitString[2];
 471         try {
 472             getLockingServiceHandle().deQueueLockRef(keyspace, table, primaryKeyValue, lockRef,MusicUtil.getRetryCount());
 473         } catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
 474             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.DESTROYLOCK+lockRef,
 475                 ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR, e);
 476             throw new MusicLockingException(e.getMessage());
 477         }
 478         long end = System.currentTimeMillis();
 479         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to destroy lock reference:" + (end - start) + " ms");
 480     }
 481
 482     public MusicLockState destroyLockRef(String fullyQualifiedKey, String lockReference) throws MusicLockingException {
 483         long start = System.currentTimeMillis();
 484         String[] splitString = fullyQualifiedKey.split("\\.");
 485         String keyspace = splitString[0];
 486         String table = splitString[1];
 487         String primaryKeyValue = splitString[2];
 488         try {
 489             getLockingServiceHandle().deQueueLockRef(keyspace, table, primaryKeyValue, lockReference,MusicUtil.getRetryCount());
 490         } catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
 491             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.DESTROYLOCK + lockReference,
 492                 ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR,e);
 493             throw new MusicLockingException(e.getMessage());
 494         }
 495         long end = System.currentTimeMillis();
 496         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to destroy lock reference:" + (end - start) + " ms");
 497         return new MusicLockState(LockStatus.UNLOCKED, "");
 498     }
 499
 500     @Override
 501     public MusicLockState releaseLock(String lockId, boolean voluntaryRelease) throws MusicLockingException {
 502         String fullyQualifiedKey = lockId.substring(1, lockId.lastIndexOf("$"));
 503         String lockRef = lockId.substring(lockId.lastIndexOf('$')+1);
 504         if (voluntaryRelease) {
 505             return voluntaryReleaseLock(fullyQualifiedKey, lockRef);
 506         } else {
 507             return forciblyReleaseLock(fullyQualifiedKey, lockRef);
 508         }
 509     }
 510
 511     public MusicLockState voluntaryReleaseLock(String fullyQualifiedKey, String lockReference)
 512             throws MusicLockingException {
 513         MusicLockState result = null;
 514         try {
 515             result = destroyLockRef(fullyQualifiedKey, lockReference);
 516         } catch (Exception ex) {
 517             logger.info(EELFLoggerDelegate.applicationLogger,
 518                     "Exception in voluntaryReleaseLock() for " + fullyQualifiedKey + "ref: " + lockReference);
 519             throw new MusicLockingException(ex.getMessage());
 520         }
 521         return result;
 522     }
 523
 524     public MusicLockState forciblyReleaseLock(String fullyQualifiedKey, String lockReference) throws MusicLockingException {
 525         String[] splitString = fullyQualifiedKey.split("\\.");
 526         String keyspace = splitString[0];
 527         String table = splitString[1];
 528
 529         //leave a signal that this key could potentially be unsynchronized
 530         String syncTable = keyspace+".unsyncedKeys_"+table;
 531         PreparedQueryObject queryObject = new PreparedQueryObject();
 532         String values = "(?)";
 533         queryObject.addValue(fullyQualifiedKey);
 534         String insQuery = "insert into "+syncTable+" (key) values "+values+";";
 535         queryObject.appendQueryString(insQuery);
 536         try {
 537             MusicDataStoreHandle.getDSHandle().executePut(queryObject, "critical");
 538         } catch (Exception e) {
 539             logger.error("Cannot forcibly release lock: " + fullyQualifiedKey + " " + lockReference + ". "
 540                         + e.getMessage(), e);
 541         }
 542
 543         //now release the lock
 544         return destroyLockRef(fullyQualifiedKey, lockReference);
 545     }
 546
 547     @Override
 548     public List<String> releaseAllLocksForOwner(String ownerId, String keyspace, String table) throws MusicLockingException, MusicServiceException, MusicQueryException {
 549 //        System.out.println("IN RELEASEALLLOCKSFOROWNER, ");
 550
 551         List<String> lockIds = getLockingServiceHandle().getAllLocksForOwner(ownerId, keyspace, table);
 552         for (String lockId : lockIds) {
 553 //            System.out.println(" LOCKID = " + lockId);
 554             //return "$" + keyspace + "." + table + "." + lockName + "$" + String.valueOf(lockRef);
 555             releaseLock("$" + keyspace + "." + table + "." + lockId, true);
 556         }
 557         return lockIds;
 558     }
 559
 560     /**
 561      *
 562      * @param lockName
 563      * @throws MusicLockingException
 564      */
 565     @Deprecated
 566     public  void deleteLock(String lockName) throws MusicLockingException {
 567         throw new MusicLockingException("Depreciated Method Delete Lock");
 568     }
 569
 570     // Prepared Query Additions.
 571
 572     /**
 573      *
 574      * @param queryObject
 575      * @return ReturnType
 576      * @throws MusicServiceException
 577      */
 578     public  ReturnType eventualPut(PreparedQueryObject queryObject) {
 579         boolean result = false;
 580         try {
 581             result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, MusicUtil.EVENTUAL);
 582         } catch (MusicServiceException | MusicQueryException ex) {
 583             logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage(), "[ERR512E] Failed to get Lock Handle "  ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
 584             logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage() + "  " + ex.getCause() + " " + ex);
 585             return new ReturnType(ResultType.FAILURE, ex.getMessage());
 586         }
 587         if (result) {
 588             return new ReturnType(ResultType.SUCCESS, "Eventual Operation Successfully performed");
 589         } else {
 590             return new ReturnType(ResultType.FAILURE, "Eventual Operation failed to perform");
 591         }
 592     }
 593
 594     /**
 595      *
 596      * @param queryObject
 597      * @return ReturnType
 598      * @throws MusicServiceException
 599      */
 600     public  ReturnType eventualPut_nb(PreparedQueryObject queryObject,String keyspace,String tablename,String primaryKey) {
 601         boolean result = false;
 602         long guard = 0;
 603         PreparedQueryObject getGaurd = new PreparedQueryObject();
 604         getGaurd.appendQueryString("SELECT guard FROM "+keyspace+".lockq_"+tablename+ " WHERE key = ? ;");
 605         getGaurd.addValue(primaryKey);
 606         try {
 607             ResultSet getGaurdResult = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(getGaurd);
 608             Row row = getGaurdResult.one();
 609             if (row != null) {
 610                 guard = row.getLong("guard");
 611                 long timeOfWrite = System.currentTimeMillis();
 612                 long ts = MusicUtil.v2sTimeStampInMicroseconds(guard, timeOfWrite);
 613                 String query = queryObject.getQuery();
 614                 if (!queryObject.getQuery().contains("USING TIMESTAMP")) {
 615                     if (queryObject.getOperation().equalsIgnoreCase("delete"))
 616                         query = query.replaceFirst("WHERE", " USING TIMESTAMP " + ts + " WHERE ");
 617                     else
 618                         query = query.replaceFirst("SET", "USING TIMESTAMP " + ts + " SET");
 619                 }
 620                 queryObject.replaceQueryString(query);
 621             }
 622
 623         } catch (MusicServiceException | MusicQueryException e) {
 624             logger.error(EELFLoggerDelegate.applicationLogger,e.getMessage(), e);
 625         }
 626         try {
 627             result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, MusicUtil.EVENTUAL);
 628         } catch (MusicServiceException | MusicQueryException ex) {
 629             logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage(),"[ERR512E] Failed to get Lock Handle ",
 630                 ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
 631             logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage() + "  " + ex.getCause() + " ", ex);
 632             return new ReturnType(ResultType.FAILURE, ex.getMessage());
 633         }
 634         if (result) {
 635             return new ReturnType(ResultType.SUCCESS, "Eventual Operation Successfully performed");
 636         } else {
 637             return new ReturnType(ResultType.FAILURE, "Eventual Operation failed to perform");
 638         }
 639     }
 640
 641     /**
 642      *
 643      * @param keyspace
 644      * @param table
 645      * @param primaryKeyValue
 646      * @param queryObject
 647      * @param lockId
 648      * @return
 649      */
 650     public ReturnType criticalPut(String keyspace, String table, String primaryKeyValue,
 651             PreparedQueryObject queryObject, String lockId, Condition conditionInfo) {
 652         long start = System.currentTimeMillis();
 653         try {
 654             String keyLock = lockId.substring(lockId.lastIndexOf(".") + 1,lockId.lastIndexOf("$"));
 655             if (lockId.contains(".") && !keyLock.equals(primaryKeyValue)) {
 656                 return new ReturnType(ResultType.FAILURE,"Lock value '" + keyLock + "' and key value '"
 657                 + primaryKeyValue + "' not match. Please check your values: "
 658                 + lockId + " .");
 659             }
 660             LockObject lockObject = getLockingServiceHandle().getLockInfo(keyspace, table, primaryKeyValue,
 661                     lockId.substring(lockId.lastIndexOf("$") + 1));
 662
 663             if ( lockObject == null ) {
 664                 return new ReturnType(ResultType.FAILURE, lockId + " does not exist.");
 665             } else if (!lockObject.getIsLockOwner()) {
 666                 return new ReturnType(ResultType.FAILURE, lockId + " is not the lock holder");
 667             } else if (lockObject.getLocktype() != LockType.WRITE) {
 668                 return new ReturnType(ResultType.FAILURE,
 669                         "Attempting to do write operation, but " + lockId + " is a read lock");
 670             }
 671
 672             if (conditionInfo != null) {
 673                 try {
 674                     if (conditionInfo.testCondition() == false)
 675                         return new ReturnType(ResultType.FAILURE, "Lock acquired but the condition is not true");
 676                 } catch (Exception e) {
 677                     logger.error(EELFLoggerDelegate.errorLogger, e);
 678                     return new ReturnType(ResultType.FAILURE,
 679                             "Exception thrown while checking the condition, check its sanctity:\n" + e.getMessage());
 680                 }
 681             }
 682             String query = queryObject.getQuery();
 683             long timeOfWrite = System.currentTimeMillis();
 684             long lockOrdinal = Long.parseLong(lockId.substring(lockId.lastIndexOf("$") + 1));
 685             long ts = MusicUtil.v2sTimeStampInMicroseconds(lockOrdinal, timeOfWrite);
 686             // TODO: use Statement instead of modifying query
 687             if (!queryObject.getQuery().contains("USING TIMESTAMP")) {
 688                 if (queryObject.getOperation().equalsIgnoreCase("delete"))
 689                     query = query.replaceFirst("WHERE", " USING TIMESTAMP " + ts + " WHERE ");
 690                 else if (queryObject.getOperation().equalsIgnoreCase("insert"))
 691                     query = query.replaceFirst(";", " USING TIMESTAMP " + ts + " ; ");
 692                 else
 693                     query = query.replaceFirst("SET", "USING TIMESTAMP " + ts + " SET");
 694             }
 695             queryObject.replaceQueryString(query);
 696             MusicDataStore dsHandle = MusicDataStoreHandle.getDSHandle();
 697             dsHandle.executePut(queryObject, MusicUtil.CRITICAL);
 698             long end = System.currentTimeMillis();
 699             logger.info(EELFLoggerDelegate.applicationLogger,"Time taken for the critical put:" + (end - start) + " ms");
 700         } catch (MusicQueryException | MusicServiceException | MusicLockingException  e) {
 701             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), e);
 702             return new ReturnType(ResultType.FAILURE,
 703                 "Exception thrown while doing the critical put: "
 704                 + e.getMessage());
 705         }
 706         return new ReturnType(ResultType.SUCCESS, "Update performed");
 707     }
 708
 709
 710     /**
 711      *
 712      * @param queryObject
 713      * @param consistency
 714      * @return Boolean Indicates success or failure
 715      * @throws MusicServiceException
 716      *
 717      *
 718      */
 719     public ResultType nonKeyRelatedPut(PreparedQueryObject queryObject, String consistency) throws MusicServiceException,MusicQueryException {
 720         // this is mainly for some functions like keyspace creation etc which does not
 721         // really need the bells and whistles of Music locking.
 722         boolean result = false;
 723 //        try {
 724         result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, consistency);
 725 //        } catch (MusicQueryException | MusicServiceException ex) {
 726             // logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(), AppMessages.UNKNOWNERROR,
 727             //     ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR, ex);
 728 //            throw new MusicServiceException(ex.getMessage(),ex);
 729 //        }
 730         return result ? ResultType.SUCCESS : ResultType.FAILURE;
 731     }
 732
 733     /**
 734      * This method performs DDL operation on cassandra.
 735      *
 736      * @param queryObject query object containing prepared query and values
 737      * @return ResultSet
 738      * @throws MusicServiceException
 739      */
 740     public ResultSet get(PreparedQueryObject queryObject) throws MusicServiceException {
 741         ResultSet results = null;
 742         try {
 743             results = MusicDataStoreHandle.getDSHandle().executeOneConsistencyGet(queryObject);
 744         } catch (MusicQueryException | MusicServiceException e) {
 745             logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), e);
 746             throw new MusicServiceException(e.getMessage());
 747         }
 748         return results;
 749     }
 750
 751     /**
 752      * This method performs DDL operations on cassandra, if the the resource is available. Lock ID
 753      * is used to check if the resource is free.
 754      *
 755      * @param keyspace name of the keyspace
 756      * @param table name of the table
 757      * @param primaryKeyValue primary key value
 758      * @param queryObject query object containing prepared query and values
 759      * @param lockId lock ID to check if the resource is free to perform the operation.
 760      * @return ResultSet
 761      */
 762     public ResultSet criticalGet(String keyspace, String table, String primaryKeyValue,
 763                     PreparedQueryObject queryObject, String lockId) throws MusicServiceException {
 764         ResultSet results = null;
 765         String keyLock = lockId.substring(lockId.lastIndexOf(".") + 1,lockId.lastIndexOf("$"));
 766         try {
 767             if (lockId.contains(".") && !keyLock.equals(primaryKeyValue)) {
 768                 throw new MusicLockingException("Lock value '" + keyLock + "' and key value '"
 769                 + primaryKeyValue + "' do not match. Please check your values: "
 770                 + lockId + " .");
 771             }
 772             LockObject lockObject = getLockingServiceHandle().getLockInfo(keyspace, table, primaryKeyValue,
 773                 lockId.substring(lockId.lastIndexOf("$") + 1));
 774             if (null == lockObject) {
 775                 throw new MusicLockingException("No Lock Object. Please check if lock name or key is correct."
 776                     + lockId + " .");
 777             }
 778             if ( !lockObject.getIsLockOwner()) {
 779                 return null;// not top of the lock store q
 780             }
 781             results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(queryObject);
 782         } catch ( MusicLockingException e ) {
 783             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.UNKNOWNERROR  ,ErrorSeverity
 784                     .WARN, ErrorTypes.MUSICSERVICEERROR);
 785                 throw new MusicServiceException(
 786                     "Cannot perform critical get for key: " + primaryKeyValue + " : " + e.getMessage());
 787         } catch (MusicQueryException | MusicServiceException e) {
 788             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.UNKNOWNERROR  ,ErrorSeverity
 789                     .WARN, ErrorTypes.MUSICSERVICEERROR, e);
 790                 throw new MusicServiceException(
 791                     "Cannot perform critical get for key: " + primaryKeyValue + " : " + e.getMessage());
 792         }
 793         return results;
 794     }
 795
 796     /**
 797      * This method performs DML operation on cassandra, when the lock of the dd is acquired.
 798      *
 799      * @param keyspaceName name of the keyspace
 800      * @param tableName name of the table
 801      * @param primaryKey primary key value
 802      * @param queryObject query object containing prepared query and values
 803      * @return ReturnType
 804      * @throws MusicLockingException
 805      * @throws MusicServiceException
 806      * @throws MusicQueryException
 807      */
 808     public ReturnType atomicPut(String keyspaceName, String tableName, String primaryKey,
 809             PreparedQueryObject queryObject, Condition conditionInfo)
 810             throws MusicLockingException, MusicQueryException, MusicServiceException {
 811         long start = System.currentTimeMillis();
 812         String fullyQualifiedKey = keyspaceName + "." + tableName + "." + primaryKey;
 813         String lockId = createLockReferenceAtomic(fullyQualifiedKey, LockType.WRITE);
 814         long lockCreationTime = System.currentTimeMillis();
 815         ReturnType lockAcqResult = null;
 816         logger.info(EELFLoggerDelegate.applicationLogger,
 817                 "***Acquiring lock for atomicPut() query : " + queryObject.getQuery() + " : " + primaryKey);
 818         logger.info(EELFLoggerDelegate.applicationLogger,
 819                 "***Acquiring lock for atomicPut() values: " + queryObject.getValues().toString());
 820         if (conditionInfo != null) {
 821             logger.info(EELFLoggerDelegate.applicationLogger,
 822                     "***Acquiring lock for atomicPut() conditions: " + conditionInfo.toString());
 823         }
 824         try {
 825             lockAcqResult = acquireLockWithLease(fullyQualifiedKey, lockId, MusicUtil.getDefaultLockLeasePeriod());
 826         } catch (MusicLockingException ex) {
 827             logger.error(EELFLoggerDelegate.errorLogger,
 828                     "Exception while acquireLockWithLease() in atomic put for key: " + primaryKey);
 829             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage());
 830             throw new MusicServiceException(
 831                     "Cannot perform atomic put for key: " + primaryKey + " : " + ex.getMessage());
 832         }
 833         long lockAcqTime = System.currentTimeMillis();
 834
 835         /*
 836          * if (!lockAcqResult.getResult().equals(ResultType.SUCCESS)) { logger.info(EELFLoggerDelegate.
 837          * applicationLogger,"unable to acquire lock, id " + lockId);
 838          * voluntaryReleaseLock(fullyQualifiedKey,lockId); return lockAcqResult; }
 839          */
 840
 841         logger.info(EELFLoggerDelegate.applicationLogger, "acquired lock with id " + lockId);
 842         String lockRef = lockId.substring(lockId.lastIndexOf("$"));
 843         ReturnType criticalPutResult = null;
 844         if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
 845             criticalPutResult = criticalPut(keyspaceName, tableName, primaryKey, queryObject, lockRef, conditionInfo);
 846             long criticalPutTime = System.currentTimeMillis();
 847             long lockDeleteTime = System.currentTimeMillis();
 848             String timingInfo = "|lock creation time:" + (lockCreationTime - start) + "|lock accquire time:"
 849                     + (lockAcqTime - lockCreationTime) + "|critical put time:" + (criticalPutTime - lockAcqTime)
 850                     + "|lock delete time:" + (lockDeleteTime - criticalPutTime) + "|";
 851             criticalPutResult.setTimingInfo(timingInfo);
 852         } else {
 853             logger.info(EELFLoggerDelegate.applicationLogger, "unable to acquire lock, id " + lockId);
 854             criticalPutResult = lockAcqResult;
 855         }
 856         try {
 857             voluntaryReleaseLock(fullyQualifiedKey, lockId);
 858         } catch (MusicLockingException ex) {
 859             logger.info(EELFLoggerDelegate.applicationLogger,
 860                     "Exception occured while deleting lock after atomic put for key: " + primaryKey);
 861             criticalPutResult.setMessage(criticalPutResult.getMessage() + "Lock release failed");
 862         }
 863         return criticalPutResult;
 864     }
 865
 866
 867
 868     /**
 869      * This method performs DDL operation on cassasndra, when the lock for the resource is acquired.
 870      *
 871      * @param keyspaceName name of the keyspace
 872      * @param tableName name of the table
 873      * @param primaryKey primary key value
 874      * @param queryObject query object containing prepared query and values
 875      * @return ResultSet
 876      * @throws MusicServiceException
 877      * @throws MusicLockingException
 878      * @throws MusicQueryException
 879      */
 880     public ResultSet atomicGet(String keyspaceName, String tableName, String primaryKey,
 881             PreparedQueryObject queryObject) throws MusicServiceException, MusicLockingException, MusicQueryException {
 882         String fullyQualifiedKey = keyspaceName + "." + tableName + "." + primaryKey;
 883         String lockId = createLockReferenceAtomic(fullyQualifiedKey, LockType.READ);
 884         ReturnType lockAcqResult = null;
 885         ResultSet result = null;
 886         logger.info(EELFLoggerDelegate.applicationLogger, "Acquiring lock for atomicGet() : " + queryObject.getQuery());
 887         try {
 888             lockAcqResult = acquireLockWithLease(fullyQualifiedKey, lockId, MusicUtil.getDefaultLockLeasePeriod());
 889         } catch (MusicLockingException ex) {
 890             logger.error(EELFLoggerDelegate.errorLogger,
 891                     "Exception while acquireLockWithLease() in atomic get for key: " + primaryKey);
 892             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage());
 893             throw new MusicServiceException(
 894                     "Cannot perform atomic get for key: " + primaryKey + " : " + ex.getMessage());
 895         }
 896         if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
 897             logger.info(EELFLoggerDelegate.applicationLogger, "acquired lock with id " + lockId);
 898             String lockRef = lockId.substring(lockId.lastIndexOf("$"));
 899             result = criticalGet(keyspaceName, tableName, primaryKey, queryObject, lockRef);
 900         } else {
 901             logger.info(EELFLoggerDelegate.applicationLogger, "unable to acquire lock, id " + lockId);
 902         }
 903         try {
 904             voluntaryReleaseLock(fullyQualifiedKey, lockId);
 905         } catch (MusicLockingException ex) {
 906             logger.info(EELFLoggerDelegate.applicationLogger,
 907                     "Exception occured while deleting lock after atomic put for key: " + primaryKey);
 908             throw new MusicLockingException(ex.getMessage());
 909         }
 910
 911         return result;
 912     }
 913
 914
 915
 916     /**
 917      * @param lockName
 918      * @return
 919      */
 920     public Map<String, Object> validateLock(String lockName) {
 921         return MusicUtil.validateLock(lockName);
 922     }
 923
 924     @Override
 925     @Deprecated
 926     public ReturnType atomicPutWithDeleteLock(String keyspaceName, String tableName, String primaryKey,
 927         PreparedQueryObject queryObject, Condition conditionInfo) throws MusicLockingException {
 928         return null;
 929     }
 930
 931     @Override
 932     public List<String> getLockQueue(String fullyQualifiedKey)
 933         throws MusicServiceException, MusicQueryException, MusicLockingException {
 934         String[] splitString = fullyQualifiedKey.split("\\.");
 935         String keyspace = splitString[0];
 936         String table = splitString[1];
 937         String primaryKeyValue = splitString[2];
 938
 939         return getLockingServiceHandle().getLockQueue(keyspace, table, primaryKeyValue);
 940     }
 941     @Override
 942     public long getLockQueueSize(String fullyQualifiedKey)
 943         throws MusicServiceException, MusicQueryException, MusicLockingException {
 944         String[] splitString = fullyQualifiedKey.split("\\.");
 945         String keyspace = splitString[0];
 946         String table = splitString[1];
 947         String primaryKeyValue = splitString[2];
 948
 949         return getLockingServiceHandle().getLockQueueSize(keyspace, table, primaryKeyValue);
 950     }
 951
 952     @Override
 953     @Deprecated
 954     public ResultSet atomicGetWithDeleteLock(String keyspaceName, String tableName, String primaryKey,
 955             PreparedQueryObject queryObject) throws MusicServiceException, MusicLockingException {
 956         //deprecated
 957         return null;
 958     }
 959
 960     //Methods added for ORM changes
 961
 962     public ResultType createKeyspace(JsonKeySpace jsonKeySpaceObject,String consistencyInfo)
 963             throws MusicServiceException,MusicQueryException {
 964         ResultType result = nonKeyRelatedPut(jsonKeySpaceObject.genCreateKeyspaceQuery(), consistencyInfo);
 965         logger.info(EELFLoggerDelegate.applicationLogger, " Keyspace Creation Process completed successfully");
 966
 967         return result;
 968     }
 969
 970     public ResultType dropKeyspace(JsonKeySpace jsonKeySpaceObject, String consistencyInfo)
 971             throws MusicServiceException,MusicQueryException {
 972         ResultType result = nonKeyRelatedPut(jsonKeySpaceObject.genDropKeyspaceQuery(),
 973                     consistencyInfo);
 974         logger.info(EELFLoggerDelegate.applicationLogger, " Keyspace deletion Process completed successfully");
 975         return result;
 976     }
 977
 978     public ResultType createTable(JsonTable jsonTableObject, String consistencyInfo)
 979             throws MusicServiceException, MusicQueryException {
 980         ResultType result = null;
 981         try {
 982             result = createTable(jsonTableObject.getKeyspaceName(),
 983                     jsonTableObject.getTableName(), jsonTableObject.genCreateTableQuery(), consistencyInfo);
 984
 985         } catch (MusicServiceException ex) {
 986             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(), AppMessages.UNKNOWNERROR, ErrorSeverity.WARN,
 987                     ErrorTypes.MUSICSERVICEERROR);
 988             throw new MusicServiceException(ex.getMessage());
 989         }
 990         logger.info(EELFLoggerDelegate.applicationLogger, " Table Creation Process completed successfully ");
 991         return result;
 992     }
 993
 994     public ResultType dropTable(JsonTable jsonTableObject,String consistencyInfo)
 995             throws MusicServiceException,MusicQueryException {
 996         ResultType result = nonKeyRelatedPut(jsonTableObject.genDropTableQuery(),
 997                     consistencyInfo);
 998         logger.info(EELFLoggerDelegate.applicationLogger, " Table deletion Process completed successfully ");
 999
1000         return result;
1001     }
1002
1003     @Override
1004     public ResultType createIndex(JsonIndex jsonIndexObject, String consistencyInfo)
1005             throws MusicServiceException, MusicQueryException{
1006         ResultType result = nonKeyRelatedPut(jsonIndexObject.genCreateIndexQuery(),
1007                     consistencyInfo);
1008
1009         logger.info(EELFLoggerDelegate.applicationLogger, " Index creation Process completed successfully ");
1010         return result;
1011     }
1012
1013     /**
1014      * This method performs DDL operation on cassandra.
1015      *
1016      * @param queryObject query object containing prepared query and values
1017      * @return ResultSet
1018      * @throws MusicServiceException
1019      */
1020     public  ResultSet select(JsonSelect jsonSelect, MultivaluedMap<String, String> rowParams)
1021             throws MusicServiceException, MusicQueryException {
1022         ResultSet results = null;
1023         try {
1024             results = get(jsonSelect.genSelectQuery(rowParams));
1025         } catch (MusicServiceException e) {
1026             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage());
1027             throw new MusicServiceException(e.getMessage());
1028         }
1029         return results;
1030     }
1031
1032     /**
1033      * Select Critical
1034      */
1035     public ResultSet selectCritical(JsonInsert jsonInsertObj, MultivaluedMap<String, String> rowParams)
1036             throws MusicLockingException, MusicQueryException, MusicServiceException {
1037
1038         ResultSet results = null;
1039         String consistency = "";
1040         if(null != jsonInsertObj && null != jsonInsertObj.getConsistencyInfo()) {
1041             consistency = jsonInsertObj.getConsistencyInfo().get("type");
1042         }
1043
1044         String lockId = jsonInsertObj.getConsistencyInfo().get("lockId");
1045
1046         PreparedQueryObject queryObject = jsonInsertObj.genSelectCriticalPreparedQueryObj(rowParams);
1047
1048         if (consistency.equalsIgnoreCase(MusicUtil.CRITICAL)) {
1049             results = criticalGet(jsonInsertObj.getKeyspaceName(), jsonInsertObj.getTableName(),
1050                     jsonInsertObj.getPrimaryKeyVal(), queryObject,lockId);
1051         } else if (consistency.equalsIgnoreCase(MusicUtil.ATOMIC)) {
1052             results = atomicGet(jsonInsertObj.getKeyspaceName(), jsonInsertObj.getTableName(),
1053                     jsonInsertObj.getPrimaryKeyVal(), queryObject);
1054         }
1055
1056         return results;
1057     }
1058
1059     /**
1060      * this is insert row into Table
1061      */
1062     public ReturnType insertIntoTable(JsonInsert jsonInsertObj)
1063             throws MusicLockingException, MusicQueryException, MusicServiceException {
1064
1065         String consistency = "";
1066         if(null != jsonInsertObj && null != jsonInsertObj.getConsistencyInfo()) {
1067             consistency = jsonInsertObj.getConsistencyInfo().get("type");
1068         }
1069
1070         ReturnType result = null;
1071
1072         try {
1073             PreparedQueryObject queryObj = null;
1074             queryObj = jsonInsertObj.genInsertPreparedQueryObj();
1075
1076             if (consistency.equalsIgnoreCase(MusicUtil.EVENTUAL)) {
1077                 result = eventualPut(jsonInsertObj.genInsertPreparedQueryObj());
1078             } else if (consistency.equalsIgnoreCase(MusicUtil.CRITICAL)) {
1079                 String lockId = jsonInsertObj.getConsistencyInfo().get("lockId");
1080                 if(lockId == null) {
1081                     logger.error(EELFLoggerDelegate.errorLogger,"LockId cannot be null. Create lock reference or"
1082                             + " use ATOMIC instead of CRITICAL", ErrorSeverity.FATAL, ErrorTypes.MUSICSERVICEERROR);
1083                     return new ReturnType(ResultType.FAILURE, "LockId cannot be null. Create lock "
1084                             + "and acquire lock or use ATOMIC instead of CRITICAL");
1085                 }
1086                 result = criticalPut(jsonInsertObj.getKeyspaceName(),
1087                         jsonInsertObj.getTableName(), jsonInsertObj.getPrimaryKeyVal(), jsonInsertObj.genInsertPreparedQueryObj(), lockId,null);
1088             } else if (consistency.equalsIgnoreCase(MusicUtil.ATOMIC)) {
1089                 result = atomicPut(jsonInsertObj.getKeyspaceName(), jsonInsertObj.getTableName(),
1090                         jsonInsertObj.getPrimaryKeyVal(), jsonInsertObj.genInsertPreparedQueryObj(), null);
1091             }
1092         } catch (Exception ex) {
1093             logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage(), AppMessages.UNKNOWNERROR  ,ErrorSeverity
1094                 .WARN, ErrorTypes.MUSICSERVICEERROR, ex);
1095             return new ReturnType(ResultType.FAILURE, ex.getMessage());
1096         }
1097
1098         return result;
1099     }
1100
1101      /**
1102      * This is insert row into Table
1103      */
1104     public ReturnType updateTable(JsonUpdate jsonUpdateObj, MultivaluedMap<String, String> rowParams)
1105             throws MusicLockingException, MusicQueryException, MusicServiceException {
1106
1107         ReturnType result = null;
1108         String consistency = "";
1109         if(null != jsonUpdateObj && null != jsonUpdateObj.getConsistencyInfo()) {
1110             consistency = jsonUpdateObj.getConsistencyInfo().get("type");
1111         }
1112         PreparedQueryObject queryObject = jsonUpdateObj.genUpdatePreparedQueryObj(rowParams);
1113
1114         Condition conditionInfo;
1115         if (jsonUpdateObj.getConditions() == null) {
1116             conditionInfo = null;
1117         } else {
1118             // to avoid parsing repeatedly, just send the select query to obtain row
1119             PreparedQueryObject selectQuery = new PreparedQueryObject();
1120             selectQuery.appendQueryString("SELECT *  FROM " + jsonUpdateObj.getKeyspaceName() + "." + jsonUpdateObj.getTableName() + " WHERE "
1121                 + jsonUpdateObj.getRowIdString() + ";");
1122             selectQuery.addValue(jsonUpdateObj.getPrimarKeyValue());
1123             conditionInfo = new Condition(jsonUpdateObj.getConditions(), selectQuery);
1124         }
1125
1126
1127         if (consistency.equalsIgnoreCase(MusicUtil.EVENTUAL)) {
1128             result = eventualPut(queryObject);
1129         } else if (consistency.equalsIgnoreCase(MusicUtil.CRITICAL)) {
1130             String lockId = jsonUpdateObj.getConsistencyInfo().get("lockId");
1131             if(lockId == null) {
1132                 logger.error(EELFLoggerDelegate.errorLogger,"LockId cannot be null. Create lock reference or"
1133                         + " use ATOMIC instead of CRITICAL", ErrorSeverity.FATAL, ErrorTypes.MUSICSERVICEERROR);
1134
1135                 return new ReturnType(ResultType.FAILURE, "LockId cannot be null. Create lock "
1136                         + "and acquire lock or use ATOMIC instead of CRITICAL");
1137             }
1138             result = criticalPut(jsonUpdateObj.getKeyspaceName(), jsonUpdateObj.getTableName(), jsonUpdateObj.getPrimarKeyValue(),
1139                             queryObject, lockId, conditionInfo);
1140         } else if (consistency.equalsIgnoreCase("atomic_delete_lock")) {
1141             // this function is mainly for the benchmarks
1142             try {
1143                 result = atomicPutWithDeleteLock(jsonUpdateObj.getKeyspaceName(), jsonUpdateObj.getTableName(),
1144                         jsonUpdateObj.getPrimarKeyValue(), queryObject, conditionInfo);
1145             } catch (MusicLockingException e) {
1146                 logger.error(EELFLoggerDelegate.errorLogger,e, AppMessages.UNKNOWNERROR  ,ErrorSeverity.WARN,
1147                     ErrorTypes.GENERALSERVICEERROR, e);
1148                 throw new MusicLockingException(AppMessages.UNKNOWNERROR.toString());
1149
1150             }
1151         } else if (consistency.equalsIgnoreCase(MusicUtil.ATOMIC)) {
1152             try {
1153                 result = atomicPut(jsonUpdateObj.getKeyspaceName(), jsonUpdateObj.getTableName(), jsonUpdateObj.getPrimarKeyValue(),
1154                     queryObject, conditionInfo);
1155             } catch (MusicLockingException e) {
1156                 logger.error(EELFLoggerDelegate.errorLogger,e, AppMessages.UNKNOWNERROR  ,ErrorSeverity.WARN, ErrorTypes.GENERALSERVICEERROR, e);
1157                 throw new MusicLockingException(AppMessages.UNKNOWNERROR.toString());
1158             }
1159         } else if (consistency.equalsIgnoreCase(MusicUtil.EVENTUAL_NB)) {
1160             try {
1161                 result = eventualPut_nb(queryObject, jsonUpdateObj.getKeyspaceName(),
1162                         jsonUpdateObj.getTableName(), jsonUpdateObj.getPrimarKeyValue());
1163             }catch (Exception e) {
1164                 return new ReturnType(ResultType.FAILURE, e.getMessage());
1165             }
1166
1167         }
1168
1169         return result;
1170     }
1171
1172     /**
1173      * This method is for Delete From Table
1174      */
1175     public ReturnType deleteFromTable(JsonDelete jsonDeleteObj, MultivaluedMap<String, String> rowParams)
1176             throws MusicLockingException, MusicQueryException, MusicServiceException {
1177
1178         ReturnType result = null;
1179         String consistency = "";
1180         if(null != jsonDeleteObj && null != jsonDeleteObj.getConsistencyInfo()) {
1181             consistency = jsonDeleteObj.getConsistencyInfo().get("type");
1182         }
1183         PreparedQueryObject queryObject = jsonDeleteObj.genDeletePreparedQueryObj(rowParams);
1184
1185         // get the conditional, if any
1186         Condition conditionInfo;
1187         if (jsonDeleteObj.getConditions() == null) {
1188             conditionInfo = null;
1189         } else {
1190             // to avoid parsing repeatedly, just send the select query to obtain row
1191             PreparedQueryObject selectQuery = new PreparedQueryObject();
1192             selectQuery.appendQueryString("SELECT *  FROM " + jsonDeleteObj.getKeyspaceName() + "." + jsonDeleteObj.getTableName() + " WHERE "
1193                 + jsonDeleteObj.getRowIdString() + ";");
1194             selectQuery.addValue(jsonDeleteObj.getPrimarKeyValue());
1195             conditionInfo = new Condition(jsonDeleteObj.getConditions(), selectQuery);
1196         }
1197
1198         if (consistency.equalsIgnoreCase(MusicUtil.EVENTUAL))
1199             result = eventualPut(queryObject);
1200         else if (consistency.equalsIgnoreCase(MusicUtil.CRITICAL)) {
1201             String lockId = jsonDeleteObj.getConsistencyInfo().get("lockId");
1202             if(lockId == null) {
1203                 logger.error(EELFLoggerDelegate.errorLogger,"LockId cannot be null. Create lock reference or"
1204                     + " use ATOMIC instead of CRITICAL", ErrorSeverity.FATAL, ErrorTypes.MUSICSERVICEERROR);
1205
1206                 return new ReturnType(ResultType.FAILURE, "LockId cannot be null. Create lock "
1207                         + "and acquire lock or use ATOMIC instead of CRITICAL");
1208             }
1209             result = criticalPut(jsonDeleteObj.getKeyspaceName(),
1210                     jsonDeleteObj.getTableName(), jsonDeleteObj.getPrimarKeyValue(),
1211                 queryObject, lockId, conditionInfo);
1212         } else if (consistency.equalsIgnoreCase(MusicUtil.ATOMIC)) {
1213             result = atomicPut(jsonDeleteObj.getKeyspaceName(),
1214                     jsonDeleteObj.getTableName(), jsonDeleteObj.getPrimarKeyValue(),
1215                 queryObject, conditionInfo);
1216         } else if(consistency.equalsIgnoreCase(MusicUtil.EVENTUAL_NB)) {
1217             result = eventualPut_nb(queryObject, jsonDeleteObj.getKeyspaceName(),
1218                     jsonDeleteObj.getTableName(), jsonDeleteObj.getPrimarKeyValue());
1219         }
1220
1221         return result;
1222     }
1223
1224
1225 }