Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Discard message after rebalance #223

Merged
merged 7 commits into from
Nov 29, 2021
Merged

Discard message after rebalance #223

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
8a95d4c
Fix Recovery._resume_streams
Nov 11, 2021
f6d6df7
Merge branch 'master' of https://github.com/faust-streaming/faust int…
Nov 18, 2021
99450cf
Discard fetched messages on rebalance
Nov 18, 2021
5da8dca
update cython
Nov 18, 2021
07448ad
Add logs for skipped events
Nov 19, 2021
576b2fc
QA fixes
Nov 19, 2021
46cc9c7
Track generation_id in more places
Nov 19, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Add logs for skipped events
  • Loading branch information
Eric Kerstens committed Nov 19, 2021
commit 07448ad55986b60a048b3fa83474f8d281181efb
6 changes: 6 additions & 0 deletions faust/_cython/streams.pyx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,12 @@ cdef class StreamIterator:
consumer = self.consumer

if message.generation_id != self.app.consumer_generation_id:
self.app.log.dev(
"Skipping message %r with generation_id %r because app generation_id is %r",
message,
message.generation_id,
self.app.consumer_generation_id,
)
return None, self._skipped_value, stream_state
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we log this?


if topic in self.acking_topics and not message.tracked:
Expand Down
6 changes: 6 additions & 0 deletions faust/streams.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -976,6 +976,12 @@ async def _py_aiter(self) -> AsyncIterator[T_co]:

if message.generation_id != self.app.consumer_generation_id:
value = skipped_value
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we log this? It will be easier to debug if this were to happen

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good idea

self.log.dev(
"Skipping message %r with generation_id %r because app generation_id is %r",
message,
message.generation_id,
self.app.consumer_generation_id,
)
break
if topic in acking_topics and not message.tracked:
message.tracked = True
Expand Down