feat(uptime): Add ability to use queues to manage parallelism#7
feat(uptime): Add ability to use queues to manage parallelism#7akshayutture-augment wants to merge 1 commit into
Conversation
One potential problem we have with batch processing is that any one slow item will clog up the whole batch. This pr implements a queueing method instead, where we keep N queues that each have their own workers. There's still a chance of individual items backlogging a queue, but we can try increased concurrency here to reduce the chances of that happening <!-- Describe your PR here. -->
![augmentcode[bot]](https://avatars.githubusercontent.com/in/1027498?s=60&v=4){kind=small}
| try: | ||
| work_item = self.work_queue.get() | ||
| except queue.ShutDown: | ||
| break |
There was a problem hiding this comment.
queue.Queue does not define a ShutDown exception; catching queue.ShutDown will never trigger and can mask real shutdown conditions (also applies to the later except queue.ShutDown).
🤖 Was this useful? React with 👍 or 👎
|
|
||
| for q in self.queues: | ||
| try: | ||
| q.shutdown(immediate=False) |
There was a problem hiding this comment.
queue.Queue has no shutdown method; calling q.shutdown(...) will raise and threads blocked on get() won’t be woken, risking a hang during shutdown.
🤖 Was this useful? React with 👍 or 👎
| "Unexpected error in queue worker", extra={"worker_id": self.worker_id} | ||
| ) | ||
| finally: | ||
| self.offset_tracker.complete_offset(work_item.partition, work_item.offset) |
There was a problem hiding this comment.
Offsets are marked complete in the finally block even when processing raises, which violates the stated guarantee of committing only after successful processing and can advance commits past failed items.
🤖 Was this useful? React with 👍 or 👎
2 participants
No description provided.