src/main/java/org/onap/music/datastore/MusicDataStore.java

   1 /*
   2  * ============LICENSE_START==========================================
   3  * org.onap.music
   4  * ===================================================================
   5  *  Copyright (c) 2017 AT&T Intellectual Property
   6  * ===================================================================
   7  *  Modifications Copyright (c) 2018-2019 IBM
   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.datastore;
  26
  27 import java.net.InetAddress;
  28 import java.net.NetworkInterface;
  29 import java.net.SocketException;
  30 import java.nio.ByteBuffer;
  31 import java.util.ArrayList;
  32 import java.util.Enumeration;
  33 import java.util.HashMap;
  34 import java.util.Iterator;
  35 import java.util.Map;
  36
  37 import org.apache.commons.jcs.access.CacheAccess;
  38 import org.onap.music.authentication.CachingUtil;
  39 import org.onap.music.eelf.logging.EELFLoggerDelegate;
  40 import org.onap.music.eelf.logging.format.AppMessages;
  41 import org.onap.music.eelf.logging.format.ErrorSeverity;
  42 import org.onap.music.eelf.logging.format.ErrorTypes;
  43 import org.onap.music.exceptions.MusicQueryException;
  44 import org.onap.music.exceptions.MusicServiceException;
  45 import org.onap.music.main.MusicUtil;
  46
  47 import com.codahale.metrics.JmxReporter;
  48 import com.datastax.driver.core.Cluster;
  49 import com.datastax.driver.core.ColumnDefinitions;
  50 import com.datastax.driver.core.ColumnDefinitions.Definition;
  51 import com.datastax.driver.core.ConsistencyLevel;
  52 import com.datastax.driver.core.DataType;
  53 import com.datastax.driver.core.HostDistance;
  54 import com.datastax.driver.core.KeyspaceMetadata;
  55 import com.datastax.driver.core.Metadata;
  56 import com.datastax.driver.core.PoolingOptions;
  57 import com.datastax.driver.core.PreparedStatement;
  58 import com.datastax.driver.core.ResultSet;
  59 import com.datastax.driver.core.Row;
  60 import com.datastax.driver.core.Session;
  61 import com.datastax.driver.core.SimpleStatement;
  62 import com.datastax.driver.core.TableMetadata;
  63 import com.datastax.driver.core.exceptions.AlreadyExistsException;
  64 import com.datastax.driver.core.exceptions.InvalidQueryException;
  65 import com.datastax.driver.core.exceptions.NoHostAvailableException;
  66 import com.sun.jersey.core.util.Base64;
  67
  68 /**
  69  * @author nelson24
  70  *
  71  */
  72 public class MusicDataStore {
  73
  74     public static final String CONSISTENCY_LEVEL_ONE = "ONE";
  75     public static final String CONSISTENCY_LEVEL_QUORUM = "QUORUM";
  76     private Session session;
  77     private Cluster cluster;
  78
  79
  80     /**
  81      * @param session
  82      */
  83     public void setSession(Session session) {
  84         this.session = session;
  85     }
  86
  87     /**
  88      * @param session
  89      */
  90     public Session getSession() {
  91         return session;
  92     }
  93
  94     /**
  95      * @param cluster
  96      */
  97     public void setCluster(Cluster cluster) {
  98         this.cluster = cluster;
  99     }
 100
 101
 102     private EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(MusicDataStore.class);
 103
 104     /**
 105      *
 106      */
 107     public MusicDataStore() {
 108         connectToCassaCluster();
 109     }
 110
 111
 112     /**
 113      * @param cluster
 114      * @param session
 115      */
 116     public MusicDataStore(Cluster cluster, Session session) {
 117         this.session = session;
 118         this.cluster = cluster;
 119     }
 120
 121     /**
 122      *
 123      * @param remoteIp
 124      * @throws MusicServiceException
 125      */
 126     public MusicDataStore(String remoteIp) {
 127         try {
 128             connectToCassaCluster(remoteIp);
 129         } catch (MusicServiceException e) {
 130             logger.error(EELFLoggerDelegate.errorLogger, e.getMessage());
 131         }
 132     }
 133
 134     /**
 135      *
 136      * @return
 137      */
 138     private ArrayList<String> getAllPossibleLocalIps() {
 139         ArrayList<String> allPossibleIps = new ArrayList<>();
 140         try {
 141             Enumeration<NetworkInterface> en = NetworkInterface.getNetworkInterfaces();
 142             while (en.hasMoreElements()) {
 143                 NetworkInterface ni = en.nextElement();
 144                 Enumeration<InetAddress> ee = ni.getInetAddresses();
 145                 while (ee.hasMoreElements()) {
 146                     InetAddress ia = ee.nextElement();
 147                     allPossibleIps.add(ia.getHostAddress());
 148                 }
 149             }
 150         } catch (SocketException e) {
 151             logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), AppMessages.CONNCECTIVITYERROR, ErrorSeverity.ERROR, ErrorTypes.CONNECTIONERROR);
 152         }catch(Exception e) {
 153             logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), ErrorSeverity.ERROR, ErrorTypes.GENERALSERVICEERROR);
 154         }
 155         return allPossibleIps;
 156     }
 157
 158     /**
 159      * This method iterates through all available IP addresses and connects to multiple cassandra
 160      * clusters.
 161      */
 162     private void connectToCassaCluster() {
 163         Iterator<String> it = getAllPossibleLocalIps().iterator();
 164         String address = "localhost";
 165         String[] addresses = null;
 166         address = MusicUtil.getMyCassaHost();
 167         addresses = address.split(",");
 168
 169         logger.info(EELFLoggerDelegate.applicationLogger,
 170                         "Connecting to cassa cluster: Iterating through possible ips:"
 171                                         + getAllPossibleLocalIps());
 172         PoolingOptions poolingOptions = new PoolingOptions();
 173         poolingOptions
 174         .setConnectionsPerHost(HostDistance.LOCAL,  4, 10)
 175         .setConnectionsPerHost(HostDistance.REMOTE, 2, 4);
 176         while (it.hasNext()) {
 177             try {
 178                 if(MusicUtil.getCassName() != null && MusicUtil.getCassPwd() != null) {
 179                     logger.info(EELFLoggerDelegate.applicationLogger,
 180                             "Building with credentials "+MusicUtil.getCassName()+" & "+MusicUtil.getCassPwd());
 181                     cluster = Cluster.builder().withPort(MusicUtil.getCassandraPort())
 182                                        .withCredentials(MusicUtil.getCassName(), MusicUtil.getCassPwd())
 183                                        //.withLoadBalancingPolicy(new RoundRobinPolicy())
 184                                        .withoutJMXReporting()
 185                                        .withPoolingOptions(poolingOptions)
 186                                        .addContactPoints(addresses).build();
 187                 }
 188                 else
 189                     cluster = Cluster.builder().withPort(MusicUtil.getCassandraPort())
 190                                          //.withLoadBalancingPolicy(new RoundRobinPolicy())
 191                                          .addContactPoints(addresses).build();
 192
 193                 Metadata metadata = cluster.getMetadata();
 194                 logger.info(EELFLoggerDelegate.applicationLogger, "Connected to cassa cluster "
 195                                 + metadata.getClusterName() + " at " + address);
 196                 session = cluster.connect();
 197
 198                 break;
 199             } catch (NoHostAvailableException e) {
 200                 address = it.next();
 201                 logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.HOSTUNAVAILABLE, ErrorSeverity.ERROR, ErrorTypes.CONNECTIONERROR);
 202             }
 203         }
 204     }
 205
 206     /**
 207      *
 208      */
 209     public void close() {
 210         session.close();
 211     }
 212
 213     /**
 214      * This method connects to cassandra cluster on specific address.
 215      *
 216      * @param address
 217      */
 218     private void connectToCassaCluster(String address) throws MusicServiceException {
 219         String[] addresses = null;
 220         addresses = address.split(",");
 221         PoolingOptions poolingOptions = new PoolingOptions();
 222         poolingOptions
 223         .setConnectionsPerHost(HostDistance.LOCAL,  4, 10)
 224         .setConnectionsPerHost(HostDistance.REMOTE, 2, 4);
 225         if(MusicUtil.getCassName() != null && MusicUtil.getCassPwd() != null) {
 226             logger.info(EELFLoggerDelegate.applicationLogger,
 227                     "Building with credentials "+MusicUtil.getCassName()+" & "+MusicUtil.getCassPwd());
 228             cluster = Cluster.builder().withPort(MusicUtil.getCassandraPort())
 229                        .withCredentials(MusicUtil.getCassName(), MusicUtil.getCassPwd())
 230                        //.withLoadBalancingPolicy(new RoundRobinPolicy())
 231                        .withoutJMXReporting()
 232                        .withPoolingOptions(poolingOptions)
 233                        .addContactPoints(addresses).build();
 234         }
 235         else {
 236             cluster = Cluster.builder().withPort(MusicUtil.getCassandraPort())
 237                         //.withLoadBalancingPolicy(new RoundRobinPolicy())
 238                         .withoutJMXReporting()
 239                         .withPoolingOptions(poolingOptions)
 240                         .addContactPoints(addresses).build();
 241         }
 242
 243         // JmxReporter reporter =
 244         //         JmxReporter.forRegistry(cluster.getMetrics().getRegistry())
 245         //             .inDomain(cluster.getClusterName() + "-metrics")
 246         //             .build();
 247
 248         //     reporter.start();
 249
 250         Metadata metadata = cluster.getMetadata();
 251         logger.info(EELFLoggerDelegate.applicationLogger, "Connected to cassa cluster "
 252                         + metadata.getClusterName() + " at " + address);
 253         try {
 254             session = cluster.connect();
 255         } catch (Exception ex) {
 256             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(),AppMessages.CASSANDRACONNECTIVITY, ErrorSeverity.ERROR, ErrorTypes.SERVICEUNAVAILABLE);
 257             throw new MusicServiceException(
 258                             "Error while connecting to Cassandra cluster.. " + ex.getMessage());
 259         }
 260     }
 261
 262     /**
 263      *
 264      * @param keyspace
 265      * @param tableName
 266      * @param columnName
 267      * @return DataType
 268      */
 269     public DataType returnColumnDataType(String keyspace, String tableName, String columnName) {
 270         KeyspaceMetadata ks = cluster.getMetadata().getKeyspace(keyspace);
 271         TableMetadata table = ks.getTable(tableName);
 272         return table.getColumn(columnName).getType();
 273
 274     }
 275
 276     /**
 277      *
 278      * @param keyspace
 279      * @param tableName
 280      * @return TableMetadata
 281      */
 282     public TableMetadata returnColumnMetadata(String keyspace, String tableName) {
 283         KeyspaceMetadata ks = cluster.getMetadata().getKeyspace(keyspace);
 284         return ks.getTable(tableName);
 285     }
 286
 287
 288     /**
 289      * Utility function to return the Java specific object type.
 290      *
 291      * @param row
 292      * @param colName
 293      * @param colType
 294      * @return
 295      */
 296     public Object getColValue(Row row, String colName, DataType colType) {
 297
 298         switch (colType.getName()) {
 299             case VARCHAR:
 300                 return row.getString(colName);
 301             case UUID:
 302                 return row.getUUID(colName);
 303             case VARINT:
 304                 return row.getVarint(colName);
 305             case BIGINT:
 306                 return row.getLong(colName);
 307             case INT:
 308                 return row.getInt(colName);
 309             case FLOAT:
 310                 return row.getFloat(colName);
 311             case DOUBLE:
 312                 return row.getDouble(colName);
 313             case BOOLEAN:
 314                 return row.getBool(colName);
 315             case MAP:
 316                 return row.getMap(colName, String.class, String.class);
 317             case LIST:
 318                 return row.getList(colName, String.class);
 319             default:
 320                 return null;
 321         }
 322     }
 323
 324     public byte[] getBlobValue(Row row, String colName, DataType colType) {
 325         ByteBuffer bb = row.getBytes(colName);
 326         return bb.array();
 327     }
 328
 329     public boolean doesRowSatisfyCondition(Row row, Map<String, Object> condition) throws Exception {
 330         ColumnDefinitions colInfo = row.getColumnDefinitions();
 331
 332         for (Map.Entry<String, Object> entry : condition.entrySet()) {
 333             String colName = entry.getKey();
 334             DataType colType = colInfo.getType(colName);
 335             Object columnValue = getColValue(row, colName, colType);
 336             Object conditionValue = MusicUtil.convertToActualDataType(colType, entry.getValue());
 337             if (columnValue.equals(conditionValue) == false)
 338                 return false;
 339         }
 340         return true;
 341     }
 342
 343     /**
 344      * Utility function to store ResultSet values in to a MAP for output.
 345      *
 346      * @param results
 347      * @return MAP
 348      */
 349     public Map<String, HashMap<String, Object>> marshalData(ResultSet results) {
 350         Map<String, HashMap<String, Object>> resultMap =
 351                         new HashMap<>();
 352         int counter = 0;
 353         for (Row row : results) {
 354             ColumnDefinitions colInfo = row.getColumnDefinitions();
 355             HashMap<String, Object> resultOutput = new HashMap<>();
 356             for (Definition definition : colInfo) {
 357                 if (!(("vector_ts").equals(definition.getName()))) {
 358                     if(definition.getType().toString().toLowerCase().contains("blob")) {
 359                         resultOutput.put(definition.getName(),
 360                                 getBlobValue(row, definition.getName(), definition.getType()));
 361                     }
 362                     else
 363                         resultOutput.put(definition.getName(),
 364                                     getColValue(row, definition.getName(), definition.getType()));
 365                 }
 366             }
 367             resultMap.put("row " + counter, resultOutput);
 368             counter++;
 369         }
 370         return resultMap;
 371     }
 372
 373
 374     // Prepared Statements 1802 additions
 375
 376     public boolean executePut(PreparedQueryObject queryObject, String consistency)
 377             throws MusicServiceException, MusicQueryException {
 378         return executePut(queryObject, consistency, 0);
 379     }
 380     /**
 381      * This Method performs DDL and DML operations on Cassandra using specified consistency level
 382      *
 383      * @param queryObject Object containing cassandra prepared query and values.
 384      * @param consistency Specify consistency level for data synchronization across cassandra
 385      *        replicas
 386      * @return Boolean Indicates operation success or failure
 387      * @throws MusicServiceException
 388      * @throws MusicQueryException
 389      */
 390     public boolean executePut(PreparedQueryObject queryObject, String consistency,long timeSlot)
 391                     throws MusicServiceException, MusicQueryException {
 392
 393         boolean result = false;
 394         long timeOfWrite = System.currentTimeMillis();
 395         if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
 396             logger.error(EELFLoggerDelegate.errorLogger, queryObject.getQuery(),AppMessages.QUERYERROR, ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 397             throw new MusicQueryException("Ill formed queryObject for the request = " + "["
 398                             + queryObject.getQuery() + "]");
 399         }
 400         logger.info(EELFLoggerDelegate.applicationLogger,
 401                         "In preprared Execute Put: the actual insert query:"
 402                                         + queryObject.getQuery() + "; the values"
 403                                         + queryObject.getValues());
 404 /*<<<<<<< HEAD
 405         PreparedStatement preparedInsert = null;
 406         try {
 407
 408                 preparedInsert = session.prepare(queryObject.getQuery());
 409
 410         } catch(InvalidQueryException iqe) {
 411             logger.error("Exception", iqe);
 412             logger.error(EELFLoggerDelegate.errorLogger, iqe.getMessage(),AppMessages.QUERYERROR, ErrorSeverity.CRITICAL, ErrorTypes.QUERYERROR);
 413             throw new MusicQueryException(iqe.getMessage());
 414         }catch(Exception e) {
 415             logger.error("Exception", e);
 416             logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.QUERYERROR, ErrorSeverity.CRITICAL, ErrorTypes.QUERYERROR);
 417             throw new MusicQueryException(e.getMessage());
 418         }
 419
 420 =======*/
 421         SimpleStatement preparedInsert = null;
 422
 423         try {
 424             preparedInsert = new SimpleStatement(queryObject.getQuery(), queryObject.getValues().toArray());
 425             if (consistency.equalsIgnoreCase(MusicUtil.CRITICAL)) {
 426                 logger.info(EELFLoggerDelegate.applicationLogger, "Executing critical put query");
 427                 preparedInsert.setConsistencyLevel(ConsistencyLevel.QUORUM);
 428             } else if (consistency.equalsIgnoreCase(MusicUtil.EVENTUAL)) {
 429                 logger.info(EELFLoggerDelegate.applicationLogger, "Executing simple put query");
 430                 if(queryObject.getConsistency() == null)
 431                     preparedInsert.setConsistencyLevel(ConsistencyLevel.ONE);
 432                else
 433                     preparedInsert.setConsistencyLevel(MusicUtil.getConsistencyLevel(queryObject.getConsistency()));
 434             } else if (consistency.equalsIgnoreCase(MusicUtil.ONE)) {
 435                 preparedInsert.setConsistencyLevel(ConsistencyLevel.ONE);
 436             }  else if (consistency.equalsIgnoreCase(MusicUtil.QUORUM)) {
 437                 preparedInsert.setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
 438             } else if (consistency.equalsIgnoreCase(MusicUtil.ALL)) {
 439                 preparedInsert.setConsistencyLevel(ConsistencyLevel.ALL);
 440             }
 441             long timestamp = MusicUtil.v2sTimeStampInMicroseconds(timeSlot, timeOfWrite);
 442             preparedInsert.setDefaultTimestamp(timestamp);
 443
 444             ResultSet rs = session.execute(preparedInsert);
 445             result = rs.wasApplied();
 446
 447         }
 448         catch (AlreadyExistsException ae) {
 449             logger.error(EELFLoggerDelegate.errorLogger, ae.getMessage(),AppMessages.SESSIONFAILED+ " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 450             throw new MusicServiceException(ae.getMessage());
 451         }
 452         catch (Exception e) {
 453             logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.SESSIONFAILED+ " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 454             throw new MusicQueryException("Executing Session Failure for Request = " + "["
 455                             + queryObject.getQuery() + "]" + " Reason = " + e.getMessage());
 456         }
 457
 458
 459         return result;
 460     }
 461
 462  /*   *//**
 463      * This method performs DDL operations on Cassandra using consistency level ONE.
 464      *
 465      * @param queryObject Object containing cassandra prepared query and values.
 466      * @return ResultSet
 467      * @throws MusicServiceException
 468      * @throws MusicQueryException
 469      *//*
 470     public ResultSet executeEventualGet(PreparedQueryObject queryObject)
 471                     throws MusicServiceException, MusicQueryException {
 472         CacheAccess<String, PreparedStatement> queryBank = CachingUtil.getStatementBank();
 473         PreparedStatement preparedEventualGet = null;
 474         if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
 475             logger.error(EELFLoggerDelegate.errorLogger, "",AppMessages.QUERYERROR+ " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 476             throw new MusicQueryException("Ill formed queryObject for the request = " + "["
 477                             + queryObject.getQuery() + "]");
 478         }
 479         logger.info(EELFLoggerDelegate.applicationLogger,
 480                         "Executing Eventual  get query:" + queryObject.getQuery());
 481
 482         ResultSet results = null;
 483         try {
 484             if(queryBank.get(queryObject.getQuery()) != null )
 485                 preparedEventualGet=queryBank.get(queryObject.getQuery());
 486             else {
 487                 preparedEventualGet = session.prepare(queryObject.getQuery());
 488                 CachingUtil.updateStatementBank(queryObject.getQuery(), preparedEventualGet);
 489             }
 490             if(queryObject.getConsistency() == null) {
 491                 preparedEventualGet.setConsistencyLevel(ConsistencyLevel.ONE);
 492             }
 493            else {
 494                preparedEventualGet.setConsistencyLevel(MusicUtil.getConsistencyLevel(queryObject.getConsistency()));
 495            }
 496            results = session.execute(preparedEventualGet.bind(queryObject.getValues().toArray()));
 497
 498         } catch (Exception ex) {
 499             logger.error("Exception", ex);
 500             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(),AppMessages.UNKNOWNERROR+ "[" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 501             throw new MusicServiceException(ex.getMessage());
 502         }
 503         return results;
 504     }
 505
 506     *//**
 507      *
 508      * This method performs DDL operation on Cassandra using consistency level QUORUM.
 509      *
 510      * @param queryObject Object containing cassandra prepared query and values.
 511      * @return ResultSet
 512      * @throws MusicServiceException
 513      * @throws MusicQueryException
 514      *//*
 515     public ResultSet executeCriticalGet(PreparedQueryObject queryObject)
 516                     throws MusicServiceException, MusicQueryException {
 517         if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
 518             logger.error(EELFLoggerDelegate.errorLogger, "",AppMessages.QUERYERROR+ " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 519             throw new MusicQueryException("Error processing Prepared Query Object for the request = " + "["
 520                             + queryObject.getQuery() + "]");
 521         }
 522         logger.info(EELFLoggerDelegate.applicationLogger,
 523                         "Executing Critical get query:" + queryObject.getQuery());
 524         PreparedStatement preparedEventualGet = session.prepare(queryObject.getQuery());
 525         preparedEventualGet.setConsistencyLevel(ConsistencyLevel.QUORUM);
 526         ResultSet results = null;
 527         try {
 528             results = session.execute(preparedEventualGet.bind(queryObject.getValues().toArray()));
 529         } catch (Exception ex) {
 530             logger.error("Exception", ex);
 531             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(),AppMessages.UNKNOWNERROR+ "[" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 532             throw new MusicServiceException(ex.getMessage());
 533         }
 534         return results;
 535
 536     }
 537     */
 538     public ResultSet executeGet(PreparedQueryObject queryObject,String consistencyLevel) throws MusicQueryException, MusicServiceException {
 539         if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
 540             logger.error(EELFLoggerDelegate.errorLogger, "",AppMessages.QUERYERROR+ " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 541             throw new MusicQueryException("Error processing Prepared Query Object for the request = " + "["
 542                             + queryObject.getQuery() + "]");
 543         }
 544         ResultSet results = null;
 545         try {
 546             SimpleStatement statement = new SimpleStatement(queryObject.getQuery(), queryObject.getValues().toArray());
 547
 548             if (consistencyLevel.equalsIgnoreCase(CONSISTENCY_LEVEL_ONE)) {
 549                 if(queryObject.getConsistency() == null) {
 550                     statement.setConsistencyLevel(ConsistencyLevel.ONE);
 551                 }
 552                else {
 553                    statement.setConsistencyLevel(MusicUtil.getConsistencyLevel(queryObject.getConsistency()));
 554                }
 555             }
 556             else if (consistencyLevel.equalsIgnoreCase(CONSISTENCY_LEVEL_QUORUM)) {
 557                 statement.setConsistencyLevel(ConsistencyLevel.QUORUM);
 558             }
 559
 560             results = session.execute(statement);
 561
 562         } catch (Exception ex) {
 563             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(),AppMessages.UNKNOWNERROR+ "[" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 564             throw new MusicServiceException(ex.getMessage());
 565         }
 566
 567         return results;
 568
 569     }
 570
 571     /**
 572      * This method performs DDL operations on Cassandra using consistency level ONE.
 573      *
 574      * @param queryObject Object containing cassandra prepared query and values.
 575      */
 576     public ResultSet executeOneConsistencyGet(PreparedQueryObject queryObject)
 577                     throws MusicServiceException, MusicQueryException {
 578         return executeGet(queryObject, CONSISTENCY_LEVEL_ONE);
 579     }
 580
 581     /**
 582      *
 583      * This method performs DDL operation on Cassandra using consistency level QUORUM.
 584      *
 585      * @param queryObject Object containing cassandra prepared query and values.
 586      */
 587     public ResultSet executeQuorumConsistencyGet(PreparedQueryObject queryObject)
 588                     throws MusicServiceException, MusicQueryException {
 589         return executeGet(queryObject, CONSISTENCY_LEVEL_QUORUM);
 590     }
 591
 592 }