Skip to content

Commit

Permalink
Merge pull request #2 from ourzora/catch-exceptions
Browse files Browse the repository at this point in the history
Catch exceptions
  • Loading branch information
owen-zora authored May 25, 2023
2 parents 2c4fd18 + 5a52d78 commit 76a97b6
Show file tree
Hide file tree
Showing 2 changed files with 18 additions and 9 deletions.
7 changes: 6 additions & 1 deletion faust/streams.py
Original file line number Diff line number Diff line change
Expand Up @@ -1202,6 +1202,7 @@ async def _py_aiter(self) -> AsyncIterator[T_co]:
sensor_state = None

# reduce using processors
processor = None
try:
for processor in processors:
with trace(f"processor-{_shortlabel(processor)}"):
Expand All @@ -1213,8 +1214,12 @@ async def _py_aiter(self) -> AsyncIterator[T_co]:
value = skipped_value
except StopAsyncIteration:
raise
except Exception:
except Exception as exc:
value = skipped_value
self.log.exception(
f"Error in processor {_shortlabel(processor)}: {exc}"
)

try:
if value is not skipped_value:
self.events_total += 1
Expand Down
20 changes: 12 additions & 8 deletions faust/transport/consumer.py
Original file line number Diff line number Diff line change
Expand Up @@ -749,10 +749,17 @@ async def getmany(self, timeout: float) -> AsyncIterator[Tuple[TP, Message]]:
or tp in active_partitions
or tp in self._buffered_partitions
):
highwater_mark = self.highwater(tp)
self.app.monitor.track_tp_end_offset(tp, highwater_mark)
# convert timestamp to seconds from int milliseconds.
yield tp, to_message(tp, record)
try:
highwater_mark = self.highwater(tp)
self.app.monitor.track_tp_end_offset(tp, highwater_mark)
# convert timestamp to seconds from int milliseconds.
yield tp, to_message(tp, record)
except StopAsyncIteration:
raise
except Exception as exc:
self.log.exception(
"Error while processing message", exc_info=exc
)
else:
self.log.dev(
"getmany called while flow not active. Seek back to committed offsets."
Expand Down Expand Up @@ -1118,10 +1125,7 @@ def _new_offset(self, tp: TP) -> Optional[int]:
# the return value will be None (the same as 31)
if self._committed_offset[tp]:
if min(acked) - self._committed_offset[tp] > 1:
new_acked = list(range(self._committed_offset[tp] + 1, min(acked)))
self.log.dev(f"insert {new_acked=}")
# return None
acked = new_acked + acked
return None

# Note: acked is always kept sorted.
# find first list of consecutive numbers
Expand Down

0 comments on commit 76a97b6

Please sign in to comment.