check timestamp parsing

src/main/scala/ch/memobase/MsgFilter.scala

@@ -23,10 +23,13 @@ import java.util.Calendar
 
 import ch.memobase.models.Report
 import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.apache.logging.log4j.scala.Logging
 
 import scala.collection.JavaConverters._
+import scala.util.{Failure, Success, Try}
 
 trait MsgFilter {
+  self: Logging =>
 
   private val dateFormatter = new SimpleDateFormat("YYYY-MM-dd'T'HH:mm:ss.SSS")
 
@@ -86,18 +89,27 @@ trait MsgFilter {
       Report(rec.value()).id == recordId
 
   private val buildCreatedAfterFilter: String => FilterFun =
-    timestamp => rec => {
-      val recordTimestamp = Report(rec.value()).timestamp
-      recordTimestamp == timestamp ||
-        dateFormatter.parse(timestamp).after(dateFormatter.parse(recordTimestamp))
-    }
+    timestamp => rec =>
+      Try {
+        dateFormatter.parse(timestamp).after(dateFormatter.parse(Report(rec.value()).timestamp))
+      } match {
+        case Success(res) => res
+      case Failure(_) =>
+        logger.warn(s"Parsing of timestamp ${Report(rec.value()).timestamp} failed! Ignoring record")
+        false
+      }
 
   private val buildCreatedBeforeFilter: String => FilterFun =
-    timestamp => rec => {
-      val recordTimestamp = Report(rec.value()).timestamp
-      recordTimestamp == timestamp ||
-        dateFormatter.parse(timestamp).before(dateFormatter.parse(recordTimestamp))
-    }
+    timestamp => rec =>
+      Try {
+        dateFormatter.parse(timestamp).before(dateFormatter.parse(Report(rec.value()).timestamp))
+      } match {
+        case Success(res) => res
+        case Failure(_) =>
+          logger.warn(s"Parsing of timestamp ${Report(rec.value()).timestamp} failed! Ignoring record")
+          false
+      }
+
 
   private val buildStatusFilter: String => FilterFun =
     status => rec => Report(rec.value()).status == status