hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From te...@apache.org
Subject git commit: HBASE-10153 improve VerifyReplication to compute BADROWS more accurately (Jianwei)
Date Fri, 03 Oct 2014 03:27:36 GMT
Repository: hbase
Updated Branches:
  refs/heads/0.98 93871c670 -> a3cfd5233


HBASE-10153 improve VerifyReplication to compute BADROWS more accurately (Jianwei)


Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/a3cfd523
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/a3cfd523
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/a3cfd523

Branch: refs/heads/0.98
Commit: a3cfd5233dfbfdd57ac445acd0886df2f8bae895
Parents: 93871c6
Author: Ted Yu <tedyu@apache.org>
Authored: Fri Oct 3 03:27:24 2014 +0000
Committer: Ted Yu <tedyu@apache.org>
Committed: Fri Oct 3 03:27:24 2014 +0000

----------------------------------------------------------------------
 .../replication/VerifyReplication.java          | 67 ++++++++++++++++----
 1 file changed, 55 insertions(+), 12 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/a3cfd523/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java
index bb17e64..1088688 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java
@@ -37,6 +37,7 @@ import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
 import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
 import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
 import org.apache.hadoop.hbase.mapreduce.TableMapper;
+import org.apache.hadoop.hbase.mapreduce.TableSplit;
 import org.apache.hadoop.hbase.replication.ReplicationException;
 import org.apache.hadoop.hbase.replication.ReplicationFactory;
 import org.apache.hadoop.hbase.replication.ReplicationPeer;
@@ -78,9 +79,11 @@ public class VerifyReplication extends Configured implements Tool {
   public static class Verifier
       extends TableMapper<ImmutableBytesWritable, Put> {
 
-    public static enum Counters {GOODROWS, BADROWS}
+    public static enum Counters {
+      GOODROWS, BADROWS, ONLY_IN_SOURCE_TABLE_ROWS, ONLY_IN_PEER_TABLE_ROWS, CONTENT_DIFFERENT_ROWS}
 
     private ResultScanner replicatedScanner;
+    private Result currentCompareRowInPeerTable;
 
     /**
      * Map method that compares every scanned row with the equivalent from
@@ -111,6 +114,8 @@ public class VerifyReplication extends Configured implements Tool {
         if (versions >= 0) {
           scan.setMaxVersions(versions);
         }
+
+        final TableSplit tableSplit = (TableSplit)(context.getInputSplit());
         HConnectionManager.execute(new HConnectable<Void>(conf) {
           @Override
           public Void connect(HConnection conn) throws IOException {
@@ -119,26 +124,64 @@ public class VerifyReplication extends Configured implements Tool {
             ZKUtil.applyClusterKeyToConf(peerConf, zkClusterKey);
 
             HTable replicatedTable = new HTable(peerConf, conf.get(NAME + ".tableName"));
-            scan.setStartRow(value.getRow());
+            scan.setStartRow(tableSplit.getStartRow());
+            scan.setStopRow(tableSplit.getEndRow());
             replicatedScanner = replicatedTable.getScanner(scan);
             return null;
           }
         });
+        currentCompareRowInPeerTable = replicatedScanner.next();
       }
-      Result res = replicatedScanner.next();
-      try {
-        Result.compareResults(value, res);
-        context.getCounter(Counters.GOODROWS).increment(1);
-      } catch (Exception e) {
-        LOG.warn("Bad row", e);
-        context.getCounter(Counters.BADROWS).increment(1);
+      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);
+          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);
+            context.getCounter(Counters.GOODROWS).increment(1);
+          } catch (Exception e) {
+            logFailRowAndIncreaseCounter(context, Counters.CONTENT_DIFFERENT_ROWS, value);
+          }
+          currentCompareRowInPeerTable = replicatedScanner.next();
+          break;
+        } else if (rowCmpRet < 0) {
+          // row only exists in source table
+          logFailRowAndIncreaseCounter(context, Counters.ONLY_IN_SOURCE_TABLE_ROWS, value);
+          break;
+        } else {
+          // row only exists in peer table
+          logFailRowAndIncreaseCounter(context, Counters.ONLY_IN_PEER_TABLE_ROWS,
+            currentCompareRowInPeerTable);
+          currentCompareRowInPeerTable = replicatedScanner.next();
+        }
       }
     }
-
+    
+    private void logFailRowAndIncreaseCounter(Context context, Counters counter, Result row)
{
+      context.getCounter(counter).increment(1);
+      context.getCounter(Counters.BADROWS).increment(1);
+      LOG.error(counter.toString() + ", rowkey=" + Bytes.toString(row.getRow()));
+    }
+    
     protected void cleanup(Context context) {
       if (replicatedScanner != null) {
-        replicatedScanner.close();
-        replicatedScanner = null;
+        try {
+          while (currentCompareRowInPeerTable != null) {
+            logFailRowAndIncreaseCounter(context, Counters.ONLY_IN_PEER_TABLE_ROWS,
+              currentCompareRowInPeerTable);
+            currentCompareRowInPeerTable = replicatedScanner.next();
+          }
+        } catch (Exception e) {
+          LOG.error("fail to scan peer table in cleanup", e);
+        } finally {
+          replicatedScanner.close();
+          replicatedScanner = null;
+        }
       }
     }
   }


Mime
View raw message