LocalLockMediator { public boolean lock(KeyColumn kc, T requestor, Timepoint expires) { assert null != kc; assert null != requestor; AuditRecord<T> audit = new AuditRecord<>(requestor, expires); AuditRecord<T> inmap = locks.putIfAbsent(kc, audit); boolean success = false; if (null == inmap) { if (log.isTraceEnabled()) { log.trace("New local lock created: {} namespace={} txn={}", kc, name, requestor); } success = true; } else if (inmap.equals(audit)) { success = locks.replace(kc, inmap, audit); if (log.isTraceEnabled()) { if (success) { log.trace( "Updated local lock expiration: {} namespace={} txn={} oldexp={} newexp={}", kc, name, requestor, inmap.expires, audit.expires); } else { log.trace( "Failed to update local lock expiration: {} namespace={} txn={} oldexp={} newexp={}", kc, name, requestor, inmap.expires, audit.expires); } } } else if (0 > inmap.expires.compareTo(times.getTime())) { success = locks.replace(kc, inmap, audit); if (log.isTraceEnabled()) { log.trace( "Discarding expired lock: {} namespace={} txn={} expired={}", kc, name, inmap.holder, inmap.expires); } } else { if (log.isTraceEnabled()) { log.trace( "Local lock failed: {} namespace={} txn={} (already owned by {})", kc, name, requestor, inmap); } } if (success) { expiryQueue.add(new ExpirableKeyColumn(kc, expires)); } return success; } LocalLockMediator(String name, TimestampProvider times); boolean lock(KeyColumn kc, T requestor, Timepoint expires); boolean unlock(KeyColumn kc, T requestor); String toString();  }
@Test public void testLock() throws InterruptedException { TimestampProvider times = Timestamps.MICRO; LocalLockMediator<HBaseTransaction> llm = new LocalLockMediator<>(LOCK_NAMESPACE, times); Assert.assertTrue(llm.lock(kc, mockTx1, times.getTime(0, TimeUnit.NANOSECONDS))); Assert.assertTrue(llm.lock(kc, mockTx2, times.getTime(Long.MAX_VALUE, TimeUnit.NANOSECONDS))); llm = new LocalLockMediator<>(LOCK_NAMESPACE, times); Assert.assertTrue(llm.lock(kc, mockTx1, times.getTime(Long.MAX_VALUE, TimeUnit.NANOSECONDS))); Assert.assertFalse(llm.lock(kc, mockTx2, times.getTime(Long.MAX_VALUE, TimeUnit.NANOSECONDS))); Assert.assertTrue(llm.unlock(kc, mockTx1)); Assert.assertTrue(llm.lock(kc, mockTx2, times.getTime(Long.MAX_VALUE, TimeUnit.NANOSECONDS))); }