@Override
    public void map(ImmutableBytesWritable row, final Result value, Context context)
      throws IOException {
      if (replicatedScanner == null) {
        Configuration conf = context.getConfiguration();
        reCompareTries = conf.getInt(NAME + ".recompareTries", 0);
        reCompareBackoffExponent = conf.getInt(NAME + ".recompareBackoffExponent", 1);
        sleepMsBeforeReCompare = conf.getInt(NAME + ".sleepMsBeforeReCompare", 0);
        if (sleepMsBeforeReCompare > 0) {
          reCompareTries = Math.max(reCompareTries, 1);
        }
        delimiter = conf.get(NAME + ".delimiter", "");
        verbose = conf.getBoolean(NAME + ".verbose", false);
        batch = conf.getInt(NAME + ".batch", -1);
        final Scan scan = new Scan();
        if (batch > 0) {
          scan.setBatch(batch);
        }
        scan.setCacheBlocks(false);
        scan.setCaching(conf.getInt(TableInputFormat.SCAN_CACHEDROWS, 1));
        long startTime = conf.getLong(NAME + ".startTime", 0);
        long endTime = conf.getLong(NAME + ".endTime", Long.MAX_VALUE);
        String families = conf.get(NAME + ".families", null);
        if (families != null) {
          String[] fams = families.split(",");
          for (String fam : fams) {
            scan.addFamily(Bytes.toBytes(fam));
          }
        }
        boolean includeDeletedCells = conf.getBoolean(NAME + ".includeDeletedCells", false);
        scan.setRaw(includeDeletedCells);
        String rowPrefixes = conf.get(NAME + ".rowPrefixes", null);
        setRowPrefixFilter(scan, rowPrefixes);
        scan.setTimeRange(startTime, endTime);
        int versions = conf.getInt(NAME + ".versions", -1);
        LOG.info("Setting number of version inside map as: " + versions);
        if (versions >= 0) {
          scan.readVersions(versions);
        }
        int reCompareThreads = conf.getInt(NAME + ".recompareThreads", 0);
        reCompareExecutor = buildReCompareExecutor(reCompareThreads, context);
        TableName tableName = TableName.valueOf(conf.get(NAME + ".tableName"));
        sourceConnection = ConnectionFactory.createConnection(conf);
        sourceTable = sourceConnection.getTable(tableName);
        tableScan = scan;

        final InputSplit tableSplit = context.getInputSplit();

        String peerQuorumAddress = conf.get(NAME + ".peerQuorumAddress");
        URI connectionUri = ConnectionRegistryFactory.tryParseAsConnectionURI(peerQuorumAddress);
        Configuration peerConf;
        if (connectionUri != null) {
          peerConf = HBaseConfiguration.create(conf);
        } else {
          peerConf =
            HBaseConfiguration.createClusterConf(conf, peerQuorumAddress, PEER_CONFIG_PREFIX);
        }
        String peerName = peerConf.get(NAME + ".peerTableName", tableName.getNameAsString());
        TableName peerTableName = TableName.valueOf(peerName);
        replicatedConnection = ConnectionFactory.createConnection(connectionUri, peerConf);
        replicatedTable = replicatedConnection.getTable(peerTableName);
        scan.withStartRow(value.getRow());

        byte[] endRow = null;
        if (tableSplit instanceof TableSnapshotInputFormat.TableSnapshotRegionSplit) {
          endRow = ((TableSnapshotInputFormat.TableSnapshotRegionSplit) tableSplit).getRegion()
            .getEndKey();
        } else {
          endRow = ((TableSplit) tableSplit).getEndRow();
        }

        scan.withStopRow(endRow);

        String peerSnapshotName = conf.get(NAME + ".peerSnapshotName", null);
        if (peerSnapshotName != null) {
          String peerSnapshotTmpDir = conf.get(NAME + ".peerSnapshotTmpDir", null);
          String peerFSAddress = conf.get(NAME + ".peerFSAddress", null);
          String peerHBaseRootAddress = conf.get(NAME + ".peerHBaseRootAddress", null);
          FileSystem.setDefaultUri(peerConf, peerFSAddress);
          CommonFSUtils.setRootDir(peerConf, new Path(peerHBaseRootAddress));
          LOG.info("Using peer snapshot:" + peerSnapshotName + " with temp dir:"
            + peerSnapshotTmpDir + " peer root uri:" + CommonFSUtils.getRootDir(peerConf)
            + " peerFSAddress:" + peerFSAddress);

          replicatedScanner = new TableSnapshotScanner(peerConf, CommonFSUtils.getRootDir(peerConf),
            new Path(peerFSAddress, peerSnapshotTmpDir), peerSnapshotName, scan, true);
        } else {
          replicatedScanner = replicatedTable.getScanner(scan);
        }
        currentCompareRowInPeerTable = replicatedScanner.next();
      }
      while (true) {
        if (currentCompareRowInPeerTable == null) {
          // reach the region end of peer table, row only in source table
          logFailRowAndIncreaseCounter(context, Counters.ONLY_IN_SOURCE_TABLE_ROWS, value, null);
          break;
        }
        int rowCmpRet = Bytes.compareTo(value.getRow(), currentCompareRowInPeerTable.getRow());
        if (rowCmpRet == 0) {
          // rowkey is same, need to compare the content of the row
          try {
            Result.compareResults(value, currentCompareRowInPeerTable, false);
            context.getCounter(Counters.GOODROWS).increment(1);
            if (verbose) {
              LOG.info(
                "Good row key: " + delimiter + Bytes.toStringBinary(value.getRow()) + delimiter);
            }
          } catch (Exception e) {
            logFailRowAndIncreaseCounter(context, Counters.CONTENT_DIFFERENT_ROWS, value,
              currentCompareRowInPeerTable);
          }
          currentCompareRowInPeerTable = replicatedScanner.next();
          break;
        } else if (rowCmpRet < 0) {
          // row only exists in source table
          logFailRowAndIncreaseCounter(context, Counters.ONLY_IN_SOURCE_TABLE_ROWS, value, null);
          break;
        } else {
          // row only exists in peer table
          logFailRowAndIncreaseCounter(context, Counters.ONLY_IN_PEER_TABLE_ROWS, null,
            currentCompareRowInPeerTable);
          currentCompareRowInPeerTable = replicatedScanner.next();
        }
      }
    }
Map method that compares every scanned row with the equivalent from a distant cluster.
@param row     The current table row key.
@param value   The columns.
@param context The current context.
@throws IOException When something is broken with the data.
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java