@Test
public void testProducer() throws Exception {

  // Create some maps for KV stores for this test.
  // A small map whose value will be checked.
  final Map<String, String> smallMap = new HashMap<String, String>();
  smallMap.put("lorem", "ipsum");
  // A larger map to ensure that we can safely encode non-trivial amounts of data.
  final Map<String, String> largeMap = new HashMap<String, String>(LARGE_KV_SIZE);
  for (int i = 0; i < LARGE_KV_SIZE; i++) {
    largeMap.put(Integer.toString(i), Integer.toString(i));
  }
  final KijiURI tableURI = KijiURI.newBuilder(getKiji().getURI()).withTableName("test").build();
  final KijiMapReduceJob job = KijiProduceJobBuilder.create()
      .withConf(getConf())
      .withProducer(TestingProducer.class)
      .withInputTable(tableURI)
      .withOutput(MapReduceJobOutputs.newDirectKijiTableMapReduceJobOutput(tableURI))
      .withStore("small", InMemoryMapKeyValueStore.fromMap(smallMap))
      .withStore("large", InMemoryMapKeyValueStore.fromMap(largeMap))
      .build();
  // Be sure the job runs successfully and to completion.
  // If the producer finishes, the first_name of the main row will be changed to "lorem".
  assertTrue(job.run());
  final KijiTable table = getKiji().openTable("test");
  final KijiTableReader reader = table.openTableReader();
  String value = reader
      .get(table.getEntityId("Marsellus Wallace"), KijiDataRequest.create("info", "first_name"))
      .getMostRecentValue("info", "first_name").toString();
  assertEquals("Expected producer output not present. Did producer run successfully?",
      "lorem", value);
  reader.close();
  table.release();
}