diff --git a/faust/events.py b/faust/events.py index e45dd4fe1..3068c378a 100644 --- a/faust/events.py +++ b/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: diff --git a/tests/unit/test_events.py b/tests/unit/test_events.py index 626329893..fdf94f215 100644 --- a/tests/unit/test_events.py +++ b/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")