Rename heartbeat to wall clock

Author: gwaramadze

quixstreams/app.py

@@ -156,7 +156,7 @@ def __init__(
         topic_create_timeout: float = 60,
         processing_guarantee: ProcessingGuarantee = "at-least-once",
         max_partition_buffer_size: int = 10000,
-        heartbeat_interval: float = 0.0,
+        wall_clock_interval: float = 0.0,
     ):
         """
         :param broker_address: Connection settings for Kafka.
@@ -226,11 +226,11 @@ def __init__(
             It is a soft limit, and the actual number of buffered messages can be up to x2 higher.
             Lower value decreases the memory use, but increases the latency.
             Default - `10000`.
-        :param heartbeat_interval: the interval (seconds) at which to send heartbeat messages.
-            The heartbeat timing starts counting from application start.
-            TODO: Save and respect last heartbeat timestamp.
-            The heartbeat is sent for every partition of every topic with registered heartbeat streams.
-            If the value is 0, no heartbeat messages will be sent.
+        :param wall_clock_interval: the interval (seconds) at which to invoke
+            the registered wall clock logic.
+            The wall clock timing starts counting from application start.
+            TODO: Save and respect last wall clock timestamp.
+            If the value is 0, no wall clock logic will be invoked.
             Default - `0.0`.
 
         <br><br>***Error Handlers***<br>
@@ -380,9 +380,9 @@ def __init__(
             recovery_manager=recovery_manager,
         )
 
-        self._heartbeat_active = heartbeat_interval > 0
-        self._heartbeat_interval = heartbeat_interval
-        self._heartbeat_last_sent = datetime.now().timestamp()
+        self._wall_clock_active = wall_clock_interval > 0
+        self._wall_clock_interval = wall_clock_interval
+        self._wall_clock_last_sent = datetime.now().timestamp()
 
         self._source_manager = SourceManager()
         self._sink_manager = SinkManager()
@@ -913,7 +913,7 @@ def _run_dataframe(self, sink: Optional[VoidExecutor] = None):
         processing_context = self._processing_context
         source_manager = self._source_manager
         process_message = self._process_message
-        process_heartbeat = self._process_heartbeat
+        process_wall_clock = self._process_wall_clock
         printer = self._processing_context.printer
         run_tracker = self._run_tracker
         consumer = self._consumer
@@ -926,9 +926,9 @@ def _run_dataframe(self, sink: Optional[VoidExecutor] = None):
         )
 
         dataframes_composed = self._dataframe_registry.compose_all(sink=sink)
-        heartbeats_composed = self._dataframe_registry.compose_heartbeats()
-        if not heartbeats_composed:
-            self._heartbeat_active = False
+        wall_clock_executors = self._dataframe_registry.compose_wall_clock()
+        if not wall_clock_executors:
+            self._wall_clock_active = False
 
         processing_context.init_checkpoint()
         run_tracker.set_as_running()
@@ -940,7 +940,7 @@ def _run_dataframe(self, sink: Optional[VoidExecutor] = None):
                 run_tracker.timeout_refresh()
             else:
                 process_message(dataframes_composed)
-                process_heartbeat(heartbeats_composed)
+                process_wall_clock(wall_clock_executors)
                 processing_context.commit_checkpoint()
                 consumer.resume_backpressured()
                 source_manager.raise_for_error()
@@ -1024,18 +1024,18 @@ def _process_message(self, dataframe_composed):
         if self._on_message_processed is not None:
             self._on_message_processed(topic_name, partition, offset)
 
-    def _process_heartbeat(self, heartbeats_composed):
-        if not self._heartbeat_active:
+    def _process_wall_clock(self, wall_clock_executors):
+        if not self._wall_clock_active:
             return
 
         now = datetime.now().timestamp()
-        if self._heartbeat_last_sent > now - self._heartbeat_interval:
+        if self._wall_clock_last_sent > now - self._wall_clock_interval:
             return
 
         value, key, timestamp, headers = None, None, int(now * 1000), {}
 
         for tp in self._consumer.assignment():
-            if executor := heartbeats_composed.get(tp.topic):
+            if executor := wall_clock_executors.get(tp.topic):
                 row = Row(
                     value=value,
                     key=key,
@@ -1057,7 +1057,7 @@ def _process_heartbeat(self, heartbeats_composed):
                     if not to_suppress:
                         raise
 
-        self._heartbeat_last_sent = now
+        self._wall_clock_last_sent = now
 
     def _on_assign(self, _, topic_partitions: List[TopicPartition]):
         """

quixstreams/core/stream/functions/transform.py

@@ -4,8 +4,8 @@
 from .types import (
     TransformCallback,
     TransformExpandedCallback,
-    TransformHeartbeatCallback,
-    TransformHeartbeatExpandedCallback,
+    TransformWallClockCallback,
+    TransformWallClockExpandedCallback,
     VoidExecutor,
 )
 
