WikipediaEditsSource extends RichSourceFunction<WikipediaEditEvent> { public WikipediaEditsSource() { this(DEFAULT_HOST, DEFAULT_PORT, DEFAULT_CHANNEL); } WikipediaEditsSource();  WikipediaEditsSource(String host, int port, String channel); @Override void open(Configuration parameters); @Override void close(); @Override void run(SourceContext ctx); @Override void cancel(); static final String DEFAULT_HOST; static final int DEFAULT_PORT; static final String DEFAULT_CHANNEL; }
@Test(timeout = 120 * 1000) public void testWikipediaEditsSource() throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.getConfig().disableSysoutLogging(); DataStream<WikipediaEditEvent> edits = env.addSource(new WikipediaEditsSource()); edits.addSink(new SinkFunction<WikipediaEditEvent>() { @Override public void invoke(WikipediaEditEvent value) throws Exception { throw new Exception("Expected test exception"); } }); try { env.execute(); fail("Did not throw expected Exception."); } catch (Exception e) { assertNotNull(e.getCause()); assertEquals("Expected test exception", e.getCause().getMessage()); } }