@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());
	}
}