LocalStreamService extends AbstractService implements
ConfigConstants { @SuppressWarnings("unchecked") protected Class<? extends Mapper<Text, FileStatus, NullWritable, Text>> getMapperClass() { String className = srcCluster.getCopyMapperImpl(); if (className == null || className.isEmpty()) { return CopyMapper.class; } else { try { return (Class<? extends Mapper<Text, FileStatus, NullWritable, Text>>) Class .forName(className); } catch (ClassNotFoundException e) { throw new IllegalArgumentException("Copy mapper Impl " + className + "is not found in class path"); } } } LocalStreamService(ConduitConfig config, Cluster srcCluster,
      Cluster currentCluster, CheckpointProvider provider,
      Set<String> streamsToProcess); @Override long getMSecondsTillNextRun(long currentTime); void createListing(FileSystem fs, FileStatus fileStatus,
      Map<FileStatus, String> results, Set<FileStatus> trashSet,
      Table<String, String, String> checkpointPaths); long getLastAddedPartTime(String stream); Cluster getCurrentCluster(); String getTopicNameFromDestnPath(Path destnPath); @Override String getServiceType();  }
@Test public void testCopyMapperImplMethod() throws Exception { ConduitConfigParser parser = new ConduitConfigParser( "test-lss-conduit-s3n.xml"); Set<String> streamsToProcess = new HashSet<String>(); ConduitConfig config = parser.getConfig(); streamsToProcess.addAll(config.getSourceStreams().keySet()); Set<String> clustersToProcess = new HashSet<String>(); Set<TestLocalStreamService> services = new HashSet<TestLocalStreamService>(); for (SourceStream sStream : config.getSourceStreams().values()) { for (String cluster : sStream.getSourceClusters()) { clustersToProcess.add(cluster); } } for (String clusterName : clustersToProcess) { Cluster cluster = config.getClusters().get(clusterName); cluster.getHadoopConf().set("mapred.job.tracker", super.CreateJobConf().get("mapred.job.tracker")); TestLocalStreamService service = new TestLocalStreamService(config, cluster, null, new NullCheckPointProvider(), streamsToProcess); services.add(service); } for (TestLocalStreamService service : services) { Assert.assertEquals(service.getMapperClass(), S3NCopyMapper.class); } Assert.assertEquals(ConduitMetrics.<SlidingTimeWindowGauge>getMetric("LocalStreamService",AbstractService.RETRY_MKDIR,"test1").getValue().longValue() , 0); Assert.assertEquals(ConduitMetrics.<SlidingTimeWindowGauge>getMetric("LocalStreamService",AbstractService.RETRY_CHECKPOINT,"test1").getValue().longValue() , 0); Assert.assertEquals(ConduitMetrics.<SlidingTimeWindowGauge>getMetric("LocalStreamService",AbstractService.RETRY_RENAME,"test1").getValue().longValue() , 0); Assert.assertEquals(ConduitMetrics.<SlidingTimeWindowGauge>getMetric("LocalStreamService",AbstractService.EMPTYDIR_CREATE,"test1").getValue().longValue() , 0); Assert.assertEquals(ConduitMetrics.<SlidingTimeWindowGauge>getMetric("LocalStreamService",AbstractService.FILES_COPIED_COUNT,"test1").getValue().longValue() , 0); }