FlinkTopology { public static FlinkTopology createTopology(TopologyBuilder stormBuilder) { return new FlinkTopology(stormBuilder); } private  FlinkTopology(TopologyBuilder builder); static FlinkTopology createTopology(TopologyBuilder stormBuilder); StreamExecutionEnvironment getExecutionEnvironment(); JobExecutionResult execute(); StormTopology getStormTopology();  }
@Test(expected = RuntimeException.class) public void testUnknowSpout() { TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("spout", new TestSpout()); builder.setBolt("bolt", new TestBolt()).shuffleGrouping("unknown"); FlinkTopology.createTopology(builder); }
@Test(expected = RuntimeException.class) public void testUnknowBolt() { TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("spout", new TestSpout()); builder.setBolt("bolt1", new TestBolt()).shuffleGrouping("spout"); builder.setBolt("bolt2", new TestBolt()).shuffleGrouping("unknown"); FlinkTopology.createTopology(builder); }
@Test(expected = RuntimeException.class) public void testUndeclaredStream() { TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("spout", new TestSpout()); builder.setBolt("bolt", new TestBolt()).shuffleGrouping("spout"); FlinkTopology.createTopology(builder); }
@Test public void testFieldsGroupingOnMultipleSpoutOutputStreams() { TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("spout", new TestDummySpout()); builder.setBolt("sink", new TestSink()).fieldsGrouping("spout", TestDummySpout.spoutStreamId, new Fields("id")); FlinkTopology.createTopology(builder); }
@Test public void testFieldsGroupingOnMultipleBoltOutputStreams() { TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("spout", new TestDummySpout()); builder.setBolt("bolt", new TestDummyBolt()).shuffleGrouping("spout"); builder.setBolt("sink", new TestSink()).fieldsGrouping("bolt", TestDummyBolt.groupingStreamId, new Fields("id")); FlinkTopology.createTopology(builder); }