Fix #427: preserve original event timestamp in Event.forward

faust/events.py

@@ -28,6 +28,7 @@ class _App: ...  # noqa
 USE_EXISTING_KEY = object()
 USE_EXISTING_VALUE = object()
 USE_EXISTING_HEADERS = object()
+USE_EXISTING_TIMESTAMP = object()
 
 
 class Event(EventT):
@@ -170,7 +171,7 @@ async def forward(
         key: K = USE_EXISTING_KEY,
         value: V = USE_EXISTING_VALUE,
         partition: Optional[int] = None,
-        timestamp: Optional[float] = None,
+        timestamp: Optional[float] = USE_EXISTING_TIMESTAMP,
         headers: Any = USE_EXISTING_HEADERS,
         schema: Optional[SchemaT] = None,
         key_serializer: CodecArg = None,
@@ -183,6 +184,11 @@ async def forward(
             key = self.message.key
         if value is USE_EXISTING_VALUE:
             value = self.message.value
+        if timestamp is USE_EXISTING_TIMESTAMP:
+            # Preserve the original event time so downstream windowing keeps
+            # working after e.g. Stream.group_by (which forwards events).
+            # See issue #427.
+            timestamp = self.message.timestamp
         if headers is USE_EXISTING_HEADERS:
             headers = self.message.headers
             if not headers:

tests/unit/test_events.py

@@ -141,6 +141,16 @@ async def test_forward__USE_EXISTING_KEY_VALUE(self, *, event):
             force=False,
         )
 
+    @pytest.mark.asyncio
+    async def test_forward__default_timestamp_preserved(self, *, event):
+        # With no timestamp argument, forward keeps the original event time
+        # so downstream windowing survives e.g. group_by.  See issue #427.
+        event._send = AsyncMock(name="event._send")
+        event.message.timestamp = 987.6
+        await event.forward(channel="chan")
+        # timestamp is the 5th positional argument to _send.
+        assert event._send.call_args.args[4] == 987.6
+
     def test_attach(self, *, event, app):
         callback = Mock(name="callback")
         schema = Mock(name="schema")