chore: Update Numaflow version rollouts (#138)

Signed-off-by: Sidhant Kohli <[email protected]>
Co-authored-by: Sidhant Kohli <[email protected]>

Author: kohlisid

numaprom/tools.py

@@ -41,7 +41,7 @@ def inner_function(*args, **kwargs):
         msgs = Messages()
         for json_data in json_list:
             if json_data:
-                msgs.append(Message.to_all(json_data))
+                msgs.append(Message(json_data))
             else:
                 msgs.append(Message.to_drop())
         return msgs
@@ -55,7 +55,7 @@ def inner_function(*args, **kwargs):
         json_data = handler_func(*args, **kwargs)
         msgs = Messages()
         if json_data:
-            msgs.append(Message.to_all(value=json_data))
+            msgs.append(Message(value=json_data))
         else:
             msgs.append(Message.to_drop())
         return msgs
@@ -70,7 +70,7 @@ def inner_function(*args, **kwargs) -> Messages:
         msgs = Messages()
         for vertex, json_data in data:
             if json_data and vertex:
-                msgs.append(Message.to_vtx(key=vertex.encode(), value=json_data))
+                msgs.append(Message(value=json_data, tags=[vertex.encode()]))
             else:
                 msgs.append(Message.to_drop())
         return msgs

numaprom/udf/filter.py

@@ -12,7 +12,7 @@
 
 @catch_exception
 @msg_forward
-def metric_filter(_: str, datum: Datum) -> Optional[Messages]:
+def metric_filter(_: list[str], datum: Datum) -> Optional[Messages]:
     """
     UDF to filter metrics by labels
     """

numaprom/udf/inference.py

@@ -1,5 +1,4 @@
 import time
-
 from numalogic.config import NumalogicConf
 from numalogic.models.autoencoder import AutoencoderTrainer
 from numalogic.registry import ArtifactData, RedisRegistry
@@ -43,7 +42,7 @@ def _run_inference(
 
 
 @msg_forward
-def inference(_: str, datum: Datum) -> bytes:
+def inference(_: list[str], datum: Datum) -> bytes:
     _start_time = time.perf_counter()
 
     _in_msg = datum.value.decode("utf-8")

numaprom/udf/postprocess.py

@@ -156,7 +156,7 @@ def _publish(final_score: float, payload: StreamPayload) -> List[bytes]:
 
 
 @msgs_forward
-def postprocess(_: str, datum: Datum) -> List[bytes]:
+def postprocess(_: List[str], datum: Datum) -> List[bytes]:
     """
     UDF for performing the following steps:
 

numaprom/udf/preprocess.py

@@ -1,5 +1,6 @@
 import os
 import time
+from typing import List
 
 import orjson
 from numalogic.registry import RedisRegistry
@@ -26,7 +27,7 @@
 
 
 @msg_forward
-def preprocess(_: str, datum: Datum) -> bytes:
+def preprocess(_: List[str], datum: Datum) -> bytes:
     _start_time = time.perf_counter()
     _in_msg = datum.value.decode("utf-8")
 

numaprom/udf/threshold.py

@@ -1,5 +1,6 @@
 import time
 from collections import OrderedDict
+from typing import List
 
 from numalogic.registry import RedisRegistry
 from numalogic.tools.exceptions import RedisRegistryError
@@ -32,7 +33,7 @@ def _get_static_thresh_payload(payload, metric_config) -> bytes:
 
 
 @conditional_forward
-def threshold(_: str, datum: Datum) -> list[tuple[str, bytes]]:
+def threshold(_: List[str], datum: Datum) -> list[tuple[str, bytes]]:
     _start_time = time.perf_counter()
     _in_msg = datum.value.decode("utf-8")
 

numaprom/udf/window.py

@@ -1,7 +1,7 @@
 import os
 import time
 import uuid
-from typing import Optional
+from typing import Optional, List
 
 import numpy as np
 import numpy.typing as npt
@@ -56,7 +56,7 @@ def __aggregate_window(
 
 
 @msg_forward
-def window(_: str, datum: Datum) -> Optional[bytes]:
+def window(_: List[str], datum: Datum) -> Optional[bytes]:
     """
     UDF to construct windowing of the streaming input data, required by ML models.
     """