ApplicationFixed implements StreamingApplication { @Override public void populateDAG(DAG dag, Configuration conf) { FixedTuplesInputOperator wordGenerator = dag.addOperator("WordGenerator", FixedTuplesInputOperator.class); dag.getMeta(wordGenerator).getMeta(wordGenerator.output).getAttributes() .put(PortContext.QUEUE_CAPACITY, QUEUE_CAPACITY); wordGenerator.setCount(500000); WordCountOperator<byte[]> counter = dag.addOperator("Counter", new WordCountOperator<byte[]>()); dag.getMeta(counter).getMeta(counter.input).getAttributes().put(PortContext.QUEUE_CAPACITY, QUEUE_CAPACITY); dag.addStream("Generator2Counter", wordGenerator.output, counter.input).setLocality(locality); }  @Override void populateDAG(DAG dag, Configuration conf); static final int QUEUE_CAPACITY; }
@Test public void testApplication() throws IOException, Exception { LocalMode lma = LocalMode.newInstance(); new ApplicationFixed().populateDAG(lma.getDAG(), new Configuration(false)); DAG dag = lma.cloneDAG(); FixedTuplesInputOperator wordGenerator = (FixedTuplesInputOperator)dag .getOperatorMeta("WordGenerator").getOperator(); Assert.assertEquals("Queue Capacity", ApplicationFixed.QUEUE_CAPACITY, (int)dag.getMeta(wordGenerator) .getMeta(wordGenerator.output).getValue(PortContext.QUEUE_CAPACITY)); LocalMode.Controller lc = lma.getController(); lc.run(60000); }