KijiTableMapReduceJobInput extends MapReduceJobInput { @Override public void configure(Job job) throws IOException { super.configure(job); KijiTableInputFormat inputFormat = KijiTableInputFormat.Factory.get(mInputTableURI).getInputFormat(); job.setInputFormatClass(inputFormat.getClass()); KijiTableInputFormat.configureJob(job, mInputTableURI, mDataRequest, null != mRowOptions ? mRowOptions.getStartRow() : null, null != mRowOptions ? mRowOptions.getLimitRow() : null, null != mRowOptions ? mRowOptions.getRowFilter() : null); } KijiTableMapReduceJobInput();   KijiTableMapReduceJobInput(
      KijiURI inputTableURI, KijiDataRequest dataRequest, RowOptions rowOptions); @Override void initialize(Map<String, String> params); @Override void configure(Job job); KijiURI getInputTableURI(); RowOptions getRowOptions();  }
@Test public void testConfigure() throws IOException { final Job job = new Job(); KijiDataRequestBuilder builder = KijiDataRequest.builder(); builder.newColumnsDef().withMaxVersions(3).add("info", "email"); KijiDataRequest dataRequest = builder.build(); final EntityId startRow = HBaseEntityId.fromHBaseRowKey(Bytes.toBytes("here")); final EntityId limitRow = HBaseEntityId.fromHBaseRowKey(Bytes.toBytes("there")); final KijiRowFilter filter = new StripValueRowFilter(); final KijiTableMapReduceJobInput.RowOptions rowOptions = KijiTableMapReduceJobInput.RowOptions.create(startRow, limitRow, filter); final MapReduceJobInput kijiTableJobInput = new KijiTableMapReduceJobInput(mTable.getURI(), dataRequest, rowOptions); kijiTableJobInput.configure(job); final Configuration conf = job.getConfiguration(); assertEquals( mTable.getURI(), KijiURI.newBuilder(conf.get(KijiConfKeys.KIJI_INPUT_TABLE_URI)).build()); final KijiDataRequest decoded = (KijiDataRequest) SerializationUtils.deserialize( Base64.decodeBase64(conf.get(KijiConfKeys.KIJI_INPUT_DATA_REQUEST))); assertEquals(dataRequest, decoded); final String confStartRow = Base64.encodeBase64String(startRow.getHBaseRowKey()); final String confLimitRow = Base64.encodeBase64String(limitRow.getHBaseRowKey()); assertEquals(confStartRow, conf.get(KijiConfKeys.KIJI_START_ROW_KEY)); assertEquals(confLimitRow, conf.get(KijiConfKeys.KIJI_LIMIT_ROW_KEY)); assertEquals(filter.toJson().toString(), conf.get(KijiConfKeys.KIJI_ROW_FILTER)); }