From 27e07926f4175fb3107b7f04a13479758cff3b0c Mon Sep 17 00:00:00 2001
From: Lida He <lida.he@emc.com>
Date: Thu, 17 May 2018 17:13:17 -0400
Subject: [PATCH] apply filter earlier to let in just 500 responses

Signed-off-by: Lida He <lida.he@emc.com>
---
 .../io/pravega/examples/flink/alert/HighCountAlerter.java    | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/flink-examples/src/main/java/io/pravega/examples/flink/alert/HighCountAlerter.java b/flink-examples/src/main/java/io/pravega/examples/flink/alert/HighCountAlerter.java
index bf08fd9a..59211d13 100644
--- a/flink-examples/src/main/java/io/pravega/examples/flink/alert/HighCountAlerter.java
+++ b/flink-examples/src/main/java/io/pravega/examples/flink/alert/HighCountAlerter.java
@@ -94,7 +94,7 @@ public void flatMap(AccessLog value, Collector<ResponseCount> out) throws Except
                     out.collect(new ResponseCount(value.getStatus(), 1));
                 }
             }).filter((FilterFunction<ResponseCount>) count -> {
-                  return !count.response.isEmpty();
+                  return !count.response.equals("500");
               }).keyBy("response")
               .timeWindow(Time.seconds(Constants.ALERT_WINDOW), Time.seconds(Constants.ALERT_INTERVAL))
               .sum("count");
@@ -107,8 +107,7 @@ public void flatMap(AccessLog value, Collector<ResponseCount> out) throws Except
             .where(new SimpleCondition<ResponseCount>() {
                 @Override
                 public boolean filter(ResponseCount value) throws Exception {
-                    return value.count >= Constants.ALERT_THRESHOLD &&
-                           value.response.equals("500");
+                    return value.count >= Constants.ALERT_THRESHOLD;
                 }
             });