fix: require event time when dropping messages (#86)

Signed-off-by: Keran Yang <[email protected]>
numaproj · Dec 4, 2023 · 06b90a0 · 06b90a0
1 parent a072f6d
commit 06b90a0
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 9 deletions.
diff --git a/...numaproj/numaflow/examples/sourcetransformer/eventtimefilter/EventTimeFilterFunction.java b/...numaproj/numaflow/examples/sourcetransformer/eventtimefilter/EventTimeFilterFunction.java
@@ -12,8 +12,8 @@
  * This is a simple User Defined Function example which receives a message, applies the following
  * data transformation, and returns the message.
  * <p>
- * If the message event time is before year 2022, drop the message. If it's within year 2022, update
- * the tag to "within_year_2022" and update the message event time to Jan 1st 2022.
+ * If the message event time is before year 2022, drop the message with the original event time.
+ * If it's within year 2022, update the tag to "within_year_2022" and update the message event time to Jan 1st 2022.
  * Otherwise, (exclusively after year 2022), update the tag to "after_year_2022" and update the
  * message event time to Jan 1st 2023.
  */
@@ -31,7 +31,7 @@ public MessageList processMessage(String[] keys, Datum data) {
         Instant eventTime = data.getEventTime();
 
         if (eventTime.isBefore(januaryFirst2022)) {
-            return MessageList.newBuilder().addMessage(Message.toDrop()).build();
+            return MessageList.newBuilder().addMessage(Message.toDrop(eventTime)).build();
         } else if (eventTime.isBefore(januaryFirst2023)) {
             return MessageList
                     .newBuilder()

diff --git a/src/main/java/io/numaproj/numaflow/sourcetransformer/Message.java b/src/main/java/io/numaproj/numaflow/sourcetransformer/Message.java
@@ -11,9 +11,6 @@
 @Getter
 public class Message {
     public static final String DROP = "U+005C__DROP__";
-    // Watermark are at millisecond granularity, hence we use epoch(0) - 1 to indicate watermark is not available.
-    // EventTimeForDrop is used to indicate that the message is dropped hence, excluded from watermark calculation
-    private static final Instant EventTimeForDrop = Instant.ofEpochMilli(-1);
     private final String[] keys;
     private final byte[] value;
     private final Instant eventTime;
@@ -56,14 +53,18 @@ public Message(byte[] value, Instant eventTime, String[] keys) {
     }
 
     /**
-     * creates a Message which will be dropped
+     * creates a Message which will be dropped.
+     *
+     * @param eventTime message eventTime is required because even though a message is dropped,
+     *         we consider it as being processed, hence it should be counted in the watermark calculation
+     *         using the provided event time.
      *
      * @return returns the Message which will be dropped
      */
-    public static Message toDrop() {
+    public static Message toDrop(Instant eventTime) {
         return new Message(
                 new byte[0],
-                EventTimeForDrop,
+                eventTime,
                 null,
                 new String[]{DROP});
     }