ZKSessionLock implements SessionLock { protected void executeLockAction(final int lockEpoch, final LockAction func) { lockStateExecutor.submit(lockPath, new SafeRunnable() { @Override public void safeRun() { if (ZKSessionLock.this.epoch.get() == lockEpoch) { if (LOG.isTraceEnabled()) { LOG.trace("{} executing lock action '{}' under epoch {} for lock {}", new Object[]{lockId, func.getActionName(), lockEpoch, lockPath}); } func.execute(); if (LOG.isTraceEnabled()) { LOG.trace("{} executed lock action '{}' under epoch {} for lock {}", new Object[]{lockId, func.getActionName(), lockEpoch, lockPath}); } } else { if (LOG.isTraceEnabled()) { LOG.trace("{} skipped executing lock action '{}' for lock {}, since epoch is changed from {} to {}.", new Object[]{lockId, func.getActionName(), lockPath, lockEpoch, ZKSessionLock.this.epoch.get()}); } } } }); } ZKSessionLock(ZooKeeperClient zkClient,
                  String lockPath,
                  String clientId,
                  OrderedScheduler lockStateExecutor);  ZKSessionLock(ZooKeeperClient zkClient,
                         String lockPath,
                         String clientId,
                         OrderedScheduler lockStateExecutor,
                         long lockOpTimeout,
                         StatsLogger statsLogger,
                         DistributedLockContext lockContext); static String getLockPathPrefixV1(String lockPath); static String getLockPathPrefixV2(String lockPath, String clientId); static String getLockPathPrefixV3(String lockPath, String clientId, long sessionOwner); static byte[] serializeClientId(String clientId); static String deserializeClientId(byte[] data); static String getLockIdFromPath(String path); @Override ZKSessionLock setLockListener(LockListener lockListener); boolean isLockExpired(); @Override boolean isLockHeld(); @Override Future<LockWaiter> asyncTryLock(final long timeout, final TimeUnit unit); @Override void tryLock(long timeout, TimeUnit unit); @Override Future<BoxedUnit> asyncUnlock(); @Override void unlock();  }
@Test(timeout = 60000) public void testExecuteLockAction() throws Exception { String lockPath = "/test-execute-lock-action"; String clientId = "test-execute-lock-action-" + System.currentTimeMillis(); ZKSessionLock lock = new ZKSessionLock(zkc, lockPath, clientId, lockStateExecutor); final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch1 = new CountDownLatch(1); lock.executeLockAction(lock.getEpoch().get(), new LockAction() { @Override public void execute() { counter.incrementAndGet(); latch1.countDown(); } @Override public String getActionName() { return "increment1"; } }); latch1.await(); assertEquals("counter should be increased in same epoch", 1, counter.get()); final CountDownLatch latch2 = new CountDownLatch(1); lock.executeLockAction(lock.getEpoch().get() + 1, new LockAction() { @Override public void execute() { counter.incrementAndGet(); } @Override public String getActionName() { return "increment2"; } }); lock.executeLockAction(lock.getEpoch().get(), new LockAction() { @Override public void execute() { latch2.countDown(); } @Override public String getActionName() { return "countdown"; } }); latch2.await(); assertEquals("counter should not be increased in different epochs", 1, counter.get()); Promise<BoxedUnit> promise = new Promise<BoxedUnit>(); lock.executeLockAction(lock.getEpoch().get() + 1, new LockAction() { @Override public void execute() { counter.incrementAndGet(); } @Override public String getActionName() { return "increment3"; } }, promise); try { Await.result(promise); fail("Should satisfy promise with epoch changed exception."); } catch (EpochChangedException ece) { } assertEquals("counter should not be increased in different epochs", 1, counter.get()); lockStateExecutor.shutdown(); }