MirrorStreamValidator extends AbstractStreamValidator { public MirrorStreamValidator(ConduitConfig conduitConfig, String streamName, String clusterName, boolean fix, Date startTime, Date stopTime, int numThreads) { this.conduitConfig = conduitConfig; this.streamName = streamName; this.fix = fix; mergedCluster = conduitConfig.getPrimaryClusterForDestinationStream(streamName); mirrorCluster = conduitConfig.getClusters().get(clusterName); this.startTime = startTime; this.stopTime = stopTime; this.numThreads = numThreads; } MirrorStreamValidator(ConduitConfig conduitConfig,
      String streamName, String clusterName, boolean fix, Date startTime,
      Date stopTime, int numThreads); void execute(); List<Path> getHolesInMerge(); List<Path> getHolesInMirror();  }
@Test public void testMirrorStreamValidator() throws Exception { Date date = new Date(); Calendar cal = Calendar.getInstance(); cal.setTime(date); cal.add(Calendar.MINUTE, 1); Date nextDate = cal.getTime(); cal.add(Calendar.MINUTE, 4); Date stopDate = cal.getTime(); ConduitConfig config = setup("test-mirror-validator-conduit.xml"); cleanUp(config); createMergeData(config, date); Set<String> streamsSet = config.getSourceStreams().keySet(); FileSystem fs = FileSystem.getLocal(new Configuration()); String auditSrcJar = FileUtil.findContainingJar( com.inmobi.messaging.util.AuditUtil.class); for (String streamName : streamsSet) { for (Cluster cluster : config.getClusters().values()) { if (cluster.getMirroredStreams().contains(streamName)) { jarsPath = new Path(cluster.getTmpPath(), "jars"); auditUtilJarDestPath = new Path(jarsPath, "messaging-client-core.jar"); fs.copyFromLocalFile(new Path(auditSrcJar), auditUtilJarDestPath); createMirrorData(config, streamName, cluster, date); testStartTimeBeyondRetention(config,streamName, cluster.getName(),date, nextDate); testMirrorValidatorVerify(config,streamName, cluster.getName(),date, nextDate, false, false); testMirrorValidatorVerify(config,streamName, cluster.getName(),date, stopDate, false, true); testMirrorValidatorFixWithHolesInSource(config,streamName, cluster.getName(), date, stopDate); testMirrorValidatorFix(config,streamName, cluster.getName(), date, stopDate); testMirrorValidatorVerify(config,streamName, cluster.getName(),date, stopDate, true, true); } } } cleanUp(config); }