private void adoptHdfsOrphan()

in hbase-hbck2/src/main/java/org/apache/hbase/hbck1/HBaseFsck.java [1052:1137]
71 lines of code
13 McCabe index (conditional complexity)

  private void adoptHdfsOrphan(HbckInfo hi) throws IOException {
    Path p = hi.getHdfsRegionDir();
    FileSystem fs = p.getFileSystem(getConf());
    FileStatus[] dirs = fs.listStatus(p);
    if (dirs == null) {
      LOG.warn("Attempt to adopt orphan hdfs region skipped because no files present in " +
          p + ". This dir could probably be deleted.");
      return ;
    }

    TableName tableName = hi.getTableName();
    TableInfo tableInfo = tablesInfo.get(tableName);
    Preconditions.checkNotNull(tableInfo, "Table '" + tableName + "' not present!");
    TableDescriptor template = tableInfo.getHTD();

    // find min and max key values
    Pair<byte[],byte[]> orphanRegionRange = null;
    for (FileStatus cf : dirs) {
      String cfName= cf.getPath().getName();
      // TODO Figure out what the special dirs are
      if (cfName.startsWith(".") || cfName.equals(HConstants.SPLIT_LOGDIR_NAME)) {
        continue;
      }

      FileStatus[] hfiles = fs.listStatus(cf.getPath());
      for (FileStatus hfile : hfiles) {
        byte[] start, end;
        HFile.Reader hf = null;
        try {
          hf = HFile.createReader(fs, hfile.getPath(), CacheConfig.DISABLED,
              true, getConf());
          Optional<Cell> startKv = hf.getFirstKey();
          start = CellUtil.cloneRow(startKv.get());
          Optional<Cell> endKv = hf.getLastKey();
          end = CellUtil.cloneRow(endKv.get());
        } catch (IOException ioe) {
          LOG.warn("Problem reading orphan file " + hfile + ", skipping");
          continue;
        } catch (NullPointerException ioe) {
          LOG.warn("Orphan file " + hfile + " is possibly corrupted HFile, skipping");
          continue;
        } finally {
          if (hf != null) {
            hf.close();
          }
        }

        // expand the range to include the range of all hfiles
        if (orphanRegionRange == null) {
          // first range
          orphanRegionRange = new Pair<>(start, end);
        } else {
          // TODO add test

          // expand range only if the hfile is wider.
          if (Bytes.compareTo(orphanRegionRange.getFirst(), start) > 0) {
            orphanRegionRange.setFirst(start);
          }
          if (Bytes.compareTo(orphanRegionRange.getSecond(), end) < 0) {
            orphanRegionRange.setSecond(end);
          }
        }
      }
    }
    if (orphanRegionRange == null) {
      LOG.warn("No data in dir " + p + ", sidelining data");
      fixes++;
      sidelineRegionDir(fs, hi);
      return;
    }
    LOG.info("Min max keys are : [" + Bytes.toString(orphanRegionRange.getFirst()) + ", " +
        Bytes.toString(orphanRegionRange.getSecond()) + ")");

    // create new region on hdfs. move data into place.
    RegionInfo regionInfo = RegionInfoBuilder.newBuilder(template.getTableName())
        .setStartKey(orphanRegionRange.getFirst())
        .setEndKey(Bytes.add(orphanRegionRange.getSecond(), new byte[1]))
        .build();
    LOG.info("Creating new region : " + regionInfo);
    HRegion region = HBaseFsckRepair.createHDFSRegionDir(getConf(), regionInfo, template);
    Path target = region.getRegionFileSystem().getRegionDir();

    // rename all the data to new region
    mergeRegionDirs(target, hi);
    fixes++;
  }