HBaseKeyColumnValueStore implements KeyColumnValueStore { @Override public void acquireLock(StaticBuffer key, StaticBuffer column, StaticBuffer expectedValue, StoreTransaction txh) throws BackendException { KeyColumn lockID = new KeyColumn(key, column); logger.debug("Attempting to acquireLock on {} ", lockID); int trialCount = 0; boolean locked; while (trialCount < lockMaxRetries) { final Timepoint lockStartTime = Timestamps.MILLI.getTime(System.currentTimeMillis(), TimeUnit.MILLISECONDS); locked = localLockMediator.lock(lockID, txh, lockStartTime.add(lockExpiryTimeMs)); trialCount++; if (!locked) { handleLockFailure(txh, lockID, trialCount); } else { logger.debug("Acquired lock on {}, {}", lockID, txh); break; } } ((HBaseTransaction) txh).updateLocks(lockID, expectedValue); } HBaseKeyColumnValueStore(HBaseStoreManager storeManager, ConnectionMask cnx, String tableName, String columnFamily, String storeName, LocalLockMediator<StoreTransaction> llm); @Override void close(); @Override EntryList getSlice(KeySliceQuery query, StoreTransaction txh); @Override Map<StaticBuffer,EntryList> getSlice(List<StaticBuffer> keys, SliceQuery query, StoreTransaction txh); @Override void mutate(StaticBuffer key, List<Entry> additions, List<StaticBuffer> deletions, StoreTransaction txh); @Override void acquireLock(StaticBuffer key,
                            StaticBuffer column,
                            StaticBuffer expectedValue,
                            StoreTransaction txh); @Override KeyIterator getKeys(KeyRangeQuery query, StoreTransaction txh); @Override String getName(); @Override KeyIterator getKeys(SliceQuery query, StoreTransaction txh); static Filter getFilter(SliceQuery query);  }
@Test public void shouldSucceedInLockingIfLockMediatorSucceeds() throws BackendException { when(storeManager.getMetaDataSchema("hbase")).thenReturn(new EntryMetaData[] {EntryMetaData.TIMESTAMP}); when(storeManager.getStorageConfig()).thenReturn(storageConfig); when(storageConfig.get(GraphDatabaseConfiguration.LOCK_EXPIRE)).thenReturn( new StandardDuration(300L, TimeUnit.MILLISECONDS)); when(storageConfig.get(GraphDatabaseConfiguration.LOCK_WAIT)).thenReturn( new StandardDuration(10L, TimeUnit.MILLISECONDS)); when(storageConfig.get(GraphDatabaseConfiguration.LOCK_RETRY)).thenReturn(3); KeyColumn lockID = new KeyColumn(key, column); when(localLockMediator.lock(eq(lockID), eq(transaction), any(Timepoint.class))). thenReturn(true); HBaseKeyColumnValueStore hBaseKeyColumnValueStore = new HBaseKeyColumnValueStore(storeManager, connectionMask, "titan", "e", "hbase", localLockMediator); hBaseKeyColumnValueStore.acquireLock(key, column, expectedValue, transaction); verify(transaction).updateLocks(lockID, expectedValue); verify(localLockMediator, times(1)).lock(eq(lockID), eq(transaction), any(Timepoint.class)); }
@Test public void shouldRetryRightNumberOfTimesIfLockMediationFails() throws BackendException { when(storeManager.getMetaDataSchema("hbase")).thenReturn(new EntryMetaData[] {EntryMetaData.TIMESTAMP}); when(storeManager.getStorageConfig()).thenReturn(storageConfig); when(storageConfig.get(GraphDatabaseConfiguration.LOCK_EXPIRE)).thenReturn( new StandardDuration(300L, TimeUnit.MILLISECONDS)); when(storageConfig.get(GraphDatabaseConfiguration.LOCK_WAIT)).thenReturn( new StandardDuration(10L, TimeUnit.MILLISECONDS)); when(storageConfig.get(GraphDatabaseConfiguration.LOCK_RETRY)).thenReturn(3); KeyColumn lockID = new KeyColumn(key, column); when(localLockMediator.lock(eq(lockID), eq(transaction), any(Timepoint.class))). thenReturn(false).thenReturn(false).thenReturn(true); HBaseKeyColumnValueStore hBaseKeyColumnValueStore = new HBaseKeyColumnValueStore(storeManager, connectionMask, "titan", "e", "hbase", localLockMediator); hBaseKeyColumnValueStore.acquireLock(key, column, expectedValue, transaction); verify(transaction).updateLocks(lockID, expectedValue); verify(localLockMediator, times(3)).lock(eq(lockID), eq(transaction), any(Timepoint.class)); }
@Test(expectedExceptions = PermanentLockingException.class) public void shouldThrowExceptionAfterConfiguredRetriesIfLockMediationFails() throws BackendException { when(storeManager.getMetaDataSchema("hbase")).thenReturn(new EntryMetaData[] {EntryMetaData.TIMESTAMP}); when(storeManager.getStorageConfig()).thenReturn(storageConfig); when(storageConfig.get(GraphDatabaseConfiguration.LOCK_EXPIRE)).thenReturn( new StandardDuration(300L, TimeUnit.MILLISECONDS)); when(storageConfig.get(GraphDatabaseConfiguration.LOCK_WAIT)).thenReturn( new StandardDuration(10L, TimeUnit.MILLISECONDS)); when(storageConfig.get(GraphDatabaseConfiguration.LOCK_RETRY)).thenReturn(3); KeyColumn lockID = new KeyColumn(key, column); when(localLockMediator.lock(eq(lockID), eq(transaction), any(Timepoint.class))). thenReturn(false).thenReturn(false).thenReturn(false); HBaseKeyColumnValueStore hBaseKeyColumnValueStore = new HBaseKeyColumnValueStore(storeManager, connectionMask, "titan", "e", "hbase", localLockMediator); hBaseKeyColumnValueStore.acquireLock(key, column, expectedValue, transaction); fail("Should fail as lock could not be acquired after 3 retries."); }