	@Test
	public void testPipeline() throws Exception {


		final String alertMask = "42";

		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		HeapMetrics alert1 = testStats(0.42);
		HeapMetrics regular1 = testStats(0.452);
		HeapMetrics regular2 = testStats(0.245);
		HeapMetrics alert2 = testStats(0.9423);

		DataStreamSource<HeapMetrics> testInput = env.fromElements(alert1, alert2, regular1, regular2);
		HeapMonitorPipeline.computeHeapAlerts(testInput, ParameterTool.fromArgs(new String[]{"--alertMask", alertMask}))
			.addSink(new SinkFunction<HeapAlert>() {
				@Override
				public void invoke(HeapAlert value, Context context) {
					testOutput.add(value);
				}
			})
			.setParallelism(1);

		env.execute();

		assertEquals(Sets.newHashSet(HeapAlert.maskRatioMatch(alertMask, alert1),
				HeapAlert.maskRatioMatch(alertMask, alert2)), testOutput);
	}