@Test
  public void testReplicationMetricForFailedIteration() throws Throwable {
    String name = testName.getMethodName();
    String primaryDbName = createDB(name, driver);
    String replicaDbName = "tgtDb";
    try {
      isMetricsEnabledForTests(true);
      MetricCollector collector = MetricCollector.getInstance();
      run("create table " + primaryDbName + ".t1 (id int) STORED AS TEXTFILE", driver);
      run("insert into " + primaryDbName + ".t1 values(1)", driver);
      Tuple dumpData = replDumpDb(primaryDbName);

      ReplicationMetric metric = collector.getMetrics().getLast();
      assertEquals(metric.getProgress().getStatus(), Status.SUCCESS);

      run("repl load " + primaryDbName + " into " + replicaDbName, driverMirror);

      Path nonRecoverableFile = new Path(new Path(dumpData.dumpLocation), ReplAck.NON_RECOVERABLE_MARKER.toString());
      FileSystem fs = new Path(dumpData.dumpLocation).getFileSystem(hconf);
      fs.create(nonRecoverableFile);

      verifyFail("REPL DUMP " + primaryDbName, driver);

      metric = collector.getMetrics().getLast();
      assertEquals(metric.getProgress().getStatus(), Status.FAILED_ADMIN);
      assertEquals(metric.getProgress().getStages().get(0).getErrorLogPath(), nonRecoverableFile.toString());

      verifyFail("REPL DUMP " + primaryDbName, driver);
      metric = collector.getMetrics().getLast();
      assertEquals(metric.getProgress().getStatus(), Status.FAILED_ADMIN);
      assertEquals(metric.getProgress().getStages().get(0).getErrorLogPath(), nonRecoverableFile.toString());

      fs.delete(nonRecoverableFile, true);
      dumpData = replDumpDb(primaryDbName);

      metric = collector.getMetrics().getLast();
      assertEquals(metric.getProgress().getStatus(), Status.SUCCESS);

      run("ALTER DATABASE " + replicaDbName + " SET DBPROPERTIES('" + ReplConst.REPL_INCOMPATIBLE + "'='true')", driverMirror);

      verifyFail("REPL LOAD " + primaryDbName + " INTO " + replicaDbName, driverMirror);

      nonRecoverableFile = new Path(new Path(dumpData.dumpLocation), ReplAck.NON_RECOVERABLE_MARKER.toString());
      assertTrue(fs.exists(nonRecoverableFile));

      metric = collector.getMetrics().getLast();
      assertEquals(metric.getProgress().getStatus(), Status.FAILED_ADMIN);
      assertEquals(metric.getProgress().getStages().get(0).getErrorLogPath(), nonRecoverableFile.toString());

      verifyFail("REPL LOAD " + primaryDbName + " INTO " + replicaDbName, driverMirror);

      metric = collector.getMetrics().getLast();
      assertEquals(metric.getProgress().getStatus(), Status.FAILED_ADMIN);
      assertEquals(metric.getProgress().getStages().get(0).getErrorLogPath(), nonRecoverableFile.toString());
    } finally {
      isMetricsEnabledForTests(false);
    }
  }
Verify replication when string partition column value has special chars
@throws IOException
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenarios.java