@Override
  public <T> Task<T> synchronize(Task<T> task, long deadline) {
    return PlanLocal.get(getPlanLocalKey(), LockInternal.class)
        .flatMap(lockInternal -> {
          if (lockInternal != null) {
            // we already acquire the lock, add count only.
            lockInternal._lockCount++;
            return Task.value(lockInternal._lockNode);
          } else {
            // try acquire.
            return acquire(deadline);
          }
        })
        /* run the given task with toTry() */
        .flatMap(unused -> task).toTry()
        /* release the lock and unwind the result */
        .flatMap(result -> release().andThen(unwind(result)));
  }
@Test
  public void testAcquireTimeout()
      throws InterruptedException {
    final long timeout = 1000;
    final ZKLock lock = createZKLock();
    final SettablePromise<Void> latch = Promises.settable();

    Task<Integer> t1 = lock.synchronize(Task.async(context -> {
      // notify t2 after t1 acquires the lock.
      latch.done(null);
      SettablePromise<Integer> promise = Promises.settable();
      context.createTimer(timeout * 2, TimeUnit.MILLISECONDS, Task.action(() -> promise.done(1)));
      return promise;
    }), System.currentTimeMillis() + 3 * timeout);

    Task<Integer> inner = Task.value(1);
    Task<Integer> t2 = Task.async(() -> latch)
        .andThen(lock.synchronize(inner, System.currentTimeMillis() + timeout));

    run(t1);
    run(t2);
    Assert.assertTrue(t1.await(30, TimeUnit.SECONDS));
    Assert.assertTrue(t2.await(30, TimeUnit.SECONDS));
    Assert.assertEquals((int) t1.get(), 1);
    Assert.assertTrue(t2.isFailed());
    Assert.assertTrue(t2.getError() instanceof TimeoutException);
    Assert.assertFalse(inner.isDone());
  }