@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 testDisconnectionAndRetryFailure()
      throws InterruptedException, IOException {
    final long deadline = System.currentTimeMillis() + TimeUnit.MILLISECONDS.convert(1, TimeUnit.SECONDS);
    final ZKLock lock = createZKLock();

    Task<Integer> inner = Task.value(1);
    Task<Integer> synchronizedTask = lock.synchronize(inner, deadline);
    _zkServer.shutdown(false);

    run(synchronizedTask);
    Assert.assertTrue(synchronizedTask.await(10, TimeUnit.SECONDS));

    Assert.assertTrue(synchronizedTask.isFailed());
    Assert.assertTrue(synchronizedTask.getError() instanceof TimeoutException);
    // make sure the inner task is NOT run.
    Assert.assertFalse(inner.isDone());
  }