@Test
public void testConfigure() throws IOException {

  final Job job = new Job();
  // Request the latest 3 versions of column 'info:email':
  KijiDataRequestBuilder builder = KijiDataRequest.builder();
  builder.newColumnsDef().withMaxVersions(3).add("info", "email");
  KijiDataRequest dataRequest = builder.build();
  // Read from 'here' to 'there':
  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);
  // Check that the job was configured correctly.
  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));
}