@Test public void testcomputeHDFSBlocksDistribution() throws Exception { HBaseTestingUtility htu = new HBaseTestingUtility(); final int DEFAULT_BLOCK_SIZE = 1024; htu.getConfiguration().setLong("dfs.blocksize", DEFAULT_BLOCK_SIZE); MiniDFSCluster cluster = null; Path testFile = null; try { String hosts[] = new String[] { "host1", "host2", "host3" }; cluster = htu.startMiniDFSCluster(hosts); cluster.waitActive(); FileSystem fs = cluster.getFileSystem(); testFile = new Path("/test1.txt"); WriteDataToHDFS(fs, testFile, 2*DEFAULT_BLOCK_SIZE); final long maxTime = System.currentTimeMillis() + 2000; boolean ok; do { ok = true; FileStatus status = fs.getFileStatus(testFile); HDFSBlocksDistribution blocksDistribution = FSUtils.computeHDFSBlocksDistribution(fs, status, 0, status.getLen()); long uniqueBlocksTotalWeight = blocksDistribution.getUniqueBlocksTotalWeight(); for (String host : hosts) { long weight = blocksDistribution.getWeight(host); ok = (ok && uniqueBlocksTotalWeight == weight); } } while (!ok && System.currentTimeMillis() < maxTime); assertTrue(ok); } finally { htu.shutdownMiniDFSCluster(); } try { String hosts[] = new String[] { "host1", "host2", "host3", "host4" }; cluster = htu.startMiniDFSCluster(hosts); cluster.waitActive(); FileSystem fs = cluster.getFileSystem(); testFile = new Path("/test2.txt"); WriteDataToHDFS(fs, testFile, 3*DEFAULT_BLOCK_SIZE); final long maxTime = System.currentTimeMillis() + 2000; long weight; long uniqueBlocksTotalWeight; do { FileStatus status = fs.getFileStatus(testFile); HDFSBlocksDistribution blocksDistribution = FSUtils.computeHDFSBlocksDistribution(fs, status, 0, status.getLen()); uniqueBlocksTotalWeight = blocksDistribution.getUniqueBlocksTotalWeight(); String tophost = blocksDistribution.getTopHosts().get(0); weight = blocksDistribution.getWeight(tophost); } while (uniqueBlocksTotalWeight != weight && System.currentTimeMillis() < maxTime); assertTrue(uniqueBlocksTotalWeight == weight); } finally { htu.shutdownMiniDFSCluster(); } try { String hosts[] = new String[] { "host1", "host2", "host3", "host4" }; cluster = htu.startMiniDFSCluster(hosts); cluster.waitActive(); FileSystem fs = cluster.getFileSystem(); testFile = new Path("/test3.txt"); WriteDataToHDFS(fs, testFile, DEFAULT_BLOCK_SIZE); final long maxTime = System.currentTimeMillis() + 2000; HDFSBlocksDistribution blocksDistribution; do { FileStatus status = fs.getFileStatus(testFile); blocksDistribution = FSUtils.computeHDFSBlocksDistribution(fs, status, 0, status.getLen()); } while (blocksDistribution.getTopHosts().size() != 3 && System.currentTimeMillis() < maxTime); assertEquals("Wrong number of hosts distributing blocks.", 3, blocksDistribution.getTopHosts().size()); } finally { htu.shutdownMiniDFSCluster(); } }
static public HDFSBlocksDistribution computeHDFSBlocksDistribution( final FileSystem fs, FileStatus status, long start, long length) throws IOException { HDFSBlocksDistribution blocksDistribution = new HDFSBlocksDistribution(); BlockLocation [] blockLocations = fs.getFileBlockLocations(status, start, length); for(BlockLocation bl : blockLocations) { String [] hosts = bl.getHosts(); long len = bl.getLength(); blocksDistribution.addHostsAndBlockWeight(hosts, len); } return blocksDistribution; }
FSUtils { static public HDFSBlocksDistribution computeHDFSBlocksDistribution( final FileSystem fs, FileStatus status, long start, long length) throws IOException { HDFSBlocksDistribution blocksDistribution = new HDFSBlocksDistribution(); BlockLocation [] blockLocations = fs.getFileBlockLocations(status, start, length); for(BlockLocation bl : blockLocations) { String [] hosts = bl.getHosts(); long len = bl.getLength(); blocksDistribution.addHostsAndBlockWeight(hosts, len); } return blocksDistribution; } }
FSUtils { static public HDFSBlocksDistribution computeHDFSBlocksDistribution( final FileSystem fs, FileStatus status, long start, long length) throws IOException { HDFSBlocksDistribution blocksDistribution = new HDFSBlocksDistribution(); BlockLocation [] blockLocations = fs.getFileBlockLocations(status, start, length); for(BlockLocation bl : blockLocations) { String [] hosts = bl.getHosts(); long len = bl.getLength(); blocksDistribution.addHostsAndBlockWeight(hosts, len); } return blocksDistribution; } protected  FSUtils(); }
FSUtils { static public HDFSBlocksDistribution computeHDFSBlocksDistribution( final FileSystem fs, FileStatus status, long start, long length) throws IOException { HDFSBlocksDistribution blocksDistribution = new HDFSBlocksDistribution(); BlockLocation [] blockLocations = fs.getFileBlockLocations(status, start, length); for(BlockLocation bl : blockLocations) { String [] hosts = bl.getHosts(); long len = bl.getLength(); blocksDistribution.addHostsAndBlockWeight(hosts, len); } return blocksDistribution; } protected  FSUtils(); static boolean isStartingWithPath(final Path rootPath, final String path); static boolean isMatchingTail(final Path pathToSearch, String pathTail); static boolean isMatchingTail(final Path pathToSearch, final Path pathTail); static FSUtils getInstance(FileSystem fs, Configuration conf); static boolean deleteDirectory(final FileSystem fs, final Path dir); static long getDefaultBlockSize(final FileSystem fs, final Path path); static short getDefaultReplication(final FileSystem fs, final Path path); static int getDefaultBufferSize(final FileSystem fs); static FSDataOutputStream create(FileSystem fs, Path path,
      FsPermission perm, InetSocketAddress[] favoredNodes); static FSDataOutputStream create(FileSystem fs, Path path,
      FsPermission perm, boolean overwrite); static FsPermission getFilePermissions(final FileSystem fs,
      final Configuration conf, final String permssionConfKey); static void checkFileSystemAvailable(final FileSystem fs); static void checkDfsSafeMode(final Configuration conf); static String getVersion(FileSystem fs, Path rootdir); static void checkVersion(FileSystem fs, Path rootdir, boolean message); static void checkVersion(FileSystem fs, Path rootdir,
      boolean message, int wait, int retries); static void setVersion(FileSystem fs, Path rootdir); static void setVersion(FileSystem fs, Path rootdir, int wait, int retries); static void setVersion(FileSystem fs, Path rootdir, String version,
      int wait, int retries); static boolean checkClusterIdExists(FileSystem fs, Path rootdir,
      int wait); static ClusterId getClusterId(FileSystem fs, Path rootdir); static void setClusterId(FileSystem fs, Path rootdir, ClusterId clusterId,
      int wait); static Path validateRootPath(Path root); static String removeRootPath(Path path, final Configuration conf); static void waitOnSafeMode(final Configuration conf,
    final long wait); static String getPath(Path p); static Path getRootDir(final Configuration c); static void setRootDir(final Configuration c, final Path root); static void setFsDefault(final Configuration c, final Path root); @SuppressWarnings("deprecation") static boolean metaRegionExists(FileSystem fs, Path rootdir); static HDFSBlocksDistribution computeHDFSBlocksDistribution(
    final FileSystem fs, FileStatus status, long start, long length); static boolean isMajorCompacted(final FileSystem fs,
      final Path hbaseRootDir); static int getTotalTableFragmentation(final HMaster master); static Map<String, Integer> getTableFragmentation(
    final HMaster master); static Map<String, Integer> getTableFragmentation(
    final FileSystem fs, final Path hbaseRootDir); static boolean isPre020FileLayout(final FileSystem fs,
    final Path hbaseRootDir); static boolean isMajorCompactedPre020(final FileSystem fs,
      final Path hbaseRootDir); static Path getTableDir(Path rootdir, final TableName tableName); static TableName getTableName(Path tablePath); static Path getNamespaceDir(Path rootdir, final String namespace); static boolean isAppendSupported(final Configuration conf); static boolean isHDFS(final Configuration conf); abstract void recoverFileLease(final FileSystem fs, final Path p,
      Configuration conf, CancelableProgressable reporter); static List<Path> getTableDirs(final FileSystem fs, final Path rootdir); static List<Path> getLocalTableDirs(final FileSystem fs, final Path rootdir); static boolean isRecoveredEdits(Path path); static List<Path> getRegionDirs(final FileSystem fs, final Path tableDir); static List<Path> getFamilyDirs(final FileSystem fs, final Path regionDir); static FileSystem getCurrentFileSystem(Configuration conf); static Map<String, Path> getTableStoreFilePathMap(Map<String, Path> map,
  final FileSystem fs, final Path hbaseRootDir, TableName tableName); static Map<String, Path> getTableStoreFilePathMap(
    final FileSystem fs, final Path hbaseRootDir); static FileStatus [] listStatus(final FileSystem fs,
      final Path dir, final PathFilter filter); static FileStatus[] listStatus(final FileSystem fs, final Path dir); static boolean delete(final FileSystem fs, final Path path, final boolean recursive); static boolean isExists(final FileSystem fs, final Path path); static void checkAccess(UserGroupInformation ugi, FileStatus file,
      FsAction action); static void logFileSystemState(final FileSystem fs, final Path root, Log LOG); static boolean renameAndSetModifyTime(final FileSystem fs, final Path src, final Path dest); static Map<String, Map<String, Float>> getRegionDegreeLocalityMappingFromFS(
      final Configuration conf); static Map<String, Map<String, Float>> getRegionDegreeLocalityMappingFromFS(
      final Configuration conf, final String desiredTable, int threadPoolSize); static void setupShortCircuitRead(final Configuration conf); static void checkShortCircuitReadBufferSize(final Configuration conf); }
