fix(eventstream): Use an arroyo producer for eventstream #104763
Conversation
Migrate to the arroyo producer (to capture accurate metrics) and also increase the buffer size of the producer.
github-actions
bot
added
the
Scope: Backend
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
![sentry[bot]](https://avatars.githubusercontent.com/u/1396951?s=60&v=4){kind=small}
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #104763 +/- ##
===========================================
- Coverage 80.61% 80.54% -0.07%
===========================================
Files 9419 9385 -34
Lines 403881 403624 -257
Branches 25637 25781 +144
===========================================
- Hits 325572 325099 -473
- Misses 77840 78090 +250
+ Partials 469 435 -34 |
| ) | ||
| ) | ||
| except Exception as error: | ||
| logger.exception("Could not publish message: %s", error) |
There was a problem hiding this comment.
Wrt the if not asynchronous handling, I don't see a flush() method on the KafkaProducer. I think we want to do produce_future.result() to block until delivery completes instead.
| return self.__producers[topic] | ||
|
|
||
| def delivery_callback(self, error: KafkaError | None, message: KafkaMessage) -> None: | ||
| def delivery_callback(self, error: KafkaError | None, value: bytes) -> None: |
There was a problem hiding this comment.
Same as Sentry bot - should we handle when value is None?
And I think we can update the error type to BaseException.
|
This pull request has gone three weeks without activity. In another week, I will close it. But! If you comment or otherwise update it, I will reset the clock, and if you add the label "A weed is but an unloved flower." ― Ella Wheeler Wilcox 🥀 |
Migrate to the arroyo producer (to capture accurate metrics) and also increase the buffer size of
the producer.