@SuppressWarnings("unchecked")
  public <T> T retryOperation(ZkOperation operation)
      throws KeeperException, InterruptedException {
    KeeperException exception = null;
    for (int i = 0; i < retryCount; i++) {
      try {
        return (T) operation.execute();
      } catch (KeeperException.ConnectionLossException e) {
        if (exception == null) {
          exception = e;
        }
        if (Thread.currentThread().isInterrupted()) {
          Thread.currentThread().interrupt();
          throw new InterruptedException();
        }
        if (Thread.currentThread() instanceof ClosableThread) {
          if (((ClosableThread) Thread.currentThread()).isClosed()) {
            throw exception;
          }
        }
        retryDelay(i);
      }
    }
    throw exception;
  }