private boolean mkdirs(ChannelSftp client, Path file, FsPermission permission)
      throws IOException {
    boolean created = true;
    Path workDir;
    try {
      workDir = new Path(client.pwd());
    } catch (SftpException e) {
      throw new IOException(e);
    }
    Path absolute = makeAbsolute(workDir, file);
    String pathName = absolute.getName();
    if (!exists(client, absolute)) {
      Path parent = absolute.getParent();
      created =
          (parent == null || mkdirs(client, parent, FsPermission.getDefault()));
      if (created) {
        String parentDir = parent.toUri().getPath();
        boolean succeeded = true;
        try {
          final String previousCwd = client.pwd();
          client.cd(parentDir);
          client.mkdir(pathName);
          client.cd(previousCwd);
        } catch (SftpException e) {
          throw new IOException(String.format(E_MAKE_DIR_FORPATH, pathName,
              parentDir));
        }
        created = created & succeeded;
      }
    } else if (isFile(client, absolute)) {
      throw new IOException(String.format(E_DIR_CREATE_FROMFILE, absolute));
    }
    return created;
  }
@Test
  public void testMkDirs() throws IOException {
    Path path = new Path(localDir.toUri().getPath(),
        new Path(name.getMethodName(), "subdirectory"));
    sftpFs.mkdirs(path);
    assertTrue(localFs.exists(path));
    assertTrue(localFs.getFileStatus(path).isDirectory());
    assertThat(
        ((SFTPFileSystem) sftpFs).getConnectionPool().getLiveConnCount())
        .isEqualTo(1);
  }