public void testSingleCopyTasksConfiguration(boolean runCopyTasksOnTarget)
      throws Throwable {
    // Create five tables, 1 inside the custom parent path, 1 inside the database location and one separate.
    Path parentPath1 = new Path("/" + testName.getMethodName() + "/" + "external1");
    Path parent1Table1 = new Path(parentPath1,"table1");

    Path externalTablePath = new Path("/" + testName.getMethodName() + "/" + "externaltable2");

    DistributedFileSystem fs = primary.miniDFSCluster.getFileSystem();
    fs.mkdirs(parent1Table1, new FsPermission("777"));
    fs.mkdirs(externalTablePath, new FsPermission("777"));


    try (FSDataOutputStream outputStream =
        fs.create(new Path(externalTablePath, "file1.txt"))) {
      outputStream.write("3\n".getBytes());
      outputStream.write("4\n".getBytes());
    }

    try (FSDataOutputStream outputStream =
        fs.create(new Path(parent1Table1, "file1.txt"))) {
      outputStream.write("3\n".getBytes());
      outputStream.write("4\n".getBytes());
    }

    // Create a filter file for DistCp
    String filterFilePath = "/tmp/filter";
    FileWriter myWriter = new FileWriter(filterFilePath);
    myWriter.write(".*file1.txt.*");
    myWriter.close();

    List<String> withClause = Arrays
        .asList("'distcp.options.update'=''", "'" + REPL_EXTERNAL_WAREHOUSE_SINGLE_COPY_TASK.varname + "'='true'",
            "'" + HiveConf.ConfVars.REPL_RUN_DATA_COPY_TASKS_ON_TARGET.varname + "'='" + runCopyTasksOnTarget + "'",
            "'" + REPL_EXTERNAL_WAREHOUSE_SINGLE_COPY_TASK_PATHS.varname + "'='" + fs.makeQualified(parentPath1) +
                "'", "'hive.dbpath.distcp.options.filters'='" + filterFilePath + "'");

    WarehouseInstance.Tuple tuple =
        primary.run("use " + primaryDbName)
            .run("create external table table1 (id int) row format delimited fields terminated by ',' location '"
                + parent1Table1.toUri() + "'")
            .run("create external table table2 (id int) row format delimited fields terminated by ',' location '"
                + externalTablePath.toUri() + "'")
            .run("create external table table3 (id int) row format delimited fields terminated by ','")
            .run("insert into table1 values (1)")
            .run("insert into table3 values (3)")
            .dump(primaryDbName, withClause);

    Path primaryDb = new Path(primary.getDatabase(primaryDbName).getLocationUri());

    // Do a load and verify all the data is there.
    replica.load(replicatedDbName, primaryDbName, withClause)
        .run("use " + replicatedDbName)
        .run("select id from table1 ")
        .verifyResult("1")
        .run("select id from table2")
        .verifyResults(new String[] {"3", "4"})
        .run("select id from table3")
        .verifyResult("3");

    // Verify the tasks created for these tables, it should be one for the database, two for the two parent paths
    // that we configured.
    ReplicationTestUtils.assertExternalFileList(
        Arrays.asList("dbPath:" + primaryDb.getName(), "dbPath:" + parentPath1.getName(), "table2"), tuple.dumpLocation,
        primary);

    // Verify the preserve config only worked for db level path.
    Path parent1Table1_target = new Path(REPLICA_EXTERNAL_BASE, parent1Table1.toUri().getPath().replaceFirst("/", ""));
    assertFalse(fs.exists(new Path(parent1Table1_target, "file1.txt")));

    // Verify the filter config gets used for the only db level paths.
    Path externalTablePath_target =
        new Path(REPLICA_EXTERNAL_BASE, externalTablePath.toUri().getPath().replaceFirst("/", ""));
    assertTrue(fs.exists(new Path(externalTablePath_target, "file1.txt")));

    // Add more data to tables and do a incremental run and check if things stays same.
    tuple =
        primary.run("use " + primaryDbName)
            .run("insert into table1 values (4)")
            .run("insert into table2 values (5)")
            .run("insert into table3 values (2)")
            .dump(primaryDbName, withClause);

    // Do an incremental load and check if all the old and new data is there.
    replica.load(replicatedDbName, primaryDbName, withClause)
        .run("use " + replicatedDbName)
        .run("select id from table1 ")
        .verifyResults(new String[] {"1", "4"})
        .run("select id from table2 ")
        .verifyResults(new String[] {"3", "4", "5"})
        .run("select id from table3 ")
        .verifyResults(new String[] {"3", "2"});

    // The same tasks should be created.
    ReplicationTestUtils.assertExternalFileList(
        Arrays.asList("dbPath:" + primaryDb.getName(), "dbPath:" + parentPath1.getName(), "table2"), tuple.dumpLocation,
        primary);

    // Verify the filter config gets used for the only db level paths.
    assertFalse(fs.exists(new Path(parent1Table1_target, "file1.txt")));
    assertTrue(fs.exists(new Path(externalTablePath_target, "file1.txt")));

    // Clean up the filter file.
    new File(filterFilePath).delete();
  }
@param sourceTableName  -- Provide the fully qualified table name
@param replicaTableName -- Provide the fully qualified table name
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosExternalTables.java