Skip to content

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

Open
wbarnha wants to merge 1 commit into
masterfrom
claude/fix-427-forward-timestamp
Open

Fix #427: preserve original event timestamp in Event.forward#705
wbarnha wants to merge 1 commit into
masterfrom
claude/fix-427-forward-timestamp

Conversation

@wbarnha

@wbarnha wbarnha commented Jul 19, 2026

Copy link
Copy Markdown
Member

What

Event.forward re-sends the original message ("will not be reserialized"), and already defaults key, value and headers to the original message via USE_EXISTING_* sentinels. But timestamp defaulted to None, so the forwarded message received a fresh send-time timestamp instead of the original event time.

Stream.group_by forwards events internally, so grouping silently dropped the event timestamp — breaking downstream event-time windowing (the user has to rebuild the partitioning key by hand to keep the timestamp).

Closes #427.

How

  • Added a USE_EXISTING_TIMESTAMP sentinel alongside the existing USE_EXISTING_KEY / USE_EXISTING_VALUE / USE_EXISTING_HEADERS.
  • Event.forward now defaults timestamp to self.message.timestamp, consistent with how it already treats the other forwarded fields.

message.timestamp is stored in seconds (the consumer divides the Kafka record's millisecond timestamp by 1000), which is the same unit send() expects (it multiplies back by 1000), so the value round-trips correctly. Passing an explicit timestamp — including None — still overrides, so callers keep full control.

Test

Added test_forward__default_timestamp_preserved in tests/unit/test_events.py: with no timestamp argument, asserts forward passes self.message.timestamp through to _send. The existing test_forward (explicit timestamp) and test_forward__USE_EXISTING_KEY_VALUE (explicit timestamp=None) still pass unchanged, confirming explicit values are respected. Full tests/unit/test_events.py passes (9 passed); events + group_by/forward stream tests green (17 passed).

🤖 Generated with Claude Code


Generated by Claude Code

Event.forward already defaults key, value and headers to the original
message via USE_EXISTING_* sentinels, but timestamp defaulted to None, so
the forwarded message got a fresh (send-time) timestamp.  Stream.group_by
forwards events internally, so grouping silently dropped the event time and
broke downstream event-time windowing.

Add a USE_EXISTING_TIMESTAMP sentinel and default forward's timestamp to
self.message.timestamp, consistent with the other forwarded fields.
message.timestamp is in seconds, the same unit send() expects, so it
round-trips correctly.  Passing an explicit timestamp (including None) still
overrides.

Closes #427.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
@codecov

codecov Bot commented Jul 19, 2026

Copy link
Copy Markdown

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 94.15%. Comparing base (3073eb9) to head (a557385).

Additional details and impacted files
@@            Coverage Diff             @@
##           master     #705      +/-   ##
==========================================
+ Coverage   94.14%   94.15%   +0.01%     
==========================================
  Files         104      104              
  Lines       11136    11139       +3     
  Branches     1201     1202       +1     
==========================================
+ Hits        10484    10488       +4     
+ Misses        551      550       -1     
  Partials      101      101              

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

wbarnha added a commit that referenced this pull request Jul 19, 2026
Per review, the v0.12.0 changelog/release notes should describe only what is
already on master, not work still in open PRs.

- Remove the not-yet-merged items: the offset-commit data-loss fixes
  (#606/#707, #316/#692), the optional OpenTracing/OpenTelemetry extras
  (#685/#686, #688/#681), web_application_options (#704), and the reported-issue
  fix stack (#693-#703, #705). These will be added back as they merge.
- Add a Dependencies section noting the current runtime/client libraries:
  mode-streaming >= 0.4.0, aiokafka >= 0.10.0 (compatible with recent 0.13/0.14
  releases), the new confluent-kafka >= 2.0.0 for faust[ckafka], and the
  faust-cchardet fork replacing unmaintained cchardet.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
wbarnha added a commit to SpencerWhitehead7/faust that referenced this pull request Jul 21, 2026
…st-streaming#708)

* docs: prepare v0.12.0 release notes and changelog

Resume the Keep a Changelog format (dormant since v0.8.10) with a v0.12.0
section, and add standalone GitHub release notes covering the changes since
v0.11.3 plus the pending fix stack.

Highlights: two offset data-loss fixes (faust-streaming#606/faust-streaming#707, faust-streaming#316/faust-streaming#692), the
re-added confluent-kafka driver, Python 3.14 support (3.8/3.9 dropped),
OpenTracing/OpenTelemetry made optional, and a live-broker CI harness.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL

* docs: drop closed codecov.yml PR (faust-streaming#683) from v0.12.0 notes

PR faust-streaming#683 (codecov.yml with a 1% coverage threshold) was closed without
merging, so remove it from the changelog and release notes to keep the
v0.12.0 change list accurate.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL

* docs: derive Sphinx version from the package instead of a stale constant

`docs/conf.py` hardcoded `version_dev='1.1'` / `version_stable='1.0'` -
robinhood-era values that never matched faust-streaming's 0.x line, so the
published GitHub Pages docs advertised the wrong version.

Derive the documented major.minor from `faust.__version__` (which
setuptools_scm resolves from the git tag), so the docs always report the real
version and this can't silently drift between releases.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL

* docs: scope v0.12.0 notes to merged work; note dependency updates

Per review, the v0.12.0 changelog/release notes should describe only what is
already on master, not work still in open PRs.

- Remove the not-yet-merged items: the offset-commit data-loss fixes
  (faust-streaming#606/faust-streaming#707, faust-streaming#316/faust-streaming#692), the optional OpenTracing/OpenTelemetry extras
  (faust-streaming#685/faust-streaming#686, faust-streaming#688/faust-streaming#681), web_application_options (faust-streaming#704), and the reported-issue
  fix stack (faust-streaming#693-faust-streaming#703, faust-streaming#705). These will be added back as they merge.
- Add a Dependencies section noting the current runtime/client libraries:
  mode-streaming >= 0.4.0, aiokafka >= 0.10.0 (compatible with recent 0.13/0.14
  releases), the new confluent-kafka >= 2.0.0 for faust[ckafka], and the
  faust-cchardet fork replacing unmaintained cchardet.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL

* docs: set v0.12.0 changelog date to 2026-07-19

Replace the UNRELEASED placeholder with the release date and drop the
now-satisfied "set the date at tag time" note.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL

* Delete RELEASE_NOTES_v0.12.0.md

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Why doesn't Event.forward default to the timestamp of the event?

1 participant