HBaseAddElementsFromHdfsJobFactory implements AddElementsFromHdfsJobFactory { @Override public void setupJob(final Job job, final AddElementsFromHdfs operation, final String mapperGeneratorClassName, final Store store) throws IOException { job.setJarByClass(getClass()); job.setJobName(getJobName(mapperGeneratorClassName, operation.getOutputPath())); setupMapper(job); setupOutput(job, operation, (HBaseStore) store); job.setSortComparatorClass(HBaseComparator.class); setupReducer(job); } HBaseAddElementsFromHdfsJobFactory(final Configuration configuration);  HBaseAddElementsFromHdfsJobFactory(); @Override void prepareStore(final Store store); @Override JobConf createJobConf(final AddElementsFromHdfs operation, final String mapperGeneratorClassName, final Store store); @Override void setupJob(final Job job, final AddElementsFromHdfs operation, final String mapperGeneratorClassName, final Store store);  }
@Test public void shouldSetupJob() throws IOException, StoreException { final JobConf localConf = createLocalConf(); final FileSystem fs = FileSystem.getLocal(localConf); fs.mkdirs(new Path(outputDir)); final JobFactory factory = getJobFactory(); final Job job = mock(Job.class); final MapReduce operation = getMapReduceOperation(); final HBaseStore store = new SingleUseMiniHBaseStore(); final Schema schema = Schema.fromJson(StreamUtil.schemas(getClass())); final HBaseProperties properties = HBaseProperties.loadStoreProperties(StreamUtil.storeProps(getClass())); store.initialise("graphId", schema, properties); given(job.getConfiguration()).willReturn(localConf); factory.setupJob(job, operation, TextMapperGeneratorImpl.class.getName(), store); verify(job).setJarByClass(factory.getClass()); verify(job).setJobName("Ingest HDFS data: Generator=" + TextMapperGeneratorImpl.class.getName() + ", output=" + outputDir); verify(job).setMapperClass(AddElementsFromHdfsMapper.class); verify(job).setMapOutputKeyClass(ImmutableBytesWritable.class); verify(job).setMapOutputValueClass(KeyValue.class); verify(job).setReducerClass(AddElementsFromHdfsReducer.class); verify(job).setOutputKeyClass(ImmutableBytesWritable.class); verify(job).setOutputValueClass(KeyValue.class); verify(job).setOutputFormatClass(HFileOutputFormat2.class); assertEquals(fs.makeQualified(new Path(outputDir)).toString(), job.getConfiguration().get("mapreduce.output.fileoutputformat.outputdir")); verify(job).setNumReduceTasks(1); }