src/main/java/org/onap/music/service/impl/MusicZKCore.java

   1 /*
   2  * ============LICENSE_START==========================================
   3  * org.onap.music
   4  * ===================================================================
   5  *  Copyright (c) 2017 AT&T Intellectual Property
   6  * ===================================================================
   7  *  Modifications Copyright (c) 2019 Samsung
   8  * ===================================================================
   9  *  Licensed under the Apache License, Version 2.0 (the "License");
  10  *  you may not use this file except in compliance with the License.
  11  *  You may obtain a copy of the License at
  12  *
  13  *     http://www.apache.org/licenses/LICENSE-2.0
  14  *
  15  *  Unless required by applicable law or agreed to in writing, software
  16  *  distributed under the License is distributed on an "AS IS" BASIS,
  17  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18  *  See the License for the specific language governing permissions and
  19  *  limitations under the License.
  20  *
  21  * ============LICENSE_END=============================================
  22  * ====================================================================
  23  */
  24
  25 package org.onap.music.service.impl;
  26
  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.apache.zookeeper.KeeperException;
  34 import org.apache.zookeeper.KeeperException.NoNodeException;
  35 import org.onap.music.datastore.PreparedQueryObject;
  36 import org.onap.music.datastore.jsonobjects.JsonKeySpace;
  37 import org.onap.music.eelf.logging.EELFLoggerDelegate;
  38 import org.onap.music.eelf.logging.format.AppMessages;
  39 import org.onap.music.eelf.logging.format.ErrorSeverity;
  40 import org.onap.music.eelf.logging.format.ErrorTypes;
  41 import org.onap.music.exceptions.MusicLockingException;
  42 import org.onap.music.exceptions.MusicQueryException;
  43 import org.onap.music.exceptions.MusicServiceException;
  44 import org.onap.music.lockingservice.cassandra.MusicLockState;
  45 import org.onap.music.lockingservice.cassandra.MusicLockState.LockStatus;
  46 import org.onap.music.lockingservice.zookeeper.MusicLockingService;
  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 import org.onap.music.datastore.*;
  52
  53 import com.datastax.driver.core.DataType;
  54 import com.datastax.driver.core.ResultSet;
  55 import com.datastax.driver.core.Row;
  56 import com.datastax.driver.core.TableMetadata;
  57
  58 /**
  59  * This class .....
  60  *
  61  *
  62  */
  63 public class MusicZKCore implements MusicCoreService {
  64
  65     public static MusicLockingService mLockHandle = null;
  66     private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(MusicZKCore.class);
  67     private static MusicZKCore musicZKCoreInstance = null;
  68
  69     private MusicZKCore() {
  70
  71     }
  72     public static MusicZKCore getInstance() {
  73
  74         if(musicZKCoreInstance == null) {
  75             musicZKCoreInstance = new MusicZKCore();
  76         }
  77         return musicZKCoreInstance;
  78     }
  79
  80
  81
  82
  83
  84     public static MusicLockingService getLockingServiceHandle() throws MusicLockingException {
  85         logger.info(EELFLoggerDelegate.applicationLogger,"Acquiring lock store handle");
  86         long start = System.currentTimeMillis();
  87
  88         if (mLockHandle == null) {
  89             try {
  90                 mLockHandle = new MusicLockingService();
  91             } catch (Exception e) {
  92                 logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKHANDLE,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
  93                 throw new MusicLockingException("Failed to aquire Locl store handle " + e);
  94             }
  95         }
  96         long end = System.currentTimeMillis();
  97         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to acquire lock store handle:" + (end - start) + " ms");
  98         return mLockHandle;
  99     }
 100
 101
 102
 103     public  String createLockReference(String lockName) {
 104         logger.info(EELFLoggerDelegate.applicationLogger,"Creating lock reference for lock name:" + lockName);
 105         long start = System.currentTimeMillis();
 106         String lockId = null;
 107         try {
 108             lockId = getLockingServiceHandle().createLockId("/" + lockName);
 109         } catch (MusicLockingException e) {
 110             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.CREATELOCK+lockName,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 111
 112         }
 113         long end = System.currentTimeMillis();
 114         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to create lock reference:" + (end - start) + " ms");
 115         return lockId;
 116     }
 117
 118     /**
 119      *
 120      * @param key
 121      * @return
 122      */
 123     public static boolean isTableOrKeySpaceLock(String key) {
 124         String[] splitString = key.split("\\.");
 125         if (splitString.length > 2)
 126             return false;
 127         else
 128             return true;
 129     }
 130
 131     /**
 132      *
 133      * @param key
 134      * @return
 135      */
 136     public static MusicLockState getMusicLockState(String key) {
 137         long start = System.currentTimeMillis();
 138         try {
 139             String[] splitString = key.split("\\.");
 140             String keyspaceName = splitString[0];
 141             String tableName = splitString[1];
 142             String primaryKey = splitString[2];
 143             MusicLockState mls;
 144             String lockName = keyspaceName + "." + tableName + "." + primaryKey;
 145             mls = getLockingServiceHandle().getLockState(lockName);
 146             long end = System.currentTimeMillis();
 147             logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to get lock state:" + (end - start) + " ms");
 148             return mls;
 149         } catch (NullPointerException | MusicLockingException e) {
 150             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.INVALIDLOCK,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 151         }
 152         return null;
 153     }
 154
 155     public  ReturnType acquireLockWithLease(String key, String lockId, long leasePeriod) {
 156         try {
 157             long start = System.currentTimeMillis();
 158             /* check if the current lock has exceeded its lease and if yes, release that lock */
 159             MusicLockState mls = getMusicLockState(key);
 160             if (mls != null) {
 161                 if (mls.getLockStatus().equals(LockStatus.LOCKED)) {
 162                     logger.info(EELFLoggerDelegate.applicationLogger,"The current lock holder for " + key + " is " + mls.getLockHolder()
 163                                     + ". Checking if it has exceeded lease");
 164                     long currentLockPeriod = System.currentTimeMillis() - mls.getLeaseStartTime();
 165                     long currentLeasePeriod = mls.getLeasePeriod();
 166                     if (currentLockPeriod > currentLeasePeriod) {
 167                         logger.info(EELFLoggerDelegate.applicationLogger,"Lock period " + currentLockPeriod
 168                                         + " has exceeded lease period " + currentLeasePeriod);
 169                         boolean voluntaryRelease = false;
 170                         String currentLockHolder = mls.getLockHolder();
 171                         mls = releaseLock(currentLockHolder, voluntaryRelease);
 172                     }
 173                 }
 174             } else
 175                 logger.error(EELFLoggerDelegate.errorLogger,key, AppMessages.INVALIDLOCK,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 176
 177             /*
 178              * call the traditional acquire lock now and if the result returned is true, set the
 179              * begin time-stamp and lease period
 180              */
 181             if (acquireLock(key, lockId).getResult() == ResultType.SUCCESS) {
 182                 mls = getMusicLockState(key);// get latest state
 183                 if ( mls == null ) {
 184                     logger.info(EELFLoggerDelegate.applicationLogger,"Music Lock State is null");
 185                     return new ReturnType(ResultType.FAILURE, "Could not acquire lock, Lock State is null");
 186                 }
 187                 if (mls.getLeaseStartTime() == -1) {// set it again only if it is not set already
 188                     mls.setLeaseStartTime(System.currentTimeMillis());
 189                     mls.setLeasePeriod(leasePeriod);
 190                     getLockingServiceHandle().setLockState(key, mls);
 191                 }
 192                 long end = System.currentTimeMillis();
 193                 logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to acquire leased lock:" + (end - start) + " ms");
 194                 return new ReturnType(ResultType.SUCCESS, "Accquired lock");
 195             } else {
 196                 long end = System.currentTimeMillis();
 197                 logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to fail to acquire leased lock:" + (end - start) + " ms");
 198                 return new ReturnType(ResultType.FAILURE, "Could not acquire lock");
 199             }
 200         } catch (Exception e) {
 201             StringWriter sw = new StringWriter();
 202                logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), "[ERR506E] Failed to aquire lock ",ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 203
 204             String exceptionAsString = sw.toString();
 205             return new ReturnType(ResultType.FAILURE,
 206                             "Exception thrown in acquireLockWithLease:\n" + exceptionAsString);
 207         }
 208     }
 209
 210     public  ReturnType acquireLock(String key, String lockId) throws MusicLockingException {
 211         /*
 212          * first check if I am on top. Since ids are not reusable there is no need to check
 213          * lockStatus If the status is unlocked, then the above call will automatically return
 214          * false.
 215          */
 216         Boolean result = false;
 217         try {
 218             result = getLockingServiceHandle().isMyTurn(lockId);
 219         } catch (MusicLockingException e2) {
 220             logger.error(EELFLoggerDelegate.errorLogger,AppMessages.INVALIDLOCK + lockId + " " + e2);
 221             throw new MusicLockingException();
 222         }
 223         if (!result) {
 224             logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: Not your turn, someone else has the lock");
 225             try {
 226                 if (!getLockingServiceHandle().lockIdExists(lockId)) {
 227                     logger.info(EELFLoggerDelegate.applicationLogger, "In acquire lock: this lockId doesn't exist");
 228                     return new ReturnType(ResultType.FAILURE, "Lockid doesn't exist");
 229                 }
 230             } catch (MusicLockingException e) {
 231                 logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.INVALIDLOCK+lockId,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 232                  throw new MusicLockingException();
 233             }
 234             logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: returning failure");
 235             return new ReturnType(ResultType.FAILURE, "Not your turn, someone else has the lock");
 236         }
 237
 238
 239         // this is for backward compatibility where locks could also be acquired on just
 240         // keyspaces or tables.
 241         if (isTableOrKeySpaceLock(key)) {
 242             logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: A table or keyspace lock so no need to perform sync...so returning true");
 243             return new ReturnType(ResultType.SUCCESS, "A table or keyspace lock so no need to perform sync...so returning true");
 244         }
 245
 246         // read the lock name corresponding to the key and if the status is locked or being locked,
 247         // then return false
 248         MusicLockState currentMls = null;
 249         MusicLockState newMls = null;
 250         try {
 251             currentMls = getMusicLockState(key);
 252             String currentLockHolder = null;
 253             if(currentMls != null) { currentLockHolder = currentMls.getLockHolder(); };
 254             if (lockId.equals(currentLockHolder)) {
 255                 logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: You already have the lock!");
 256                 return new ReturnType(ResultType.SUCCESS, "You already have the lock!");
 257             }
 258         } catch (NullPointerException e) {
 259             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.INVALIDLOCK+lockId,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 260         }
 261
 262         // change status to "being locked". This state transition is necessary to ensure syncing
 263         // before granting the lock
 264         String lockHolder = null;
 265         boolean needToSyncQuorum = false;
 266         if (currentMls != null)
 267             needToSyncQuorum = currentMls.isNeedToSyncQuorum();
 268
 269
 270         newMls = new MusicLockState(MusicLockState.LockStatus.BEING_LOCKED, lockHolder,
 271                         needToSyncQuorum);
 272         try {
 273             getLockingServiceHandle().setLockState(key, newMls);
 274         } catch (MusicLockingException e1) {
 275             logger.error(EELFLoggerDelegate.errorLogger,e1.getMessage(), AppMessages.LOCKSTATE+key,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 276         }
 277         logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: Set lock state to being_locked");
 278
 279         // do syncing if this was a forced lock release
 280         if (needToSyncQuorum) {
 281             logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: Since there was a forcible release, need to sync quorum!");
 282             try {
 283               syncQuorum(key);
 284             } catch (Exception e) {
 285               logger.error(EELFLoggerDelegate.errorLogger,"Failed to set Lock state " + e);
 286             }
 287         }
 288
 289         // change status to locked
 290         lockHolder = lockId;
 291         needToSyncQuorum = false;
 292         newMls = new MusicLockState(MusicLockState.LockStatus.LOCKED, lockHolder, needToSyncQuorum);
 293         try {
 294             getLockingServiceHandle().setLockState(key, newMls);
 295         } catch (MusicLockingException e) {
 296             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKSTATE+key,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 297         }
 298         logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: Set lock state to locked and assigned current lock ref "
 299                         + lockId + " as holder");
 300
 301         return new ReturnType(result?ResultType.SUCCESS:ResultType.FAILURE, "Set lock state to locked and assigned a lock holder");
 302     }
 303
 304
 305
 306     /**
 307      *
 308      * @param keyspaceName
 309      * @param kspObject
 310      * @return
 311      * @throws Exception
 312      */
 313     public boolean createKeyspace(String keyspaceName, JsonKeySpace kspObject) throws Exception {
 314         return true;
 315     }
 316
 317
 318     private static void syncQuorum(String key) throws Exception {
 319         logger.info(EELFLoggerDelegate.applicationLogger,"Performing sync operation---");
 320         String[] splitString = key.split("\\.");
 321         String keyspaceName = splitString[0];
 322         String tableName = splitString[1];
 323         String primaryKeyValue = splitString[2];
 324         PreparedQueryObject selectQuery = new PreparedQueryObject();
 325         PreparedQueryObject updateQuery = new PreparedQueryObject();
 326
 327         // get the primary key d
 328         TableMetadata tableInfo = MusicDataStoreHandle.returnColumnMetadata(keyspaceName, tableName);
 329         String primaryKeyName = tableInfo.getPrimaryKey().get(0).getName();// we only support single
 330                                                                            // primary key
 331         DataType primaryKeyType = tableInfo.getPrimaryKey().get(0).getType();
 332         Object cqlFormattedPrimaryKeyValue =
 333                         MusicUtil.convertToActualDataType(primaryKeyType, primaryKeyValue);
 334
 335         // get the row of data from a quorum
 336         selectQuery.appendQueryString("SELECT *  FROM " + keyspaceName + "." + tableName + " WHERE "
 337                         + primaryKeyName + "= ?" + ";");
 338         selectQuery.addValue(cqlFormattedPrimaryKeyValue);
 339         MusicUtil.writeBackToQuorum(selectQuery, primaryKeyName, updateQuery, keyspaceName, tableName,
 340             cqlFormattedPrimaryKeyValue);
 341     }
 342
 343
 344
 345
 346     /**
 347      *
 348      * @param query
 349      * @return ResultSet
 350      */
 351     public  ResultSet quorumGet(PreparedQueryObject query) {
 352         ResultSet results = null;
 353         try {
 354             results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(query);
 355         } catch (MusicServiceException | MusicQueryException e) {
 356             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.UNKNOWNERROR ,ErrorSeverity.MAJOR, ErrorTypes.GENERALSERVICEERROR);
 357
 358         }
 359         return results;
 360
 361     }
 362
 363
 364
 365     /**
 366      *
 367      * @param lockName
 368      * @return
 369      */
 370     public  String whoseTurnIsIt(String lockName) {
 371
 372         try {
 373             return getLockingServiceHandle().whoseTurnIsIt("/" + lockName) + "";
 374         } catch (MusicLockingException e) {
 375              logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKINGERROR+lockName ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 376         }
 377         return null;
 378
 379
 380     }
 381
 382     /**
 383      *
 384      * @param lockId
 385      * @return
 386      */
 387     public static String getLockNameFromId(String lockId) {
 388         StringTokenizer st = new StringTokenizer(lockId);
 389         return st.nextToken("$");
 390     }
 391
 392     public void destroyLockRef(String lockId) {
 393         long start = System.currentTimeMillis();
 394         try {
 395             getLockingServiceHandle().unlockAndDeleteId(lockId);
 396         } catch (MusicLockingException | NoNodeException e) {
 397             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.DESTROYLOCK+lockId  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 398         }
 399         long end = System.currentTimeMillis();
 400         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to destroy lock reference:" + (end - start) + " ms");
 401     }
 402
 403     public MusicLockState releaseLock(String lockId, boolean voluntaryRelease) {
 404         long start = System.currentTimeMillis();
 405         try {
 406             getLockingServiceHandle().unlockAndDeleteId(lockId);
 407         } catch (MusicLockingException e1) {
 408             logger.error(EELFLoggerDelegate.errorLogger,e1.getMessage(), AppMessages.RELEASELOCK+lockId  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 409         } catch (KeeperException.NoNodeException nne) {
 410             logger.error(EELFLoggerDelegate.errorLogger,"Failed to release Lock " + lockId + " " + nne);
 411             MusicLockState mls = new MusicLockState("Lock doesn't exists. Release lock operation failed.");
 412             return mls;
 413         }
 414         String lockName = getLockNameFromId(lockId);
 415         MusicLockState mls;
 416         String lockHolder = null;
 417         if (voluntaryRelease) {
 418             mls = new MusicLockState(MusicLockState.LockStatus.UNLOCKED, lockHolder);
 419             logger.info(EELFLoggerDelegate.applicationLogger,"In unlock: lock voluntarily released for " + lockId);
 420         } else {
 421             boolean needToSyncQuorum = true;
 422             mls = new MusicLockState(MusicLockState.LockStatus.UNLOCKED, lockHolder,
 423                             needToSyncQuorum);
 424             logger.info(EELFLoggerDelegate.applicationLogger,"In unlock: lock forcibly released for " + lockId);
 425         }
 426         try {
 427             getLockingServiceHandle().setLockState(lockName, mls);
 428         } catch (MusicLockingException e) {
 429             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.RELEASELOCK+lockId  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 430         }
 431         long end = System.currentTimeMillis();
 432         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to release lock:" + (end - start) + " ms");
 433         return mls;
 434     }
 435
 436     public static  void  voluntaryReleaseLock(String lockId) throws MusicLockingException{
 437         try {
 438             getLockingServiceHandle().unlockAndDeleteId(lockId);
 439         } catch (KeeperException.NoNodeException e) {
 440             // ??? No way
 441         }
 442     }
 443
 444     /**
 445      *
 446      * @param lockName
 447      * @throws MusicLockingException
 448      */
 449     public  void deleteLock(String lockName) throws MusicLockingException {
 450         long start = System.currentTimeMillis();
 451         logger.info(EELFLoggerDelegate.applicationLogger,"Deleting lock for " + lockName);
 452         try {
 453             getLockingServiceHandle().deleteLock("/" + lockName);
 454         } catch (MusicLockingException e) {
 455              logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.DELTELOCK+lockName  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 456              throw new MusicLockingException(e.getMessage());
 457         }
 458         long end = System.currentTimeMillis();
 459         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to delete lock:" + (end - start) + " ms");
 460     }
 461
 462
 463     /**
 464      *
 465      * @param nodeName
 466      */
 467     public static void pureZkCreate(String nodeName) {
 468         try {
 469             getLockingServiceHandle().getzkLockHandle().createNode(nodeName);
 470         } catch (MusicLockingException e) {
 471              logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), "[ERR512E] Failed to get ZK Lock Handle "  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 472         }
 473     }
 474
 475     /**
 476      *
 477      * @param nodeName
 478      * @param data
 479      */
 480     public static void pureZkWrite(String nodeName, byte[] data) {
 481         long start = System.currentTimeMillis();
 482         logger.info(EELFLoggerDelegate.applicationLogger,"Performing zookeeper write to " + nodeName);
 483         try {
 484             getLockingServiceHandle().getzkLockHandle().setNodeData(nodeName, data);
 485         } catch (MusicLockingException e) {
 486             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), "[ERR512E] Failed to get ZK Lock Handle "  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 487         }
 488         logger.info(EELFLoggerDelegate.applicationLogger,"Performed zookeeper write to " + nodeName);
 489         long end = System.currentTimeMillis();
 490         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken for the actual zk put:" + (end - start) + " ms");
 491     }
 492
 493     /**
 494      *
 495      * @param nodeName
 496      * @return
 497      */
 498     public static byte[] pureZkRead(String nodeName) {
 499         long start = System.currentTimeMillis();
 500         byte[] data = null;
 501         try {
 502             data = getLockingServiceHandle().getzkLockHandle().getNodeData(nodeName);
 503         } catch (MusicLockingException e) {
 504             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), "[ERR512E] Failed to get ZK Lock Handle "  ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
 505         }
 506         long end = System.currentTimeMillis();
 507         logger.info(EELFLoggerDelegate.applicationLogger,"Time taken for the actual zk put:" + (end - start) + " ms");
 508         return data;
 509     }
 510
 511
 512
 513     // Prepared Query Additions.
 514
 515     /**
 516      *
 517      * @param keyspaceName
 518      * @param tableName
 519      * @param primaryKey
 520      * @param queryObject
 521      * @return ReturnType
 522      * @throws MusicServiceException
 523      */
 524     public  ReturnType eventualPut(PreparedQueryObject queryObject) {
 525         boolean result = false;
 526         try {
 527             result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, MusicUtil.EVENTUAL);
 528         } catch (MusicServiceException | MusicQueryException ex) {
 529             logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage(), "[ERR512E] Failed to get ZK Lock Handle "  ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
 530             logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage() + "  " + ex.getCause() + " " + ex);
 531             return new ReturnType(ResultType.FAILURE, ex.getMessage());
 532         }
 533         if (result) {
 534             return new ReturnType(ResultType.SUCCESS, "Eventual Operation Successfully performed");
 535         } else {
 536             return new ReturnType(ResultType.FAILURE, "Eventual Operation failed to perform");
 537         }
 538     }
 539
 540     /**
 541      *
 542      * @param keyspaceName
 543      * @param tableName
 544      * @param primaryKey
 545      * @param queryObject
 546      * @param lockId
 547      * @return
 548      */
 549     public  ReturnType criticalPut(String keyspaceName, String tableName, String primaryKey,
 550                     PreparedQueryObject queryObject, String lockId, Condition conditionInfo) {
 551         long start = System.currentTimeMillis();
 552
 553         try {
 554             MusicLockState mls = getLockingServiceHandle()
 555                             .getLockState(keyspaceName + "." + tableName + "." + primaryKey);
 556             if (mls.getLockHolder().equals(lockId) == true) {
 557                 if (conditionInfo != null)
 558                   try {
 559                     if (conditionInfo.testCondition() == false)
 560                         return new ReturnType(ResultType.FAILURE,
 561                                         "Lock acquired but the condition is not true");
 562                   } catch (Exception e) {
 563                     return new ReturnType(ResultType.FAILURE,
 564                             "Exception thrown while doing the critical put, check sanctity of the row/conditions:\n"
 565                                             + e.getMessage());
 566                   }
 567                 boolean result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, MusicUtil.CRITICAL);
 568                 long end = System.currentTimeMillis();
 569                 logger.info(EELFLoggerDelegate.applicationLogger,"Time taken for the critical put:" + (end - start) + " ms");
 570                 if (result) {
 571                     return new ReturnType(ResultType.SUCCESS, "Update performed");
 572                 } else {
 573                     return new ReturnType(ResultType.FAILURE, "Update failed to perform");
 574                 }
 575             } else
 576                 return new ReturnType(ResultType.FAILURE,
 577                                 "Cannot perform operation since you are the not the lock holder");
 578         } catch (MusicQueryException | MusicServiceException  e) {
 579             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage());
 580             return new ReturnType(ResultType.FAILURE,
 581                             "Exception thrown while doing the critical put, check sanctity of the row/conditions:\n"
 582                                             + e.getMessage());
 583         }catch(MusicLockingException ex){
 584             return new ReturnType(ResultType.FAILURE,ex.getMessage());
 585         }
 586
 587     }
 588
 589     /**
 590      *
 591      * @param queryObject
 592      * @param consistency
 593      * @return Boolean Indicates success or failure
 594      * @throws MusicServiceException
 595      *
 596      *
 597      */
 598     public  ResultType nonKeyRelatedPut(PreparedQueryObject queryObject, String consistency) throws MusicServiceException {
 599         // this is mainly for some functions like keyspace creation etc which does not
 600         // really need the bells and whistles of Music locking.
 601         boolean result = false;
 602         try {
 603             result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, consistency);
 604         } catch (MusicQueryException | MusicServiceException ex) {
 605             logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage(), AppMessages.UNKNOWNERROR  ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
 606             throw new MusicServiceException(ex.getMessage());
 607         }
 608         return result?ResultType.SUCCESS:ResultType.FAILURE;
 609     }
 610
 611     /**
 612      * This method performs DDL operation on cassandra.
 613      *
 614      * @param queryObject query object containing prepared query and values
 615      * @return ResultSet
 616      * @throws MusicServiceException
 617      */
 618     public  ResultSet get(PreparedQueryObject queryObject) throws MusicServiceException {
 619         ResultSet results = null;
 620         try {
 621             results = MusicDataStoreHandle.getDSHandle().executeOneConsistencyGet(queryObject);
 622         } catch (MusicQueryException | MusicServiceException e) {
 623             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage());
 624             throw new MusicServiceException(e.getMessage());
 625         }
 626         return results;
 627     }
 628
 629     public static String getMyHostId() {
 630         PreparedQueryObject pQuery = new PreparedQueryObject();
 631         pQuery.appendQueryString("SELECT HOST_ID FROM SYSTEM.LOCAL");
 632         ResultSet rs = null;
 633         try {
 634             rs = MusicDataStoreHandle.getDSHandle().executeOneConsistencyGet(pQuery);
 635             Row row = rs.one();
 636             return (row == null) ? "UNKNOWN" : row.getUUID("HOST_ID").toString();
 637         } catch (Exception e) {
 638             e.printStackTrace();
 639             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage());
 640         }
 641         logger.error(EELFLoggerDelegate.errorLogger, "Some issue during MusicCore.getMyHostId");
 642         return "UNKNOW";
 643     }
 644
 645     /**
 646      * This method performs DDL operations on cassandra, if the the resource is available. Lock ID
 647      * is used to check if the resource is free.
 648      *
 649      * @param keyspaceName name of the keyspace
 650      * @param tableName name of the table
 651      * @param primaryKey primary key value
 652      * @param queryObject query object containing prepared query and values
 653      * @param lockId lock ID to check if the resource is free to perform the operation.
 654      * @return ResultSet
 655      */
 656     public  ResultSet criticalGet(String keyspaceName, String tableName, String primaryKey,
 657                     PreparedQueryObject queryObject, String lockId) throws MusicServiceException {
 658         ResultSet results = null;
 659         try {
 660             MusicLockState mls = getLockingServiceHandle()
 661                             .getLockState(keyspaceName + "." + tableName + "." + primaryKey);
 662             if (mls.getLockHolder().equals(lockId)) {
 663                 results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(queryObject);
 664             } else
 665                 throw new MusicServiceException("YOU DO NOT HAVE THE LOCK");
 666         } catch (MusicQueryException | MusicServiceException | MusicLockingException e) {
 667             logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.UNKNOWNERROR  ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
 668         }
 669         return results;
 670     }
 671
 672     /**
 673      * This method performs DML operation on cassandra, when the lock of the dd is acquired.
 674      *
 675      * @param keyspaceName name of the keyspace
 676      * @param tableName name of the table
 677      * @param primaryKey primary key value
 678      * @param queryObject query object containing prepared query and values
 679      * @return ReturnType
 680      * @throws MusicLockingException
 681      */
 682     public  ReturnType atomicPut(String keyspaceName, String tableName, String primaryKey,
 683                     PreparedQueryObject queryObject, Condition conditionInfo) throws MusicLockingException {
 684
 685         long start = System.currentTimeMillis();
 686         String key = keyspaceName + "." + tableName + "." + primaryKey;
 687         String lockId = createLockReference(key);
 688         long lockCreationTime = System.currentTimeMillis();
 689         ReturnType lockAcqResult = acquireLock(key, lockId);
 690         long lockAcqTime = System.currentTimeMillis();
 691         if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
 692             logger.info(EELFLoggerDelegate.applicationLogger,"acquired lock with id " + lockId);
 693             ReturnType criticalPutResult = criticalPut(keyspaceName, tableName, primaryKey,
 694                             queryObject, lockId, conditionInfo);
 695             long criticalPutTime = System.currentTimeMillis();
 696             voluntaryReleaseLock(lockId);
 697             long lockDeleteTime = System.currentTimeMillis();
 698             String timingInfo = "|lock creation time:" + (lockCreationTime - start)
 699                             + "|lock accquire time:" + (lockAcqTime - lockCreationTime)
 700                             + "|critical put time:" + (criticalPutTime - lockAcqTime)
 701                             + "|lock delete time:" + (lockDeleteTime - criticalPutTime) + "|";
 702             criticalPutResult.setTimingInfo(timingInfo);
 703             return criticalPutResult;
 704         } else {
 705             logger.info(EELFLoggerDelegate.applicationLogger,"unable to acquire lock, id " + lockId);
 706             destroyLockRef(lockId);
 707             return lockAcqResult;
 708         }
 709     }
 710
 711     /**
 712      * this function is mainly for the benchmarks to see the effect of lock deletion.
 713      *
 714      * @param keyspaceName
 715      * @param tableName
 716      * @param primaryKey
 717      * @param queryObject
 718      * @param conditionInfo
 719      * @return
 720      * @throws MusicLockingException
 721      */
 722     public  ReturnType atomicPutWithDeleteLock(String keyspaceName, String tableName,
 723                     String primaryKey, PreparedQueryObject queryObject, Condition conditionInfo) throws MusicLockingException {
 724
 725         long start = System.currentTimeMillis();
 726         String key = keyspaceName + "." + tableName + "." + primaryKey;
 727         String lockId = createLockReference(key);
 728         long lockCreationTime = System.currentTimeMillis();
 729         long leasePeriod = MusicUtil.getDefaultLockLeasePeriod();
 730         ReturnType lockAcqResult = acquireLock(key, lockId);
 731         long lockAcqTime = System.currentTimeMillis();
 732         if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
 733             logger.info(EELFLoggerDelegate.applicationLogger,"acquired lock with id " + lockId);
 734             ReturnType criticalPutResult = criticalPut(keyspaceName, tableName, primaryKey,
 735                             queryObject, lockId, conditionInfo);
 736             long criticalPutTime = System.currentTimeMillis();
 737             deleteLock(key);
 738             long lockDeleteTime = System.currentTimeMillis();
 739             String timingInfo = "|lock creation time:" + (lockCreationTime - start)
 740                             + "|lock accquire time:" + (lockAcqTime - lockCreationTime)
 741                             + "|critical put time:" + (criticalPutTime - lockAcqTime)
 742                             + "|lock delete time:" + (lockDeleteTime - criticalPutTime) + "|";
 743             criticalPutResult.setTimingInfo(timingInfo);
 744             return criticalPutResult;
 745         } else {
 746             logger.info(EELFLoggerDelegate.applicationLogger,"unable to acquire lock, id " + lockId);
 747             deleteLock(key);
 748             return lockAcqResult;
 749         }
 750     }
 751
 752
 753
 754
 755     /**
 756      * This method performs DDL operation on cassasndra, when the lock for the resource is acquired.
 757      *
 758      * @param keyspaceName name of the keyspace
 759      * @param tableName name of the table
 760      * @param primaryKey primary key value
 761      * @param queryObject query object containing prepared query and values
 762      * @return ResultSet
 763      * @throws MusicServiceException
 764      * @throws MusicLockingException
 765      */
 766     public  ResultSet atomicGet(String keyspaceName, String tableName, String primaryKey,
 767                     PreparedQueryObject queryObject) throws MusicServiceException, MusicLockingException {
 768         String key = keyspaceName + "." + tableName + "." + primaryKey;
 769         String lockId = createLockReference(key);
 770         long leasePeriod = MusicUtil.getDefaultLockLeasePeriod();
 771         ReturnType lockAcqResult = acquireLock(key, lockId);
 772         if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
 773             logger.info(EELFLoggerDelegate.applicationLogger,"acquired lock with id " + lockId);
 774             ResultSet result =
 775                             criticalGet(keyspaceName, tableName, primaryKey, queryObject, lockId);
 776             voluntaryReleaseLock(lockId);
 777             return result;
 778         } else {
 779             destroyLockRef(lockId);
 780             logger.info(EELFLoggerDelegate.applicationLogger,"unable to acquire lock, id " + lockId);
 781             return null;
 782         }
 783     }
 784
 785     public  ResultSet atomicGetWithDeleteLock(String keyspaceName, String tableName, String primaryKey,
 786             PreparedQueryObject queryObject) throws MusicServiceException, MusicLockingException {
 787         String key = keyspaceName + "." + tableName + "." + primaryKey;
 788         String lockId = createLockReference(key);
 789         long leasePeriod = MusicUtil.getDefaultLockLeasePeriod();
 790
 791         ReturnType lockAcqResult = acquireLock(key, lockId);
 792
 793         if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
 794             logger.info(EELFLoggerDelegate.applicationLogger, "acquired lock with id " + lockId);
 795             ResultSet result = criticalGet(keyspaceName, tableName, primaryKey, queryObject, lockId);
 796             deleteLock(key);
 797             return result;
 798         } else {
 799             deleteLock(key);
 800             logger.info(EELFLoggerDelegate.applicationLogger, "unable to acquire lock, id " + lockId);
 801             return null;
 802         }
 803     }
 804
 805     /**
 806      * @param lockName
 807      * @return
 808      */
 809     public Map<String, Object> validateLock(String lockName) {
 810         return MusicUtil.validateLock(lockName);
 811     }
 812
 813     @Override
 814     public ResultType createTable(String keyspace, String table, PreparedQueryObject tableQueryObject,
 815             String consistency) throws MusicServiceException {
 816         boolean result = false;
 817         try {
 818             //create shadow locking table
 819             //result = createLockQueue(keyspace, table);
 820             if(result == false)
 821               return ResultType.FAILURE;
 822
 823             result = false;
 824
 825             //create table to track unsynced_keys
 826             table = "unsyncedKeys_"+table;
 827
 828             String tabQuery = "CREATE TABLE IF NOT EXISTS "+keyspace+"."+table
 829                     + " ( key text,PRIMARY KEY (key) );";
 830             System.out.println(tabQuery);
 831             PreparedQueryObject queryObject = new PreparedQueryObject();
 832
 833             queryObject.appendQueryString(tabQuery);
 834             result = false;
 835             result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, "eventual");
 836
 837
 838             //create actual table
 839             result = MusicDataStoreHandle.getDSHandle().executePut(tableQueryObject, consistency);
 840         } catch (MusicQueryException | MusicServiceException ex) {
 841             logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage(), AppMessages.UNKNOWNERROR  ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
 842             throw new MusicServiceException(ex.getMessage());
 843         }
 844         return result?ResultType.SUCCESS:ResultType.FAILURE;
 845     }
 846
 847     public static boolean createLockQueue(String keyspace, String table) throws MusicServiceException, MusicQueryException {
 848         logger.info(EELFLoggerDelegate.applicationLogger,
 849                 "Create lock queue/table for " +  keyspace+"."+table);
 850         table = "lockQ_"+table;
 851         String tabQuery = "CREATE TABLE IF NOT EXISTS "+keyspace+"."+table
 852                 + " ( key text, lockReference bigint, createTime text, acquireTime text, guard bigint static, PRIMARY KEY ((key), lockReference) ) "
 853                 + "WITH CLUSTERING ORDER BY (lockReference ASC);";
 854         System.out.println(tabQuery);
 855         PreparedQueryObject queryObject = new PreparedQueryObject();
 856
 857         queryObject.appendQueryString(tabQuery);
 858         boolean result;
 859         result = MusicDataStoreHandle.mDstoreHandle.executePut(queryObject, "eventual");
 860         return result;
 861     }
 862
 863
 864     @Override
 865     public List<String> getLockQueue(String fullyQualifiedKey)
 866             throws MusicServiceException, MusicQueryException, MusicLockingException {
 867         // TODO Auto-generated method stub
 868         return null;
 869     }
 870
 871
 872
 873     @Override
 874     public long getLockQueueSize(String fullyQualifiedKey)
 875             throws MusicServiceException, MusicQueryException, MusicLockingException {
 876         // TODO Auto-generated method stub
 877         return 0;
 878     }
 879     @Override
 880     public ReturnType eventualPut_nb(PreparedQueryObject queryObject, String keyspace, String tablename,
 881             String primaryKey) {
 882         return eventualPut(queryObject);
 883     }
 884
 885
 886
 887
 888 }