@@ -32,61 +32,61 @@ def __init__(
         self,
         func: TransformCallback,
         expand: Literal[False] = False,
-        heartbeat: Literal[False] = False,
+        wall_clock: Literal[False] = False,
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         func: TransformExpandedCallback,
         expand: Literal[True],
-        heartbeat: Literal[False] = False,
+        wall_clock: Literal[False] = False,
     ) -> None: ...
 
     @overload
     def __init__(
         self,
-        func: TransformHeartbeatCallback,
+        func: TransformWallClockCallback,
         expand: Literal[False] = False,
-        heartbeat: Literal[True] = True,
+        wall_clock: Literal[True] = True,
     ) -> None: ...
 
     @overload
     def __init__(
         self,
-        func: TransformHeartbeatExpandedCallback,
+        func: TransformWallClockExpandedCallback,
         expand: Literal[True],
-        heartbeat: Literal[True],
+        wall_clock: Literal[True],
     ) -> None: ...
 
     def __init__(
         self,
         func: Union[
             TransformCallback,
             TransformExpandedCallback,
-            TransformHeartbeatCallback,
-            TransformHeartbeatExpandedCallback,
+            TransformWallClockCallback,
+            TransformWallClockExpandedCallback,
         ],
         expand: bool = False,
-        heartbeat: bool = False,
+        wall_clock: bool = False,
     ):
         super().__init__(func)
 
         self.func: Union[
             TransformCallback,
             TransformExpandedCallback,
-            TransformHeartbeatCallback,
-            TransformHeartbeatExpandedCallback,
+            TransformWallClockCallback,
+            TransformWallClockExpandedCallback,
         ]
         self.expand = expand
-        self.heartbeat = heartbeat
+        self.wall_clock = wall_clock
 
     def get_executor(self, *child_executors: VoidExecutor) -> VoidExecutor:
         child_executor = self._resolve_branching(*child_executors)
 
