HBASE-29359 VerifyReplication needs to join on its verification tasks

Author: ndimiduk

hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java

@@ -156,7 +156,6 @@ public enum Counters {
     private int sleepMsBeforeReCompare;
     private String delimiter = "";
     private boolean verbose = false;
-    private int batch = -1;
 
     /**
      * Map method that compares every scanned row with the equivalent from a distant cluster.
@@ -178,7 +177,7 @@ public void map(ImmutableBytesWritable row, final Result value, Context context)
         }
         delimiter = conf.get(NAME + ".delimiter", "");
         verbose = conf.getBoolean(NAME + ".verbose", false);
-        batch = conf.getInt(NAME + ".batch", -1);
+        int batch = conf.getInt(NAME + ".batch", -1);
         final Scan scan = new Scan();
         if (batch > 0) {
           scan.setBatch(batch);
@@ -200,7 +199,7 @@ public void map(ImmutableBytesWritable row, final Result value, Context context)
         setRowPrefixFilter(scan, rowPrefixes);
         scan.setTimeRange(startTime, endTime);
         int versions = conf.getInt(NAME + ".versions", -1);
-        LOG.info("Setting number of version inside map as: " + versions);
+        LOG.info("Setting number of version inside map as: {}", versions);
         if (versions >= 0) {
           scan.readVersions(versions);
         }
@@ -245,9 +244,7 @@ public void map(ImmutableBytesWritable row, final Result value, Context context)
           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);
+          LOG.info("Using peer snapshot:{} with temp dir:{} peer root uri:{} peerFSAddress:{}", peerSnapshotName, peerSnapshotTmpDir, CommonFSUtils.getRootDir(peerConf), peerFSAddress);
 
           replicatedScanner = new TableSnapshotScanner(peerConf, CommonFSUtils.getRootDir(peerConf),
             new Path(peerFSAddress, peerSnapshotTmpDir), peerSnapshotName, scan, true);
@@ -270,7 +267,7 @@ public void map(ImmutableBytesWritable row, final Result value, Context context)
             context.getCounter(Counters.GOODROWS).increment(1);
             if (verbose) {
               LOG.info(
-                "Good row key: " + delimiter + Bytes.toStringBinary(value.getRow()) + delimiter);
+                "Good row key: {}", delimiter + Bytes.toStringBinary(value.getRow()) + delimiter);
             }
           } catch (Exception e) {
             logFailRowAndIncreaseCounter(context, Counters.CONTENT_DIFFERENT_ROWS, value,
@@ -291,7 +288,6 @@ public void map(ImmutableBytesWritable row, final Result value, Context context)
       }
     }
 
-    @SuppressWarnings("FutureReturnValueIgnored")
     private void logFailRowAndIncreaseCounter(Context context, Counters counter, Result row,
       Result replicatedRow) {
       byte[] rowKey = getRow(row, replicatedRow);
@@ -311,7 +307,11 @@ private void logFailRowAndIncreaseCounter(Context context, Counters counter, Res
         return;
       }
 
-      reCompareExecutor.submit(runnable);
+      try {
+        reCompareExecutor.submit(runnable).get();
+      } catch (Exception e) {
+        throw new RuntimeException(e);
+      }
     }
 
     @Override
@@ -389,18 +389,14 @@ protected void cleanup(Context context) {
 
   private static Pair<ReplicationPeerConfig, Configuration>
     getPeerQuorumConfig(final Configuration conf, String peerId) throws IOException {
-    ZKWatcher localZKW = null;
-    try {
-      localZKW = new ZKWatcher(conf, "VerifyReplication", new Abortable() {
-        @Override
-        public void abort(String why, Throwable e) {
-        }
+    try (ZKWatcher localZKW = new ZKWatcher(conf, "VerifyReplication", new Abortable() {
+      @Override public void abort(String why, Throwable e) {
+      }
 
-        @Override
-        public boolean isAborted() {
-          return false;
-        }
-      });
+      @Override public boolean isAborted() {
+        return false;
+      }
+    })) {
       ReplicationPeerStorage storage =
         ReplicationStorageFactory.getReplicationPeerStorage(FileSystem.get(conf), localZKW, conf);
       ReplicationPeerConfig peerConfig = storage.getPeerConfig(peerId);
@@ -409,10 +405,6 @@ public boolean isAborted() {
     } catch (ReplicationException e) {
       throw new IOException("An error occurred while trying to connect to the remote peer cluster",
         e);
-    } finally {
-      if (localZKW != null) {
-        localZKW.close();
-      }
     }
   }
 
@@ -471,25 +463,25 @@ public Job createSubmittableJob(Configuration conf, String[] args) throws IOExce
       peerConfigPair = getPeerQuorumConfig(conf, peerId);
       ReplicationPeerConfig peerConfig = peerConfigPair.getFirst();
       peerQuorumAddress = peerConfig.getClusterKey();
-      LOG.info("Peer Quorum Address: " + peerQuorumAddress + ", Peer Configuration: "
-        + peerConfig.getConfiguration());
+      LOG.info("Peer Quorum Address: {}, Peer Configuration: {}", peerQuorumAddress,
+        peerConfig.getConfiguration());
       conf.set(NAME + ".peerQuorumAddress", peerQuorumAddress);
       HBaseConfiguration.setWithPrefix(conf, PEER_CONFIG_PREFIX,
         peerConfig.getConfiguration().entrySet());
     } else {
       assert this.peerQuorumAddress != null;
       peerQuorumAddress = this.peerQuorumAddress;
-      LOG.info("Peer Quorum Address: " + peerQuorumAddress);
+      LOG.info("Peer Quorum Address: {}", peerQuorumAddress);
       conf.set(NAME + ".peerQuorumAddress", peerQuorumAddress);
     }
 
     if (peerTableName != null) {
-      LOG.info("Peer Table Name: " + peerTableName);
+      LOG.info("Peer Table Name: {}", peerTableName);
       conf.set(NAME + ".peerTableName", peerTableName);
     }
 
     conf.setInt(NAME + ".versions", versions);
-    LOG.info("Number of version: " + versions);
+    LOG.info("Number of version: {}", versions);
 
     conf.setInt(NAME + ".recompareTries", reCompareTries);
     conf.setInt(NAME + ".recompareBackoffExponent", reCompareBackoffExponent);
@@ -524,7 +516,7 @@ public Job createSubmittableJob(Configuration conf, String[] args) throws IOExce
     }
     if (versions >= 0) {
       scan.readVersions(versions);
-      LOG.info("Number of versions set to " + versions);
+      LOG.info("Number of versions set to {}", versions);
     }
     if (families != null) {
       String[] fams = families.split(",");
@@ -537,8 +529,8 @@ public Job createSubmittableJob(Configuration conf, String[] args) throws IOExce
 
     if (sourceSnapshotName != null) {
       Path snapshotTempPath = new Path(sourceSnapshotTmpDir);
-      LOG.info(
-        "Using source snapshot-" + sourceSnapshotName + " with temp dir:" + sourceSnapshotTmpDir);
+      LOG.info("Using source snapshot-{} with temp dir:{}", sourceSnapshotName,
+        sourceSnapshotTmpDir);
       TableMapReduceUtil.initTableSnapshotMapperJob(sourceSnapshotName, scan, Verifier.class, null,
         null, job, true, snapshotTempPath);
       restoreSnapshotForPeerCluster(conf, peerQuorumAddress);
@@ -819,7 +811,7 @@ private boolean isPeerQuorumAddress(String cmd) {
    * @param errorMsg Error message. Can be null.
    */
   private static void printUsage(final String errorMsg) {
-    if (errorMsg != null && errorMsg.length() > 0) {
+    if (errorMsg != null && !errorMsg.isEmpty()) {
       System.err.println("ERROR: " + errorMsg);
     }
     System.err.println("Usage: verifyrep [--starttime=X]"
@@ -914,7 +906,7 @@ private static void printUsage(final String errorMsg) {
         + "2181:/cluster-b \\\n" + "     TestTable");
   }
 
-  private static ThreadPoolExecutor buildReCompareExecutor(int maxThreads, Mapper.Context context) {
+  private static ThreadPoolExecutor buildReCompareExecutor(int maxThreads, Mapper<?, ?, ?, ?>.Context context) {
     if (maxThreads == 0) {
       return null;
     }
@@ -923,7 +915,7 @@ private static ThreadPoolExecutor buildReCompareExecutor(int maxThreads, Mapper.
       buildRejectedReComparePolicy(context));
   }
 
-  private static CallerRunsPolicy buildRejectedReComparePolicy(Mapper.Context context) {
+  private static CallerRunsPolicy buildRejectedReComparePolicy(Mapper<?, ?, ?, ?>.Context context) {
     return new CallerRunsPolicy() {
       @Override
       public void rejectedExecution(Runnable runnable, ThreadPoolExecutor e) {
@@ -938,9 +930,10 @@ public void rejectedExecution(Runnable runnable, ThreadPoolExecutor e) {
   @Override
   public int run(String[] args) throws Exception {
     Configuration conf = this.getConf();
-    Job job = createSubmittableJob(conf, args);
-    if (job != null) {
-      return job.waitForCompletion(true) ? 0 : 1;
+    try (Job job = createSubmittableJob(conf, args)) {
+      if (job != null) {
+        return job.waitForCompletion(true) ? 0 : 1;
+      }
     }
     return 1;
   }

hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplicationRecompareRunnable.java

@@ -34,7 +34,7 @@ public class VerifyReplicationRecompareRunnable implements Runnable {
   private static final Logger LOG =
     LoggerFactory.getLogger(VerifyReplicationRecompareRunnable.class);
 
-  private final Mapper.Context context;
+  private final Mapper<?, ?, ?, ?>.Context context;
   private final VerifyReplication.Verifier.Counters originalCounter;
   private final String delimiter;
   private final byte[] row;
@@ -50,7 +50,7 @@ public class VerifyReplicationRecompareRunnable implements Runnable {
   private Result sourceResult;
   private Result replicatedResult;
 
-  public VerifyReplicationRecompareRunnable(Mapper.Context context, Result sourceResult,
+  public VerifyReplicationRecompareRunnable(Mapper<?, ?, ?, ?>.Context context, Result sourceResult,
     Result replicatedResult, VerifyReplication.Verifier.Counters originalCounter, String delimiter,
     Scan tableScan, Table sourceTable, Table replicatedTable, int reCompareTries,
     int sleepMsBeforeReCompare, int reCompareBackoffExponent, boolean verbose) {