fix: update event time filter transformer to use tag (numaproj#43)

Signed-off-by: Keran Yang <[email protected]>
KeranYang · Apr 26, 2023 · cfbe68e · cfbe68e
1 parent 792d481
commit cfbe68e
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 2 deletions.
diff --git a/...a/io/numaproj/numaflow/examples/function/map/eventtimefilter/EventTimeFilterFunction.java b/...a/io/numaproj/numaflow/examples/function/map/eventtimefilter/EventTimeFilterFunction.java
@@ -13,8 +13,8 @@
  * 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 key to "within_year_2022" and update the message event time to Jan 1st 2022.
- * Otherwise, (exclusively after year 2022), update the key to "after_year_2022" and update the
+ * 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.
  */
 public class EventTimeFilterFunction extends MapTHandler {
@@ -40,6 +40,7 @@ public MessageTList processMessage(String[] keys, Datum data) {
                             new MessageT(
                                     data.getValue(),
                                     januaryFirst2022,
+                                    null,
                                     new String[]{"within_year_2022"}))
                     .build();
         } else {
@@ -48,6 +49,7 @@ public MessageTList processMessage(String[] keys, Datum data) {
                     .addMessage(new MessageT(
                             data.getValue(),
                             januaryFirst2023,
+                            null,
                             new String[]{"after_year_2022"}))
                     .build();
         }

diff --git a/src/main/java/io/numaproj/numaflow/function/FunctionServer.java b/src/main/java/io/numaproj/numaflow/function/FunctionServer.java
@@ -102,6 +102,7 @@ public void start() throws Exception {
                 Language.JAVA,
                 serverInfoAccessor.getSDKVersion(),
                 new HashMap<>());
+        log.info("Writing server info {} to %s", serverInfo, infoFilePath);
         serverInfoAccessor.write(serverInfo, infoFilePath);
 
         // build server

diff --git a/src/main/java/io/numaproj/numaflow/sink/SinkServer.java b/src/main/java/io/numaproj/numaflow/sink/SinkServer.java
@@ -82,6 +82,7 @@ public void start() throws Exception {
                 Language.JAVA,
                 serverInfoAccessor.getSDKVersion(),
                 new HashMap<>());
+        log.info("Writing server info {} to %s", serverInfo, infoFilePath);
         serverInfoAccessor.write(serverInfo, infoFilePath);
 
         // build server