FSUtils { static public HDFSBlocksDistribution computeHDFSBlocksDistribution( final FileSystem fs, FileStatus status, long start, long length) throws IOException { HDFSBlocksDistribution blocksDistribution = new HDFSBlocksDistribution(); BlockLocation [] blockLocations = fs.getFileBlockLocations(status, start, length); for(BlockLocation bl : blockLocations) { String [] hosts = bl.getHosts(); long len = bl.getLength(); blocksDistribution.addHostsAndBlockWeight(hosts, len); } return blocksDistribution; } protected  FSUtils(); static boolean isStartingWithPath(final Path rootPath, final String path); static boolean isMatchingTail(final Path pathToSearch, String pathTail); static boolean isMatchingTail(final Path pathToSearch, final Path pathTail); static FSUtils getInstance(FileSystem fs, Configuration conf); static boolean deleteDirectory(final FileSystem fs, final Path dir); static long getDefaultBlockSize(final FileSystem fs, final Path path); static short getDefaultReplication(final FileSystem fs, final Path path); static int getDefaultBufferSize(final FileSystem fs); static FSDataOutputStream create(FileSystem fs, Path path,
      FsPermission perm, InetSocketAddress[] favoredNodes); static FSDataOutputStream create(FileSystem fs, Path path,
      FsPermission perm, boolean overwrite); static FsPermission getFilePermissions(final FileSystem fs,
      final Configuration conf, final String permssionConfKey); static void checkFileSystemAvailable(final FileSystem fs); static void checkDfsSafeMode(final Configuration conf); static String getVersion(FileSystem fs, Path rootdir); static void checkVersion(FileSystem fs, Path rootdir, boolean message); static void checkVersion(FileSystem fs, Path rootdir,
      boolean message, int wait, int retries); static void setVersion(FileSystem fs, Path rootdir); static void setVersion(FileSystem fs, Path rootdir, int wait, int retries); static void setVersion(FileSystem fs, Path rootdir, String version,
      int wait, int retries); static boolean checkClusterIdExists(FileSystem fs, Path rootdir,
      int wait); static ClusterId getClusterId(FileSystem fs, Path rootdir); static void setClusterId(FileSystem fs, Path rootdir, ClusterId clusterId,
      int wait); static Path validateRootPath(Path root); static String removeRootPath(Path path, final Configuration conf); static void waitOnSafeMode(final Configuration conf,
    final long wait); static String getPath(Path p); static Path getRootDir(final Configuration c); static void setRootDir(final Configuration c, final Path root); static void setFsDefault(final Configuration c, final Path root); @SuppressWarnings("deprecation") static boolean metaRegionExists(FileSystem fs, Path rootdir); static HDFSBlocksDistribution computeHDFSBlocksDistribution(
    final FileSystem fs, FileStatus status, long start, long length); static boolean isMajorCompacted(final FileSystem fs,
      final Path hbaseRootDir); static int getTotalTableFragmentation(final HMaster master); static Map<String, Integer> getTableFragmentation(
    final HMaster master); static Map<String, Integer> getTableFragmentation(
    final FileSystem fs, final Path hbaseRootDir); static boolean isPre020FileLayout(final FileSystem fs,
    final Path hbaseRootDir); static boolean isMajorCompactedPre020(final FileSystem fs,
      final Path hbaseRootDir); static Path getTableDir(Path rootdir, final TableName tableName); static TableName getTableName(Path tablePath); static Path getNamespaceDir(Path rootdir, final String namespace); static boolean isAppendSupported(final Configuration conf); static boolean isHDFS(final Configuration conf); abstract void recoverFileLease(final FileSystem fs, final Path p,
      Configuration conf, CancelableProgressable reporter); static List<Path> getTableDirs(final FileSystem fs, final Path rootdir); static List<Path> getLocalTableDirs(final FileSystem fs, final Path rootdir); static boolean isRecoveredEdits(Path path); static List<Path> getRegionDirs(final FileSystem fs, final Path tableDir); static List<Path> getFamilyDirs(final FileSystem fs, final Path regionDir); static FileSystem getCurrentFileSystem(Configuration conf); static Map<String, Path> getTableStoreFilePathMap(Map<String, Path> map,
  final FileSystem fs, final Path hbaseRootDir, TableName tableName); static Map<String, Path> getTableStoreFilePathMap(
    final FileSystem fs, final Path hbaseRootDir); static FileStatus [] listStatus(final FileSystem fs,
      final Path dir, final PathFilter filter); static FileStatus[] listStatus(final FileSystem fs, final Path dir); static boolean delete(final FileSystem fs, final Path path, final boolean recursive); static boolean isExists(final FileSystem fs, final Path path); static void checkAccess(UserGroupInformation ugi, FileStatus file,
      FsAction action); static void logFileSystemState(final FileSystem fs, final Path root, Log LOG); static boolean renameAndSetModifyTime(final FileSystem fs, final Path src, final Path dest); static Map<String, Map<String, Float>> getRegionDegreeLocalityMappingFromFS(
      final Configuration conf); static Map<String, Map<String, Float>> getRegionDegreeLocalityMappingFromFS(
      final Configuration conf, final String desiredTable, int threadPoolSize); static void setupShortCircuitRead(final Configuration conf); static void checkShortCircuitReadBufferSize(final Configuration conf); static final String FULL_RWX_PERMISSIONS; static final boolean WINDOWS; }