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