### Question:
HeapMonitorPipeline { public static DataStream<HeapAlert> computeHeapAlerts(DataStream<HeapMetrics> statsInput, ParameterTool params) { return statsInput.flatMap(new AlertingFunction(params)).name("Create Alerts"); }  static void main(String[] args); static DataStream<HeapAlert> computeHeapAlerts(DataStream<HeapMetrics> statsInput, ParameterTool params);  }### Answer:
@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); }