@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 testReleaseAfterException()
      throws InterruptedException {
    final long deadline = System.currentTimeMillis() + TimeUnit.MILLISECONDS.convert(10, TimeUnit.SECONDS);
    final String path = "/testPath";
    final String errMsg = "Boom! There is an exception! But, who cares...";
    final ZKLock lock = createZKLock(path);

    Task<Integer> synchronizedTask = lock.synchronize(Task.failure(new Exception(errMsg)), deadline);

    run(synchronizedTask);

    Assert.assertTrue(synchronizedTask.await(10, TimeUnit.SECONDS));
    Assert.assertTrue(synchronizedTask.isFailed());
    Assert.assertEquals(synchronizedTask.getError().getMessage(), errMsg);

    Task<List<String>> children = _zkClient.getChildren(path);
    runAndWait("getChildren", children);

    Assert.assertEquals(children.get().size(), 0);

    Task<ZKData> acls = _zkClient.getData(path);
    runAndWait("getData", acls);

    Assert.assertEquals(acls.get().getAclList(), _acls);
  }