package org.onap.music.datastore;
+import java.util.ArrayList;
import java.util.List;
import org.onap.music.eelf.logging.EELFLoggerDelegate;
public class CassaLockStore {
private EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(CassaLockStore.class);
+ private static String table_prepend_name = "lockQ_";
public class LockObject{
public String lockRef;
public boolean createLockQueue(String keyspace, String table) throws MusicServiceException, MusicQueryException {
logger.info(EELFLoggerDelegate.applicationLogger,
"Create lock queue/table for " + keyspace+"."+table);
- table = "lockQ_"+table;
+ table = table_prepend_name+table;
String tabQuery = "CREATE TABLE IF NOT EXISTS "+keyspace+"."+table
+ " ( key text, lockReference bigint, createTime text, acquireTime text, guard bigint static, PRIMARY KEY ((key), lockReference) ) "
+ "WITH CLUSTERING ORDER BY (lockReference ASC);";
public String genLockRefandEnQueue(String keyspace, String table, String lockName) throws MusicServiceException, MusicQueryException {
logger.info(EELFLoggerDelegate.applicationLogger,
"Create lock reference for " + keyspace + "." + table + "." + lockName);
- table = "lockQ_" + table;
+ table = table_prepend_name+table;
PreparedQueryObject queryObject = new PreparedQueryObject();
return String.valueOf(lockRef);
}
+ /**
+ * Returns a result set containing the list of clients waiting for a particular lock
+ * @param keyspace
+ * @param table
+ * @param key
+ * @return list of lockrefs in the queue
+ * @throws MusicServiceException
+ * @throws MusicQueryException
+ */
+ public List<String> getLockQueue(String keyspace, String table, String key)
+ throws MusicServiceException, MusicQueryException {
+ logger.info(EELFLoggerDelegate.applicationLogger,
+ "Getting the queue for " + keyspace+"."+table+"."+key);
+ table = table_prepend_name+table;
+ String selectQuery = "select * from " + keyspace + "." + table + " where key='" + key + "';";
+ PreparedQueryObject queryObject = new PreparedQueryObject();
+ queryObject.appendQueryString(selectQuery);
+ ResultSet rs = dsHandle.executeOneConsistencyGet(queryObject);
+ ArrayList<String> lockQueue = new ArrayList<>();
+ for (Row row: rs) {
+ lockQueue.add(Long.toString(row.getLong("lockReference")));
+ }
+ return lockQueue;
+ }
+
+
+ /**
+ * Returns a result set containing the list of clients waiting for a particular lock
+ * @param keyspace
+ * @param table
+ * @param key
+ * @return size of lockrefs queue
+ * @throws MusicServiceException
+ * @throws MusicQueryException
+ */
+ public long getLockQueueSize(String keyspace, String table, String key)
+ throws MusicServiceException, MusicQueryException {
+ logger.info(EELFLoggerDelegate.applicationLogger,
+ "Getting the queue size for " + keyspace+"."+table+"."+key);
+ table = table_prepend_name+table;
+ String selectQuery = "select count(*) from " + keyspace + "." + table + " where key='" + key + "';";
+ PreparedQueryObject queryObject = new PreparedQueryObject();
+ queryObject.appendQueryString(selectQuery);
+ ResultSet rs = dsHandle.executeOneConsistencyGet(queryObject);
+ System.err.println(rs);
+
+ return rs.one().getLong("count");
+ }
+
/**
* This method returns the top of lock table/queue for the key.
public LockObject peekLockQueue(String keyspace, String table, String key) throws MusicServiceException, MusicQueryException{
logger.info(EELFLoggerDelegate.applicationLogger,
"Peek in lock table for " + keyspace+"."+table+"."+key);
- table = "lockQ_"+table;
+ table = table_prepend_name+table;
String selectQuery = "select * from "+keyspace+"."+table+" where key='"+key+"' LIMIT 1;";
PreparedQueryObject queryObject = new PreparedQueryObject();
queryObject.appendQueryString(selectQuery);
* @throws MusicQueryException
*/
public void deQueueLockRef(String keyspace, String table, String key, String lockReference) throws MusicServiceException, MusicQueryException{
- table = "lockQ_"+table;
+ table = table_prepend_name+table;
PreparedQueryObject queryObject = new PreparedQueryObject();
Long lockReferenceL = Long.parseLong(lockReference);
String deleteQuery = "delete from "+keyspace+"."+table+" where key='"+key+"' AND lockReference ="+lockReferenceL+" IF EXISTS;";
public void updateLockAcquireTime(String keyspace, String table, String key, String lockReference) throws MusicServiceException, MusicQueryException{
- table = "lockQ_"+table;
+ table = table_prepend_name+table;
PreparedQueryObject queryObject = new PreparedQueryObject();
Long lockReferenceL = Long.parseLong(lockReference);
String updateQuery = "update "+keyspace+"."+table+" set acquireTime='"+ System.currentTimeMillis()+"' where key='"+key+"' AND lockReference = "+lockReferenceL+" IF EXISTS;";
import java.io.StringWriter;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.StringTokenizer;
}
- public static ReturnType acquireLockWithLease(String fullyQualifiedKey, String lockReference, long leasePeriod) throws MusicLockingException, MusicQueryException, MusicServiceException {
+ public static ReturnType acquireLockWithLease(String fullyQualifiedKey, String lockReference, long leasePeriod)
+ throws MusicLockingException, MusicQueryException, MusicServiceException {
evictExpiredLockHolder(fullyQualifiedKey,leasePeriod);
return acquireLock(fullyQualifiedKey, lockReference);
}
- private static void evictExpiredLockHolder(String fullyQualifiedKey, long leasePeriod) throws MusicLockingException, MusicQueryException, MusicServiceException {
+ private static void evictExpiredLockHolder(String fullyQualifiedKey, long leasePeriod)
+ throws MusicLockingException, MusicQueryException, MusicServiceException {
String[] splitString = fullyQualifiedKey.split("\\.");
String keyspace = splitString[0];
}
}
- private static ReturnType isTopOfLockStore(String keyspace, String table, String primaryKeyValue, String lockReference) throws MusicLockingException, MusicQueryException, MusicServiceException {
+ private static ReturnType isTopOfLockStore(String keyspace, String table,
+ String primaryKeyValue, String lockReference)
+ throws MusicLockingException, MusicQueryException, MusicServiceException {
//return failure to lock holders too early or already evicted from the lock store
String topOfLockStoreS = getLockingServiceHandle().peekLockQueue(keyspace, table, primaryKeyValue).lockRef;
return new ReturnType(ResultType.SUCCESS, lockReference+" is top of lock store");
}
- public static ReturnType acquireLock(String fullyQualifiedKey, String lockReference) throws MusicLockingException, MusicQueryException, MusicServiceException {
+ public static ReturnType acquireLock(String fullyQualifiedKey, String lockReference)
+ throws MusicLockingException, MusicQueryException, MusicServiceException {
String[] splitString = fullyQualifiedKey.split("\\.");
String keyspace = splitString[0];
String table = splitString[1];
return new ReturnType(ResultType.SUCCESS, lockReference+" is the lock holder for the key");
}
+ /**
+ * Get the list of locks waiting in queue
+ * @param fullyQualifiedKey
+ * @return list of strings that are in the lock queue
+ * @throws MusicServiceException
+ * @throws MusicQueryException
+ * @throws MusicLockingException
+ */
+ public static List<String> getLockQueue(String fullyQualifiedKey)
+ throws MusicServiceException, MusicQueryException, MusicLockingException {
+ String[] splitString = fullyQualifiedKey.split("\\.");
+ String keyspace = splitString[0];
+ String table = splitString[1];
+ String primaryKeyValue = splitString[2];
+
+ return getLockingServiceHandle().getLockQueue(keyspace, table, primaryKeyValue);
+ }
+
+ /**
+ * Get the list of locks waiting in queue
+ * @param fullyQualifiedKey
+ * @return list of strings that are in the lock queue
+ * @throws MusicServiceException
+ * @throws MusicQueryException
+ * @throws MusicLockingException
+ */
+ public static long getLockQueueSize(String fullyQualifiedKey)
+ throws MusicServiceException, MusicQueryException, MusicLockingException {
+ String[] splitString = fullyQualifiedKey.split("\\.");
+ String keyspace = splitString[0];
+ String table = splitString[1];
+ String primaryKeyValue = splitString[2];
+ return getLockingServiceHandle().getLockQueueSize(keyspace, table, primaryKeyValue);
+ }
+
+
/**
*
*
*
*/
- public static ResultType createTable(String keyspace, String table, PreparedQueryObject tableQueryObject, String consistency) throws MusicServiceException {
+ public static ResultType createTable(String keyspace, String table,
+ PreparedQueryObject tableQueryObject, String consistency) throws MusicServiceException {
boolean result = false;
try {
* @return
* @throws MusicServiceException
*/
- public static Map<String, HashMap<String, Object>> marshallResults(ResultSet results) throws MusicServiceException {
+ public static Map<String, HashMap<String, Object>> marshallResults(ResultSet results)
+ throws MusicServiceException {
return getDSHandle().marshalData(results);
}
return getMusicLockState(fullyQualifiedKey);
}
- public static MusicLockState voluntaryReleaseLock(String fullyQualifiedKey, String lockReference) throws MusicLockingException{
+ public static MusicLockState voluntaryReleaseLock(String fullyQualifiedKey, String lockReference)
+ throws MusicLockingException{
return destroyLockRef(fullyQualifiedKey, lockReference);
}
- public static MusicLockState forciblyReleaseLock(String fullyQualifiedKey, String lockReference) throws MusicLockingException, MusicServiceException, MusicQueryException{
+ public static MusicLockState forciblyReleaseLock(String fullyQualifiedKey, String lockReference)
+ throws MusicLockingException, MusicServiceException, MusicQueryException{
String[] splitString = fullyQualifiedKey.split("\\.");
String keyspace = splitString[0];
String table = splitString[1];
* @return
* @throws MusicServiceException
*/
- public static TableMetadata returnColumnMetadata(String keyspace, String tablename) throws MusicServiceException {
+ public static TableMetadata returnColumnMetadata(String keyspace, String tablename)
+ throws MusicServiceException {
return getDSHandle().returnColumnMetadata(keyspace, tablename);
}
*
*
*/
- public static ResultType nonKeyRelatedPut(PreparedQueryObject queryObject, String consistency) throws MusicServiceException {
+ public static ResultType nonKeyRelatedPut(PreparedQueryObject queryObject, String consistency)
+ throws MusicServiceException {
// this is mainly for some functions like keyspace creation etc which does not
// really need the bells and whistles of Music locking.
boolean result = false;
* @throws MusicQueryException
*/
public static ReturnType atomicPut(String keyspaceName, String tableName, String primaryKey,
- PreparedQueryObject queryObject, Condition conditionInfo) throws MusicLockingException, MusicQueryException, MusicServiceException {
+ PreparedQueryObject queryObject, Condition conditionInfo)
+ throws MusicLockingException, MusicQueryException, MusicServiceException {
long start = System.currentTimeMillis();
String fullyQualifiedKey = keyspaceName + "." + tableName + "." + primaryKey;
* @throws MusicQueryException
*/
public static ResultSet atomicGet(String keyspaceName, String tableName, String primaryKey,
- PreparedQueryObject queryObject) throws MusicServiceException, MusicLockingException, MusicQueryException {
+ PreparedQueryObject queryObject)
+ throws MusicServiceException, MusicLockingException, MusicQueryException {
String fullyQualifiedKey = keyspaceName + "." + tableName + "." + primaryKey;
String lockReference = createLockReference(fullyQualifiedKey);
long leasePeriod = MusicUtil.getDefaultLockLeasePeriod();
resultMap.put("keyspace",keyspace);
return resultMap;
}
-
-
-// public static void main(String[] args) {
-// String x = "axe top";
-// x = x.replaceFirst("top", "sword");
-// System.out.print(x); //returns sword pickaxe
-// }
}
public class CassandraCQL {
+ public static final String keyspace = "testCassa";
+
+ public static final String table = "employees";
+
public static final String createKeySpace =
- "CREATE KEYSPACE IF NOT EXISTS testCassa WITH replication = "
+ "CREATE KEYSPACE IF NOT EXISTS " + keyspace + " WITH replication = "
+"{'class':'SimpleStrategy','replication_factor':1} AND durable_writes = true;";
- public static final String dropKeyspace = "DROP KEYSPACE IF EXISTS testCassa";
+ public static final String dropKeyspace = "DROP KEYSPACE IF EXISTS " + keyspace;
public static final String createTableEmployees =
- "CREATE TABLE IF NOT EXISTS testCassa.employees "
+ "CREATE TABLE IF NOT EXISTS " + keyspace + "." + table + " "
+ "(vector_ts text,empId uuid,empName text,empSalary varint,address Map<text,text>,PRIMARY KEY (empName)) "
+ "WITH comment='Financial Info of employees' "
+ "AND compression={'sstable_compression':'DeflateCompressor','chunk_length_kb':64} "
+ "AND compaction={'class':'SizeTieredCompactionStrategy','min_threshold':6};";
public static final String insertIntoTablePrepared1 =
- "INSERT INTO testCassa.employees (vector_ts,empId,empName,empSalary) VALUES (?,?,?,?); ";
+ "INSERT INTO " + keyspace + "." + table + " (vector_ts,empId,empName,empSalary) VALUES (?,?,?,?); ";
public static final String insertIntoTablePrepared2 =
- "INSERT INTO testCassa.employees (vector_ts,empId,empName,empSalary,address) VALUES (?,?,?,?,?);";
+ "INSERT INTO " + keyspace + "." + table + " (vector_ts,empId,empName,empSalary,address) VALUES (?,?,?,?,?);";
- public static final String selectALL = "SELECT * FROM testCassa.employees;";
+ public static final String selectALL = "SELECT * FROM " + keyspace + "." + table + ";";
public static final String selectSpecific =
- "SELECT * FROM testCassa.employees WHERE empName= ?;";
+ "SELECT * FROM " + keyspace + "." + table + " WHERE empName= ?;";
public static final String updatePreparedQuery =
- "UPDATE testCassa.employees SET vector_ts=?,address= ? WHERE empName= ?;";
+ "UPDATE " + keyspace + "." + table + " SET vector_ts=?,address= ? WHERE empName= ?;";
public static final String deleteFromTable = " ";
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
+import java.util.List;
+
+import org.junit.After;
import org.junit.AfterClass;
+import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.FixMethodOrder;
import org.junit.Test;
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
public class MusicLockStoreTest {
- static CassaDataStore dataStore;
+ static CassaDataStore dataStore;
static CassaLockStore lockStore;
static PreparedQueryObject testObject;
@BeforeClass
- public static void init() {
+ public static void init() throws MusicServiceException, MusicQueryException {
dataStore = CassandraCQL.connectToEmbeddedCassandra();
lockStore = new CassaLockStore(dataStore);
+ testObject = new PreparedQueryObject();
+ testObject.appendQueryString(CassandraCQL.createKeySpace);
+ dataStore.executePut(testObject, "eventual");
+ testObject = new PreparedQueryObject();
}
@AfterClass
}
- @Test
- public void Test1_SetUp() throws MusicServiceException, MusicQueryException {
- boolean result = false;
+ @Before
+ public void beforeEachTest() throws MusicServiceException, MusicQueryException {
testObject = new PreparedQueryObject();
testObject.appendQueryString(CassandraCQL.createKeySpace);
- result = dataStore.executePut(testObject, "eventual");;
+ dataStore.executePut(testObject, "eventual");
testObject = new PreparedQueryObject();
- testObject.appendQueryString(CassandraCQL.createTableEmployees);
- result = dataStore.executePut(testObject, "eventual");
- assertEquals(true, result);
-
}
-
- @Test
- public void Test_createLockQueue() throws MusicServiceException, MusicQueryException {
- String keyspace = "testCassa";
- String table = "employees";
- /*
- * above two hard coded since this depends on the keyspace and table created in the
- * CassandraCQL class. Need to change this..
- */
- boolean result = lockStore.createLockQueue(keyspace,table);
- assertEquals(true, result);
- }
-
- @Test
- public void Test3_ExecutePut_critical_insert() throws MusicServiceException, MusicQueryException {
- testObject = CassandraCQL.setPreparedInsertQueryObject2();
- boolean result = dataStore.executePut(testObject, "Critical");
- assertEquals(true, result);
- }
-
- @Test
- public void Test4_ExecutePut_eventual_update() throws MusicServiceException, MusicQueryException {
- testObject = CassandraCQL.setPreparedUpdateQueryObject();
- boolean result = false;
- result = dataStore.executePut(testObject, "eventual");
- assertEquals(true, result);
- }
-
- @Test
- public void Test5_ExecuteEventualGet() throws MusicServiceException, MusicQueryException {
+ @After
+ public void afterEachTest() throws MusicServiceException, MusicQueryException {
+ testObject = new PreparedQueryObject();
+ testObject.appendQueryString(CassandraCQL.dropKeyspace);
+ dataStore.executePut(testObject, "eventual");
testObject = new PreparedQueryObject();
- testObject.appendQueryString(CassandraCQL.selectALL);
- boolean result = false;
- int count = 0;
- ResultSet output = null;
- output = dataStore.executeOneConsistencyGet(testObject);
- System.out.println(output);
- ;
- for (Row row : output) {
- count++;
- System.out.println(row.toString());
- }
- if (count == 2) {
- result = true;
- }
- assertEquals(true, result);
}
+
@Test
- public void Test6_ExecuteCriticalGet() throws MusicServiceException, MusicQueryException {
- testObject = CassandraCQL.setPreparedGetQuery();
- boolean result = false;
- int count = 0;
- ResultSet output = null;
- output = dataStore.executeQuorumConsistencyGet(testObject);
- System.out.println(output);
- ;
- for (Row row : output) {
- count++;
- System.out.println(row.toString());
- }
- if (count == 1) {
- result = true;
- }
+ public void Test_createLockQueue() throws MusicServiceException, MusicQueryException {
+ boolean result = lockStore.createLockQueue(CassandraCQL.keyspace,CassandraCQL.table);
assertEquals(true, result);
}
- @Test(expected = NullPointerException.class)
- public void Test7_exception() {
- PreparedQueryObject queryObject = null;
- try {
- dataStore.executePut(queryObject, "critical");
- } catch (MusicQueryException | MusicServiceException e) {
- System.out.println(e.getMessage());
- }
- }
-
@Test
- public void Test8_columnDataType() {
- DataType data = dataStore.returnColumnDataType("testCassa", "employees", "empName");
- String datatype = data.toString();
- assertEquals("text",datatype);
+ public void Test_testGetLockQueue() throws MusicServiceException, MusicQueryException {
+ lockStore.createLockQueue(CassandraCQL.keyspace, CassandraCQL.table);
+ String lockRef = lockStore.genLockRefandEnQueue(CassandraCQL.keyspace, CassandraCQL.table, "test");
+ List<String> lockRefs = lockStore.getLockQueue(CassandraCQL.keyspace, CassandraCQL.table, "test");
+
+ assertEquals(1, lockRefs.size());
+ assertEquals(lockRef, lockRefs.get(0));
+
+ //add more locks
+ for (int i=0; i<20; i++) {
+ lockStore.genLockRefandEnQueue(CassandraCQL.keyspace, CassandraCQL.table, "test");
+ }
+ lockRefs = lockStore.getLockQueue(CassandraCQL.keyspace, CassandraCQL.table, "test");
+ assertEquals(21, lockRefs.size());
+ assertEquals(lockRef, lockRefs.get(0));
}
+
@Test
- public void Test8_columnMetdaData() {
- TableMetadata data = dataStore.returnColumnMetadata("testCassa", "employees");
- assertNotNull(data);
+ public void Test_testGetLockQueueSize() throws MusicServiceException, MusicQueryException {
+ lockStore.createLockQueue(CassandraCQL.keyspace, CassandraCQL.table);
+ String lockRef = lockStore.genLockRefandEnQueue(CassandraCQL.keyspace, CassandraCQL.table, "test");
+ assertEquals(1, lockStore.getLockQueueSize(CassandraCQL.keyspace, CassandraCQL.table, "test"));
+
+ //add more locks
+ for (int i=0; i<20; i++) {
+ lockStore.genLockRefandEnQueue(CassandraCQL.keyspace, CassandraCQL.table, "test");
+ }
+ assertEquals(21, lockStore.getLockQueueSize(CassandraCQL.keyspace, CassandraCQL.table, "test"));
}
}
Code Review / music.git / commitdiff