LockWatchingServiceImpl implements LockWatchingService { @Override public void startWatching(LockWatchRequest locksToWatch) { Optional<LockWatches> changes = addToWatches(locksToWatch); changes.ifPresent(this::logLockWatchEvent); } LockWatchingServiceImpl(HeldLocksCollection heldLocksCollection); @VisibleForTesting  LockWatchingServiceImpl(UUID logId, HeldLocksCollection heldLocksCollection); @Override void startWatching(LockWatchRequest locksToWatch); @Override LockWatchStateUpdate getWatchStateUpdate(Optional<LockWatchVersion> lastKnownVersion); @Override ValueAndLockWatchStateUpdate<T> runTask(
            Optional<LockWatchVersion> lastKnownVersion, Supplier<T> task); @Override void registerLock(Set<LockDescriptor> locksTakenOut, LockToken token); @Override void registerUnlock(Set<LockDescriptor> unlocked);  }
@Test public void watchNewLockWatchLogsHeldLocksInRange() { LockWatchRequest request = tableRequest(); lockWatcher.startWatching(request); List<LockWatchEvent> expectedEvents = ImmutableList.of( createdEvent(request.getReferences(), ImmutableSet.of(ROW_DESCRIPTOR))); assertLoggedEvents(expectedEvents); }
@Test public void registeringWatchWithWiderScopeLogsAlreadyWatchedLocksAgain() { LockDescriptor secondRow = AtlasRowLockDescriptor.of(TABLE.getQualifiedName(), PtBytes.toBytes("other_row")); when(heldLocks.getLocks()).thenReturn(ImmutableList.of(LOCK, new ExclusiveLock(secondRow))); LockWatchRequest prefixRequest = prefixRequest(ROW); lockWatcher.startWatching(prefixRequest); LockWatchRequest entireTableRequest = tableRequest(); lockWatcher.startWatching(entireTableRequest); List<LockWatchEvent> expectedEvents = ImmutableList.of( createdEvent(prefixRequest.getReferences(), ImmutableSet.of(ROW_DESCRIPTOR)), createdEvent(entireTableRequest.getReferences(), ImmutableSet.of(ROW_DESCRIPTOR, secondRow))); assertLoggedEvents(expectedEvents); }
@Test public void registeringWatchWithOverlappingScopeLogsAlreadyWatchedLocksInScopeAgain() { LockDescriptor ab = AtlasRowLockDescriptor.of(TABLE.getQualifiedName(), PtBytes.toBytes("ab")); LockDescriptor bc = AtlasRowLockDescriptor.of(TABLE.getQualifiedName(), PtBytes.toBytes("bc")); LockDescriptor cd = AtlasRowLockDescriptor.of(TABLE.getQualifiedName(), PtBytes.toBytes("cd")); when(heldLocks.getLocks()).thenReturn(ImmutableList.of( LOCK, new ExclusiveLock(ab), new ExclusiveLock(bc), new ExclusiveLock(cd))); LockWatchReference acRange = LockWatchReferenceUtils .rowRange(TABLE, PtBytes.toBytes("a"), PtBytes.toBytes("c")); LockWatchReference bdRange = LockWatchReferenceUtils .rowRange(TABLE, PtBytes.toBytes("b"), PtBytes.toBytes("d")); lockWatcher.startWatching(LockWatchRequest.of(ImmutableSet.of(acRange))); lockWatcher.startWatching(LockWatchRequest.of(ImmutableSet.of(bdRange))); List<LockWatchEvent> expectedEvents = ImmutableList.of( createdEvent(ImmutableSet.of(acRange), ImmutableSet.of(ab, bc)), createdEvent(ImmutableSet.of(bdRange), ImmutableSet.of(bc, cd))); assertLoggedEvents(expectedEvents); }
@Test public void registeringWatchWithNarrowerScopeIsNoop() { LockWatchRequest request = tableRequest(); lockWatcher.startWatching(request); LockWatchRequest prefixRequest = prefixRequest(ROW); lockWatcher.startWatching(prefixRequest); List<LockWatchEvent> expectedEvents = ImmutableList.of( createdEvent(request.getReferences(), ImmutableSet.of(ROW_DESCRIPTOR))); assertLoggedEvents(expectedEvents); }
@Test public void onlyAlreadyWatchedRangesAreIgnoredOnRegistration() { LockWatchRequest request = tableRequest(); lockWatcher.startWatching(request); LockWatchReference newWatch = LockWatchReferenceUtils.entireTable(TABLE_2); LockWatchRequest prefixAndOtherTableRequest = LockWatchRequest.of(ImmutableSet.of( LockWatchReferenceUtils.rowPrefix(TABLE, ROW), newWatch)); lockWatcher.startWatching(prefixAndOtherTableRequest); List<LockWatchEvent> expectedEvents = ImmutableList.of( createdEvent(request.getReferences(), ImmutableSet.of(ROW_DESCRIPTOR)), createdEvent(ImmutableSet.of(newWatch), ImmutableSet.of(descriptorForOtherTable()))); assertLoggedEvents(expectedEvents); }