@Override
    protected void cleanup(Context context) {
      if (reCompareExecutor != null && !reCompareExecutor.isShutdown()) {
        reCompareExecutor.shutdown();
        try {
          boolean terminated = reCompareExecutor.awaitTermination(1, TimeUnit.MINUTES);
          if (!terminated) {
            List<Runnable> queue = reCompareExecutor.shutdownNow();
            for (Runnable runnable : queue) {
              ((VerifyReplicationRecompareRunnable) runnable).fail();
            }

            terminated = reCompareExecutor.awaitTermination(1, TimeUnit.MINUTES);

            if (!terminated) {
              int activeCount = Math.max(1, reCompareExecutor.getActiveCount());
              LOG.warn("Found {} possible recompares still running in the executable"
                + " incrementing BADROWS and FAILED_RECOMPARE", activeCount);
              context.getCounter(Counters.BADROWS).increment(activeCount);
              context.getCounter(Counters.FAILED_RECOMPARE).increment(activeCount);
            }
          }
        } catch (InterruptedException e) {
          throw new RuntimeException("Failed to await executor termination in cleanup", e);
        }
      }
      if (replicatedScanner != null) {
        try {
          while (currentCompareRowInPeerTable != null) {
            logFailRowAndIncreaseCounter(context, Counters.ONLY_IN_PEER_TABLE_ROWS, null,
              currentCompareRowInPeerTable);
            currentCompareRowInPeerTable = replicatedScanner.next();
          }
        } catch (Exception e) {
          LOG.error("fail to scan peer table in cleanup", e);
        } finally {
          replicatedScanner.close();
          replicatedScanner = null;
        }
      }

      if (sourceTable != null) {
        try {
          sourceTable.close();
        } catch (IOException e) {
          LOG.error("fail to close source table in cleanup", e);
        }
      }
      if (sourceConnection != null) {
        try {
          sourceConnection.close();
        } catch (Exception e) {
          LOG.error("fail to close source connection in cleanup", e);
        }
      }

      if (replicatedTable != null) {
        try {
          replicatedTable.close();
        } catch (Exception e) {
          LOG.error("fail to close replicated table in cleanup", e);
        }
      }
      if (replicatedConnection != null) {
        try {
          replicatedConnection.close();
        } catch (Exception e) {
          LOG.error("fail to close replicated connection in cleanup", e);
        }
      }
    }
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