public synchronized void ensureParentZNode()
      throws IOException, InterruptedException, KeeperException {
    Preconditions.checkState(!wantToBeInElection,
        "ensureParentZNode() may not be called while in the election");
    if (zkClient == null) {
      createConnection();
    }

    String pathParts[] = znodeWorkingDir.split("/");
    Preconditions.checkArgument(pathParts.length >= 1 &&
        pathParts[0].isEmpty(),
        "Invalid path: %s", znodeWorkingDir);
    
    StringBuilder sb = new StringBuilder();
    for (int i = 1; i < pathParts.length; i++) {
      sb.append("/").append(pathParts[i]);
      String prefixPath = sb.toString();
      LOG.debug("Ensuring existence of " + prefixPath);
      try {
        createWithRetries(prefixPath, new byte[]{}, zkAcl, CreateMode.PERSISTENT);
      } catch (KeeperException e) {
        if (isNodeExists(e.code())) {
          // Set ACLs for parent node, if they do not exist or are different
          try {
            setAclsWithRetries(prefixPath);
          } catch (KeeperException e1) {
            throw new IOException("Couldn't set ACLs on parent ZNode: " +
                prefixPath, e1);
          }
        } else {
          throw new IOException("Couldn't create " + prefixPath, e);
        }
      }
    }
    
    LOG.info("Successfully created " + znodeWorkingDir + " in ZK.");
  }
@Test
  public void testEnsureBaseNodeFails() throws Exception {
    Mockito.doThrow(new KeeperException.ConnectionLossException())
      .when(mockZK).create(
          Mockito.eq(ZK_PARENT_NAME), Mockito.<byte[]>any(),
          Mockito.eq(Ids.OPEN_ACL_UNSAFE), Mockito.eq(CreateMode.PERSISTENT));
    try {
      elector.ensureParentZNode();
      Assert.fail("Did not throw!");
    } catch (IOException ioe) {
      if (!(ioe.getCause() instanceof KeeperException.ConnectionLossException)) {
        throw ioe;
      }
    }
    // Should have tried three times
    Mockito.verify(mockZK, Mockito.times(3)).create(
        Mockito.eq(ZK_PARENT_NAME), Mockito.<byte[]>any(),
        Mockito.eq(Ids.OPEN_ACL_UNSAFE), Mockito.eq(CreateMode.PERSISTENT));
  }