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.List;
  30 import java.util.Map;
  31 import java.util.StringTokenizer;
  32
  33 import org.onap.music.datastore.MusicDataStore;
  34 import org.onap.music.datastore.MusicDataStoreHandle;
  35 import org.onap.music.datastore.PreparedQueryObject;
  36 import org.onap.music.eelf.logging.EELFLoggerDelegate;
  37 import org.onap.music.eelf.logging.format.AppMessages;
  38 import org.onap.music.eelf.logging.format.ErrorSeverity;
  39 import org.onap.music.eelf.logging.format.ErrorTypes;
  40 import org.onap.music.exceptions.MusicLockingException;
  41 import org.onap.music.exceptions.MusicQueryException;
  42 import org.onap.music.exceptions.MusicServiceException;
  43 import org.onap.music.lockingservice.cassandra.CassaLockStore;
  44 import org.onap.music.lockingservice.cassandra.CassaLockStore.LockObject;
  45 import org.onap.music.lockingservice.cassandra.MusicLockState;
  46 import org.onap.music.lockingservice.cassandra.MusicLockState.LockStatus;
  47 import org.onap.music.main.MusicUtil;
  48 import org.onap.music.main.ResultType;
  49 import org.onap.music.main.ReturnType;
  50 import org.onap.music.service.MusicCoreService;
  51
  52 import com.datastax.driver.core.DataType;
  53 import com.datastax.driver.core.ResultSet;
  54 import com.datastax.driver.core.Row;
  55 import com.datastax.driver.core.TableMetadata;
  56
  57 import org.onap.music.datastore.*;
  58
  59 public class MusicCassaCore implements MusicCoreService {
  60
  61     public static CassaLockStore mLockHandle = null;;
  62     private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(MusicCassaCore.class);
  63     private static boolean unitTestRun=true;
  64     private static MusicCassaCore musicCassaCoreInstance = null;
  65
  66     private MusicCassaCore() {
  67
  68     }
  69     public static MusicCassaCore getInstance() {
  70
  71         if(musicCassaCoreInstance == null) {
  72             musicCassaCoreInstance = new MusicCassaCore();
  73         }
  74         return musicCassaCoreInstance;
  75     }
  76
  77     public static CassaLockStore getLockingServiceHandle() throws MusicLockingException {
  78         logger.info(EELFLoggerDelegate.applicationLogger,"Acquiring lock store handle");
  79         long start = System.currentTimeMillis();
  80
  81         if (mLockHandle == null) {
  82             try {
  83                 mLockHandle = new CassaLockStore(MusicDataStoreHandle.getDSHandle());
  84             } catch (Exception e) {
  85                 logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKHANDLE,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
  86                 throw new MusicLockingException("Failed to aquire Locl store handle " + e);
  87             }
  88         }
  89         long end = System.currentTimeMillis();
  90         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to acquire lock store handle:" + (end - start) + " ms");
  91         return mLockHandle;
  92     }
  93
  94
  95
  96     public  String createLockReference(String fullyQualifiedKey) {
  97         String[] splitString = fullyQualifiedKey.split("\\.");
  98         String keyspace = splitString[0];
  99         String table = splitString[1];
 100         String lockName = splitString[2];
 101
 102         logger.info(EELFLoggerDelegate.applicationLogger,"Creating lock reference for lock name:" + lockName);
 103         long start = System.currentTimeMillis();
 104         String lockReference = null;
 105         try {
 106             lockReference = "" + getLockingServiceHandle().genLockRefandEnQueue(keyspace, table, lockName);
 107         } catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
 108             e.printStackTrace();
 109         }
 110         long end = System.currentTimeMillis();
 111         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to create lock reference:" + (end - start) + " ms");
 112         return lockReference;
 113     }
 114
 115
 116     public  ReturnType acquireLockWithLease(String fullyQualifiedKey, String lockReference, long leasePeriod) throws MusicLockingException, MusicQueryException, MusicServiceException  {
 117          evictExpiredLockHolder(fullyQualifiedKey,leasePeriod);
 118             return acquireLock(fullyQualifiedKey, lockReference);
 119     }
 120
 121     private  void evictExpiredLockHolder(String fullyQualifiedKey, long leasePeriod) throws MusicLockingException, MusicQueryException, MusicServiceException {
 122
 123         String[] splitString = fullyQualifiedKey.split("\\.");
 124         String keyspace = splitString[0];
 125         String table = splitString[1];
 126         String primaryKeyValue = splitString[2];
 127
 128         LockObject currentLockHolderObject = getLockingServiceHandle().peekLockQueue(keyspace, table, primaryKeyValue);
 129
 130         /* Release the lock of the previous holder if it has expired. if the update to the acquire time has not reached due to network delays, simply use the create time as the
 131          * reference*/
 132
 133         long referenceTime = Math.max(Long.parseLong(currentLockHolderObject.acquireTime), Long.parseLong(currentLockHolderObject.createTime));
 134         if((System.currentTimeMillis() - referenceTime) > leasePeriod) {
 135             forciblyReleaseLock(fullyQualifiedKey,  currentLockHolderObject.lockRef+"");
 136             logger.info(EELFLoggerDelegate.applicationLogger, currentLockHolderObject.lockRef+" forcibly released");
 137         }
 138     }
 139
 140     private static ReturnType isTopOfLockStore(String keyspace, String table, String primaryKeyValue, String lockReference) throws MusicLockingException, MusicQueryException, MusicServiceException {
 141
 142         //return failure to lock holders too early or already evicted from the lock store
 143         String topOfLockStoreS = getLockingServiceHandle().peekLockQueue(keyspace, table, primaryKeyValue).lockRef;
 144         long topOfLockStoreL = Long.parseLong(topOfLockStoreS);
 145         long lockReferenceL = Long.parseLong(lockReference);
 146
 147         if(lockReferenceL > topOfLockStoreL) {
 148             logger.info(EELFLoggerDelegate.applicationLogger, lockReference+" is not the lock holder yet");
 149             return new ReturnType(ResultType.FAILURE, lockReference+" is not the lock holder yet");
 150         }
 151
 152
 153         if(lockReferenceL < topOfLockStoreL) {
 154             logger.info(EELFLoggerDelegate.applicationLogger, lockReference+" is no longer/or was never in the lock store queue");
 155             return new ReturnType(ResultType.FAILURE, lockReference+" is no longer/or was never in the lock store queue");
 156            }
 157
 158            return new ReturnType(ResultType.SUCCESS, lockReference+" is top of lock store");
 159     }
 160
 161     public  ReturnType acquireLock(String fullyQualifiedKey, String lockId)
 162             throws MusicLockingException, MusicQueryException, MusicServiceException {
 163         String[] splitString = lockId.split("\\.");
 164         String keyspace = splitString[0].substring(1);//remove '$'
 165         String table = splitString[1];
 166         String primaryKeyValue = splitString[2].substring(0, splitString[2].lastIndexOf("$"));
 167         String localFullyQualifiedKey = lockId.substring(1, lockId.lastIndexOf("$"));
 168         String lockRef = lockId.substring(lockId.lastIndexOf("$")+1); //lockRef is "$" to end
 169
 170         ReturnType result = isTopOfLockStore(keyspace, table, primaryKeyValue, lockRef);
 171
 172         if(result.getResult().equals(ResultType.FAILURE))
 173                 return result;//not top of the lock store q
 174
 175         //check to see if the value of the key has to be synced in case there was a forceful release
 176         String syncTable = keyspace+".unsyncedKeys_"+table;
 177         String query = "select * from "+syncTable+" where key='"+localFullyQualifiedKey+"';";
 178         PreparedQueryObject readQueryObject = new PreparedQueryObject();
 179         readQueryObject.appendQueryString(query);
 180         ResultSet results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(readQueryObject);
 181         if (results.all().size() != 0) {
 182             logger.info("In acquire lock: Since there was a forcible release, need to sync quorum!");
 183             try {
 184                 syncQuorum(keyspace, table, primaryKeyValue);
 185             } catch (Exception e) {
 186                 StringWriter sw = new StringWriter();
 187                    logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), "[ERR506E] Failed to aquire lock ",ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 188                 String exceptionAsString = sw.toString();
 189                 return new ReturnType(ResultType.FAILURE, "Exception thrown while syncing key:\n" + exceptionAsString);
 190             }
 191             String cleanQuery = "delete  from music_internal.unsynced_keys where key='"+localFullyQualifiedKey+"';";
 192             PreparedQueryObject deleteQueryObject = new PreparedQueryObject();
 193             deleteQueryObject.appendQueryString(cleanQuery);
 194             MusicDataStoreHandle.getDSHandle().executePut(deleteQueryObject, "critical");
 195         }
 196
 197         getLockingServiceHandle().updateLockAcquireTime(keyspace, table, primaryKeyValue, lockRef);
 198
 199         return new ReturnType(ResultType.SUCCESS, lockRef+" is the lock holder for the key");
 200     }
 201
 202
 203
 204     /**
 205      *
 206      * @param tableQueryObject
 207      * @param consistency
 208      * @return Boolean Indicates success or failure
 209      * @throws MusicServiceException
 210      *
 211      *
 212      */
 213     public  ResultType createTable(String keyspace, String table, PreparedQueryObject tableQueryObject, String consistency) throws MusicServiceException {
 214             boolean result = false;
 215
 216             try {
 217                 //create shadow locking table
 218                 result = getLockingServiceHandle().createLockQueue(keyspace, table);
 219                 if(result == false)
 220                     return ResultType.FAILURE;
 221
 222                 result = false;
 223
 224                 //create table to track unsynced_keys
 225                 table = "unsyncedKeys_"+table;
 226
 227                 String tabQuery = "CREATE TABLE IF NOT EXISTS "+keyspace+"."+table
 228                         + " ( key text,PRIMARY KEY (key) );";
 229                 System.out.println(tabQuery);
 230                 PreparedQueryObject queryObject = new PreparedQueryObject();
 231
 232                 queryObject.appendQueryString(tabQuery);
 233                 result = false;
 234                 result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, "eventual");
 235
 236
 237                 //create actual table
 238                 result = MusicDataStoreHandle.getDSHandle().executePut(tableQueryObject, consistency);
 239             } catch (MusicQueryException | MusicServiceException | MusicLockingException ex) {
 240                 logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage(), AppMessages.UNKNOWNERROR  ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
 241                 throw new MusicServiceException(ex.getMessage());
 242             }
 243             return result?ResultType.SUCCESS:ResultType.FAILURE;
 244     }
 245
 246     private static void syncQuorum(String keyspace, String table, String primaryKeyValue) throws Exception {
 247         logger.info(EELFLoggerDelegate.applicationLogger,"Performing sync operation---");
 248         PreparedQueryObject selectQuery = new PreparedQueryObject();
 249         PreparedQueryObject updateQuery = new PreparedQueryObject();
 250
 251         // get the primary key d
 252         TableMetadata tableInfo = MusicDataStoreHandle.returnColumnMetadata(keyspace, table);
 253         String primaryKeyName = tableInfo.getPrimaryKey().get(0).getName();// we only support single
 254                                                                            // primary key
 255         DataType primaryKeyType = tableInfo.getPrimaryKey().get(0).getType();
 256         Object cqlFormattedPrimaryKeyValue =
 257                         MusicUtil.convertToActualDataType(primaryKeyType, primaryKeyValue);
 258
 259         // get the row of data from a quorum
 260         selectQuery.appendQueryString("SELECT *  FROM " + keyspace + "." + table + " WHERE "
 261                         + primaryKeyName + "= ?" + ";");
 262         selectQuery.addValue(cqlFormattedPrimaryKeyValue);
 263         MusicUtil.writeBackToQuorum(selectQuery, primaryKeyName, updateQuery, keyspace, table,
 264             cqlFormattedPrimaryKeyValue);
 265
 266     }
 267
 268     /**
 269      *
 270      * @param query
 271      * @return ResultSet
 272      */
 273     public  ResultSet quorumGet(PreparedQueryObject query) {
 274         ResultSet results = null;
 275         try {
 276             results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(query);
 277         } catch (MusicServiceException | MusicQueryException e) {
 278             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.UNKNOWNERROR ,ErrorSeverity.MAJOR, ErrorTypes.GENERALSERVICEERROR);
 279
 280         }
 281         return results;
 282
 283     }
 284
 285     /**
 286      *
 287      * @param fullyQualifiedKey lockName
 288      * @return
 289      */
 290     public  String whoseTurnIsIt(String fullyQualifiedKey) {
 291         String[] splitString = fullyQualifiedKey.split("\\.");
 292         String keyspace = splitString[0];
 293         String table = splitString[1];
 294         String primaryKeyValue = splitString[2];
 295         try {
 296             return "$" + fullyQualifiedKey + "$"
 297                     + getLockingServiceHandle().peekLockQueue(keyspace, table, primaryKeyValue).lockRef;
 298         } catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
 299              logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKINGERROR+fullyQualifiedKey ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 300         }
 301         return null;
 302     }
 303
 304     /**
 305      *
 306      * @param lockReference
 307      * @return
 308      */
 309     public static String getLockNameFromId(String lockReference) {
 310         StringTokenizer st = new StringTokenizer(lockReference);
 311         return st.nextToken("$");
 312     }
 313
 314     @Override
 315     public void destroyLockRef(String lockId) {
 316         long start = System.currentTimeMillis();
 317         String fullyQualifiedKey = lockId.substring(1, lockId.lastIndexOf("$"));
 318         String lockRef = lockId.substring(lockId.lastIndexOf('$')+1);
 319         String[] splitString = fullyQualifiedKey.split("\\.");
 320         String keyspace = splitString[0];
 321         String table = splitString[1];
 322         String primaryKeyValue = splitString[2];
 323         try {
 324             getLockingServiceHandle().deQueueLockRef(keyspace, table, primaryKeyValue, lockRef);
 325         } catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
 326             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.DESTROYLOCK+lockRef  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 327         }
 328         long end = System.currentTimeMillis();
 329         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to destroy lock reference:" + (end - start) + " ms");
 330     }
 331
 332     public  MusicLockState destroyLockRef(String fullyQualifiedKey, String lockReference) {
 333         long start = System.currentTimeMillis();
 334         String[] splitString = fullyQualifiedKey.split("\\.");
 335         String keyspace = splitString[0];
 336         String table = splitString[1];
 337         String primaryKeyValue = splitString[2];
 338         try {
 339             getLockingServiceHandle().deQueueLockRef(keyspace, table, primaryKeyValue, lockReference);
 340         } catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
 341             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.DESTROYLOCK+lockReference  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 342         }
 343         long end = System.currentTimeMillis();
 344         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to destroy lock reference:" + (end - start) + " ms");
 345         return new MusicLockState(LockStatus.UNLOCKED, "");
 346     }
 347
 348     @Override
 349     public MusicLockState releaseLock(String lockId, boolean voluntaryRelease) {
 350         String fullyQualifiedKey = lockId.substring(1, lockId.lastIndexOf("$"));
 351         String lockRef = lockId.substring(lockId.lastIndexOf('$')+1);
 352         if (voluntaryRelease) {
 353             return voluntaryReleaseLock(fullyQualifiedKey, lockRef);
 354         } else {
 355             return forciblyReleaseLock(fullyQualifiedKey, lockRef);
 356         }
 357     }
 358
 359     public   MusicLockState  voluntaryReleaseLock(String fullyQualifiedKey, String lockReference) {
 360         return destroyLockRef(fullyQualifiedKey, lockReference);
 361     }
 362
 363     public  MusicLockState  forciblyReleaseLock(String fullyQualifiedKey, String lockReference) {
 364         String[] splitString = fullyQualifiedKey.split("\\.");
 365         String keyspace = splitString[0];
 366         String table = splitString[1];
 367
 368             //leave a signal that this key could potentially be unsynchronized
 369         String syncTable = keyspace+".unsyncedKeys_"+table;
 370         PreparedQueryObject queryObject = new PreparedQueryObject();
 371         String values = "(?)";
 372         queryObject.addValue(fullyQualifiedKey);
 373         String insQuery = "insert into "+syncTable+" (key) values "+values+";";
 374         queryObject.appendQueryString(insQuery);
 375         try {
 376             MusicDataStoreHandle.getDSHandle().executePut(queryObject, "critical");
 377         } catch (Exception e) {
 378             logger.error("Cannot forcibly release lock: " + fullyQualifiedKey + " " + lockReference + ". "
 379                         + e.getMessage());
 380         }
 381
 382         //now release the lock
 383         return destroyLockRef(fullyQualifiedKey, lockReference);
 384     }
 385
 386     /**
 387      *
 388      * @param lockName
 389      * @throws MusicLockingException
 390      */
 391     public  void deleteLock(String lockName) throws MusicLockingException {
 392             //deprecated
 393         }
 394
 395     // Prepared Query Additions.
 396
 397     /**
 398      *
 399      * @param queryObject
 400      * @return ReturnType
 401      * @throws MusicServiceException
 402      */
 403     public  ReturnType eventualPut(PreparedQueryObject queryObject) {
 404         boolean result = false;
 405         try {
 406             result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, MusicUtil.EVENTUAL);
 407         } catch (MusicServiceException | MusicQueryException ex) {
 408             logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage(), "[ERR512E] Failed to get ZK Lock Handle "  ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
 409             logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage() + "  " + ex.getCause() + " " + ex);
 410             return new ReturnType(ResultType.FAILURE, ex.getMessage());
 411         }
 412         if (result) {
 413             return new ReturnType(ResultType.SUCCESS, "Eventual Operation Successfully performed");
 414         } else {
 415             return new ReturnType(ResultType.FAILURE, "Eventual Operation failed to perform");
 416         }
 417     }
 418
 419     /**
 420      *
 421      * @param queryObject
 422      * @return ReturnType
 423      * @throws MusicServiceException
 424      */
 425     public  ReturnType eventualPut_nb(PreparedQueryObject queryObject,String keyspace,String tablename,String primaryKey) {
 426         boolean result = false;
 427         long guard = 0;
 428         PreparedQueryObject getGaurd = new PreparedQueryObject();
 429         getGaurd.appendQueryString("SELECT guard FROM "+keyspace+".lockq_"+tablename+ " WHERE key = ? ;");
 430         getGaurd.addValue(primaryKey);
 431         try {
 432             ResultSet getGaurdResult = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(getGaurd);
 433             Row row = getGaurdResult.one();
 434             if (row != null) {
 435                 guard = row.getLong("guard");
 436                 long timeOfWrite = System.currentTimeMillis();
 437                 long ts = MusicUtil.v2sTimeStampInMicroseconds(guard, timeOfWrite);
 438                 String query = queryObject.getQuery();
 439                 if (!queryObject.getQuery().contains("USING TIMESTAMP")) {
 440                     if (queryObject.getOperation().equalsIgnoreCase("delete"))
 441                         query = query.replaceFirst("WHERE", " USING TIMESTAMP " + ts + " WHERE ");
 442                     else
 443                         query = query.replaceFirst("SET", "USING TIMESTAMP " + ts + " SET");
 444                 }
 445                 queryObject.replaceQueryString(query);
 446             }
 447
 448         } catch (MusicServiceException | MusicQueryException e) {
 449             logger.error(EELFLoggerDelegate.applicationLogger,e.getMessage());
 450         }
 451         try {
 452             result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, MusicUtil.EVENTUAL);
 453         } catch (MusicServiceException | MusicQueryException ex) {
 454             logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage(), "[ERR512E] Failed to get ZK Lock Handle "  ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
 455             logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage() + "  " + ex.getCause() + " " + ex);
 456             return new ReturnType(ResultType.FAILURE, ex.getMessage());
 457         }
 458         if (result) {
 459             return new ReturnType(ResultType.SUCCESS, "Eventual Operation Successfully performed");
 460         } else {
 461             return new ReturnType(ResultType.FAILURE, "Eventual Operation failed to perform");
 462         }
 463     }
 464
 465     /**
 466      *
 467      * @param keyspace
 468      * @param table
 469      * @param primaryKeyValue
 470      * @param queryObject
 471      * @param lockId
 472      * @return
 473      */
 474     public  ReturnType criticalPut(String keyspace, String table, String primaryKeyValue,
 475                     PreparedQueryObject queryObject, String lockId, Condition conditionInfo) {
 476         long start = System.currentTimeMillis();
 477         try {
 478         ReturnType result = isTopOfLockStore(keyspace, table, primaryKeyValue,
 479                 lockId.substring(lockId.lastIndexOf("$")+1));
 480         if(result.getResult().equals(ResultType.FAILURE))
 481                 return result;//not top of the lock store q
 482
 483         if (conditionInfo != null)
 484             try {
 485               if (conditionInfo.testCondition() == false)
 486                   return new ReturnType(ResultType.FAILURE,
 487                                   "Lock acquired but the condition is not true");
 488             } catch (Exception e) {
 489               return new ReturnType(ResultType.FAILURE,
 490                       "Exception thrown while checking the condition, check its sanctity:\n"
 491                                       + e.getMessage());
 492             }
 493
 494           String query = queryObject.getQuery();
 495           long timeOfWrite = System.currentTimeMillis();
 496           long lockOrdinal = Long.parseLong(lockId.substring(lockId.lastIndexOf("$")+1));
 497           long ts = MusicUtil.v2sTimeStampInMicroseconds(lockOrdinal, timeOfWrite);
 498           // TODO: use Statement instead of modifying query
 499             if (!queryObject.getQuery().contains("USING TIMESTAMP")) {
 500                 if (queryObject.getOperation().equalsIgnoreCase("delete"))
 501                     query = query.replaceFirst("WHERE", " USING TIMESTAMP " + ts + " WHERE ");
 502                 else if (queryObject.getOperation().equalsIgnoreCase("insert"))
 503                     query = query.replaceFirst(";", " USING TIMESTAMP " + ts + " ; ");
 504                 else
 505                     query = query.replaceFirst("SET", "USING TIMESTAMP " + ts + " SET");
 506             }
 507           queryObject.replaceQueryString(query);
 508           MusicDataStore dsHandle = MusicDataStoreHandle.getDSHandle();
 509           dsHandle.executePut(queryObject, MusicUtil.CRITICAL);
 510           long end = System.currentTimeMillis();
 511           logger.info(EELFLoggerDelegate.applicationLogger,"Time taken for the critical put:" + (end - start) + " ms");
 512         }catch (MusicQueryException | MusicServiceException | MusicLockingException  e) {
 513             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage());
 514             return new ReturnType(ResultType.FAILURE,
 515                             "Exception thrown while doing the critical put\n"
 516                                             + e.getMessage());
 517         }
 518          return new ReturnType(ResultType.SUCCESS, "Update performed");
 519     }
 520
 521
 522     /**
 523      *
 524      * @param queryObject
 525      * @param consistency
 526      * @return Boolean Indicates success or failure
 527      * @throws MusicServiceException
 528      *
 529      *
 530      */
 531     public  ResultType nonKeyRelatedPut(PreparedQueryObject queryObject, String consistency) throws MusicServiceException {
 532         // this is mainly for some functions like keyspace creation etc which does not
 533         // really need the bells and whistles of Music locking.
 534         boolean result = false;
 535         try {
 536             result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, consistency);
 537         } catch (MusicQueryException | MusicServiceException ex) {
 538             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(), AppMessages.UNKNOWNERROR,
 539                     ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
 540             throw new MusicServiceException(ex.getMessage());
 541         }
 542         return result ? ResultType.SUCCESS : ResultType.FAILURE;
 543     }
 544
 545     /**
 546      * This method performs DDL operation on cassandra.
 547      *
 548      * @param queryObject query object containing prepared query and values
 549      * @return ResultSet
 550      * @throws MusicServiceException
 551      */
 552     public  ResultSet get(PreparedQueryObject queryObject) throws MusicServiceException {
 553         ResultSet results = null;
 554         try {
 555             results = MusicDataStoreHandle.getDSHandle().executeOneConsistencyGet(queryObject);
 556         } catch (MusicQueryException | MusicServiceException e) {
 557             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage());
 558             throw new MusicServiceException(e.getMessage());
 559         }
 560         return results;
 561     }
 562
 563     /**
 564      * This method performs DDL operations on cassandra, if the the resource is available. Lock ID
 565      * is used to check if the resource is free.
 566      *
 567      * @param keyspace name of the keyspace
 568      * @param table name of the table
 569      * @param primaryKeyValue primary key value
 570      * @param queryObject query object containing prepared query and values
 571      * @param lockId lock ID to check if the resource is free to perform the operation.
 572      * @return ResultSet
 573      */
 574     public  ResultSet criticalGet(String keyspace, String table, String primaryKeyValue,
 575                     PreparedQueryObject queryObject, String lockId) throws MusicServiceException {
 576         ResultSet results = null;
 577
 578         try {
 579             ReturnType result = isTopOfLockStore(keyspace, table, primaryKeyValue,
 580                     lockId.substring(lockId.lastIndexOf("$")+1));
 581             if(result.getResult().equals(ResultType.FAILURE))
 582                     return null;//not top of the lock store q
 583                 results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(queryObject);
 584         } catch (MusicQueryException | MusicServiceException | MusicLockingException e) {
 585                 logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.UNKNOWNERROR  ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
 586         }
 587         return results;
 588     }
 589
 590     /**
 591      * This method performs DML operation on cassandra, when the lock of the dd is acquired.
 592      *
 593      * @param keyspaceName name of the keyspace
 594      * @param tableName name of the table
 595      * @param primaryKey primary key value
 596      * @param queryObject query object containing prepared query and values
 597      * @return ReturnType
 598      * @throws MusicLockingException
 599      * @throws MusicServiceException
 600      * @throws MusicQueryException
 601      */
 602     public  ReturnType atomicPut(String keyspaceName, String tableName, String primaryKey,
 603                     PreparedQueryObject queryObject, Condition conditionInfo) throws MusicLockingException, MusicQueryException, MusicServiceException {
 604         long start = System.currentTimeMillis();
 605         String fullyQualifiedKey = keyspaceName + "." + tableName + "." + primaryKey;
 606         String lockId = createLockReference(fullyQualifiedKey);
 607         long lockCreationTime = System.currentTimeMillis();
 608         ReturnType lockAcqResult = acquireLock(fullyQualifiedKey, lockId);
 609         long lockAcqTime = System.currentTimeMillis();
 610
 611         if (!lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
 612             logger.info(EELFLoggerDelegate.applicationLogger,"unable to acquire lock, id " + lockId);
 613             voluntaryReleaseLock(fullyQualifiedKey,lockId);
 614             return lockAcqResult;
 615         }
 616
 617         logger.info(EELFLoggerDelegate.applicationLogger,"acquired lock with id " + lockId);
 618         String lockRef = lockId.substring(lockId.lastIndexOf("$"));
 619         ReturnType criticalPutResult = criticalPut(keyspaceName, tableName, primaryKey,
 620                         queryObject, lockRef, conditionInfo);
 621         long criticalPutTime = System.currentTimeMillis();
 622         voluntaryReleaseLock(fullyQualifiedKey,lockId);
 623         long lockDeleteTime = System.currentTimeMillis();
 624         String timingInfo = "|lock creation time:" + (lockCreationTime - start)
 625                         + "|lock accquire time:" + (lockAcqTime - lockCreationTime)
 626                         + "|critical put time:" + (criticalPutTime - lockAcqTime)
 627                         + "|lock delete time:" + (lockDeleteTime - criticalPutTime) + "|";
 628         criticalPutResult.setTimingInfo(timingInfo);
 629         return criticalPutResult;
 630     }
 631
 632
 633
 634
 635     /**
 636      * This method performs DDL operation on cassasndra, when the lock for the resource is acquired.
 637      *
 638      * @param keyspaceName name of the keyspace
 639      * @param tableName name of the table
 640      * @param primaryKey primary key value
 641      * @param queryObject query object containing prepared query and values
 642      * @return ResultSet
 643      * @throws MusicServiceException
 644      * @throws MusicLockingException
 645      * @throws MusicQueryException
 646      */
 647     public  ResultSet atomicGet(String keyspaceName, String tableName, String primaryKey,
 648                     PreparedQueryObject queryObject) throws MusicServiceException, MusicLockingException, MusicQueryException {
 649         String fullyQualifiedKey = keyspaceName + "." + tableName + "." + primaryKey;
 650         String lockId = createLockReference(fullyQualifiedKey);
 651         long leasePeriod = MusicUtil.getDefaultLockLeasePeriod();
 652         ReturnType lockAcqResult = acquireLock(fullyQualifiedKey, lockId);
 653         if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
 654             logger.info(EELFLoggerDelegate.applicationLogger,"acquired lock with id " + lockId);
 655             String lockRef = lockId.substring(lockId.lastIndexOf("$"));
 656             ResultSet result =
 657                             criticalGet(keyspaceName, tableName, primaryKey, queryObject, lockRef);
 658             voluntaryReleaseLock(fullyQualifiedKey,lockId);
 659             return result;
 660         } else {
 661             voluntaryReleaseLock(fullyQualifiedKey,lockId);
 662             logger.info(EELFLoggerDelegate.applicationLogger,"unable to acquire lock, id " + lockId);
 663             return null;
 664         }
 665     }
 666
 667
 668
 669     /**
 670      * @param lockName
 671      * @return
 672      */
 673     public Map<String, Object> validateLock(String lockName) {
 674         return MusicUtil.validateLock(lockName);
 675     }
 676
 677     @Override
 678     public ReturnType atomicPutWithDeleteLock(String keyspaceName, String tableName, String primaryKey,
 679             PreparedQueryObject queryObject, Condition conditionInfo) throws MusicLockingException {
 680         //Deprecated
 681         return null;
 682     }
 683     @Override
 684     public List<String> getLockQueue(String fullyQualifiedKey)
 685             throws MusicServiceException, MusicQueryException, MusicLockingException {
 686         String[] splitString = fullyQualifiedKey.split("\\.");
 687         String keyspace = splitString[0];
 688         String table = splitString[1];
 689         String primaryKeyValue = splitString[2];
 690
 691         return getLockingServiceHandle().getLockQueue(keyspace, table, primaryKeyValue);
 692     }
 693     @Override
 694     public long getLockQueueSize(String fullyQualifiedKey)
 695             throws MusicServiceException, MusicQueryException, MusicLockingException {
 696         String[] splitString = fullyQualifiedKey.split("\\.");
 697         String keyspace = splitString[0];
 698         String table = splitString[1];
 699         String primaryKeyValue = splitString[2];
 700
 701         return getLockingServiceHandle().getLockQueueSize(keyspace, table, primaryKeyValue);
 702     }
 703     @Override
 704     public ResultSet atomicGetWithDeleteLock(String keyspaceName, String tableName, String primaryKey,
 705             PreparedQueryObject queryObject) throws MusicServiceException, MusicLockingException {
 706         //deprecated
 707         return null;
 708     }
 709
 710 }