-        if self.expand and self.heartbeat:
-            heartbeat_expanded_func = cast(
-                TransformHeartbeatExpandedCallback, self.func
+        if self.expand and self.wall_clock:
+            wall_clock_expanded_func = cast(
+                TransformWallClockExpandedCallback, self.func
             )
 
             def wrapper(
@@ -100,7 +100,7 @@ def wrapper(
                     new_key,
                     new_timestamp,
                     new_headers,
-                ) in heartbeat_expanded_func(timestamp):
+                ) in wall_clock_expanded_func(timestamp):
                     child_executor(new_value, new_key, new_timestamp, new_headers)
 
         elif self.expand:
@@ -116,16 +116,16 @@ def wrapper(
                 for new_value, new_key, new_timestamp, new_headers in result:
                     child_executor(new_value, new_key, new_timestamp, new_headers)
 
-        elif self.heartbeat:
-            heartbeat_func = cast(TransformHeartbeatCallback, self.func)
+        elif self.wall_clock:
+            wall_clock_func = cast(TransformWallClockCallback, self.func)
 
             def wrapper(
                 value: Any,
                 key: Any,
                 timestamp: int,
                 headers: Any,
             ):
-                new_value, new_key, new_timestamp, new_headers = heartbeat_func(
+                new_value, new_key, new_timestamp, new_headers = wall_clock_func(
                     timestamp
                 )
                 child_executor(new_value, new_key, new_timestamp, new_headers)

quixstreams/core/stream/functions/types.py

@@ -14,8 +14,8 @@
     "FilterWithMetadataCallback",
     "TransformCallback",
     "TransformExpandedCallback",
-    "TransformHeartbeatCallback",
-    "TransformHeartbeatExpandedCallback",
+    "TransformWallClockCallback",
+    "TransformWallClockExpandedCallback",
 )
 
 
@@ -37,8 +37,8 @@ def __bool__(self) -> bool: ...
 TransformExpandedCallback = Callable[
     [Any, Any, int, Any], Iterable[Tuple[Any, Any, int, Any]]
 ]
-TransformHeartbeatCallback = Callable[[int], Tuple[Any, Any, int, Any]]
-TransformHeartbeatExpandedCallback = Callable[
+TransformWallClockCallback = Callable[[int], Tuple[Any, Any, int, Any]]
+TransformWallClockExpandedCallback = Callable[
     [int], Iterable[Tuple[Any, Any, int, Any]]
 ]
 

quixstreams/core/stream/stream.py

@@ -30,8 +30,8 @@
     TransformCallback,
     TransformExpandedCallback,
     TransformFunction,
-    TransformHeartbeatCallback,
-    TransformHeartbeatExpandedCallback,
+    TransformWallClockCallback,
+    TransformWallClockExpandedCallback,
     UpdateCallback,
     UpdateFunction,
     UpdateWithMetadataCallback,
@@ -256,7 +256,7 @@ def add_transform(
         func: TransformCallback,
         *,
         expand: Literal[False] = False,
-        heartbeat: Literal[False] = False,
+        wall_clock: Literal[False] = False,
     ):
         pass
 
@@ -266,27 +266,27 @@ def add_transform(
         func: TransformExpandedCallback,
         *,
         expand: Literal[True],
-        heartbeat: Literal[False] = False,
+        wall_clock: Literal[False] = False,
     ):
         pass
 
     @overload
     def add_transform(
         self,
-        func: TransformHeartbeatCallback,
+        func: TransformWallClockCallback,
         *,
         expand: Literal[False] = False,
-        heartbeat: Literal[True],
+        wall_clock: Literal[True],
     ):
         pass
 
     @overload
     def add_transform(
         self,
-        func: TransformHeartbeatExpandedCallback,
+        func: TransformWallClockExpandedCallback,
         *,
         expand: Literal[True],
-        heartbeat: Literal[True],
+        wall_clock: Literal[True],
     ):
         pass
 
@@ -295,12 +295,12 @@ def add_transform(
         func: Union[
             TransformCallback,
             TransformExpandedCallback,
-            TransformHeartbeatCallback,
-            TransformHeartbeatExpandedCallback,
+            TransformWallClockCallback,
+            TransformWallClockExpandedCallback,
         ],
         *,
         expand: bool = False,
-        heartbeat: bool = False,
+        wall_clock: bool = False,
     ) -> "Stream":
         """
         Add a "transform" function to the Stream, that will mutate the input value.
@@ -316,11 +316,11 @@ def add_transform(
         :param expand: if True, expand the returned iterable into individual items
             downstream. If returned value is not iterable, `TypeError` will be raised.
             Default - `False`.
-        :param heartbeat: if True, the callback is expected to accept timestamp only.
+        :param wall_clock: if True, the callback is expected to accept timestamp only.
             Default - `False`.
         :return: a new Stream derived from the current one
         """
-        return self._add(TransformFunction(func, expand=expand, heartbeat=heartbeat))  # type: ignore[call-overload]
+        return self._add(TransformFunction(func, expand=expand, wall_clock=wall_clock))  # type: ignore[call-overload]
 
     def merge(self, other: "Stream") -> "Stream":
         """

quixstreams/dataframe/dataframe.py

@@ -1696,8 +1696,8 @@ def concat(self, other: "StreamingDataFrame") -> "StreamingDataFrame":
                 stream=merged_stream, stream_id=merged_stream_id
             )
 
-    def concat_heartbeat(self, stream: Stream) -> "StreamingDataFrame":
-        self._registry.register_heartbeat(self, stream)
+    def concat_wall_clock(self, stream: Stream) -> "StreamingDataFrame":
+        self._registry.register_wall_clock(self, stream)
         return self.__dataframe_clone__(stream=self.stream.merge(stream))
 
     def join_asof(

quixstreams/dataframe/registry.py

@@ -22,7 +22,7 @@ class DataFrameRegistry:
 
     def __init__(self) -> None:
         self._registry: dict[str, Stream] = {}
-        self._heartbeat_registry: dict[str, Stream] = {}
+        self._wall_clock_registry: dict[str, Stream] = {}
         self._topics: list[Topic] = []
         self._repartition_origins: set[str] = set()
         self._topics_to_stream_ids: dict[str, set[str]] = {}
@@ -70,19 +70,19 @@ def register_root(
         self._topics.append(topic)
         self._registry[topic.name] = dataframe.stream
 
-    def register_heartbeat(
+    def register_wall_clock(
         self, dataframe: "StreamingDataFrame", stream: Stream
     ) -> None:
         """
-        Register a heartbeat Stream for the given topic.
+        Register a wall clock stream for the given topic.
         """
         topics = dataframe.topics
         if len(topics) > 1:
             raise ValueError(
                 f"Expected a StreamingDataFrame with one topic, got {len(topics)}"
             )
         topic = topics[0]
-        self._heartbeat_registry[topic.name] = stream
+        self._wall_clock_registry[topic.name] = stream
 
     def register_groupby(
         self,
@@ -130,12 +130,12 @@ def compose_all(
         """
         return self._compose(registry=self._registry, sink=sink)
 
-    def compose_heartbeats(self) -> dict[str, VoidExecutor]:
+    def compose_wall_clock(self) -> dict[str, VoidExecutor]:
         """
-        Composes all the heartbeat Streams and returns a dict of format {<topic>: <VoidExecutor>}
+        Composes all the wall clock streams and returns a dict of format {<topic>: <VoidExecutor>}
         :return: a {topic_name: composed} dict, where composed is a callable
         """
-        return self._compose(registry=self._heartbeat_registry)
+        return self._compose(registry=self._wall_clock_registry)
 
     def _compose(
         self, registry: dict[str, Stream], sink: Optional[VoidExecutor] = None