feat(uptime): Add ability to use queues to manage parallelism#12
feat(uptime): Add ability to use queues to manage parallelism#12ShashankFC 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. -->
|
bugbot run |
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
Bugbot Autofix is OFF. To automatically fix reported issues with Cloud Agents, enable Autofix in the Cursor dashboard.
| def close(self) -> None: | ||
| self.shutdown_event.set() | ||
| self.commit_thread.join(timeout=5.0) | ||
| self.queue_pool.shutdown() |
There was a problem hiding this comment.
Shutdown ordering causes uncommitted offsets for drained items
Medium Severity
In close(), the commit thread is joined before the queue pool is shut down. queue_pool.shutdown() drains remaining items from worker queues — workers process those items and call complete_offset — but the commit thread is already dead at that point, so those newly completed offsets are never committed. On restart, the consumer re-reads and reprocesses those messages. Swapping the order — calling self.queue_pool.shutdown() first, then signaling and joining the commit thread — would allow the final _commit_loop iteration to commit all offsets including those from drained items.
2 participants
Test 9
Summary by CodeRabbit
New Features
Improvements
✏️ Tip: You can customize this high-level summary in your review settings.
Replicated from ai-code-review-evaluation/sentry-coderabbit#9
Note
Medium Risk
Changes message processing and offset-commit behavior for a Kafka consumer via new multi-threaded queues and a background commit loop, which could cause ordering/commit regressions or shutdown edge cases if incorrect.
Overview
Adds a new
thread-queue-parallelexecution mode for uptime/remote subscription result consumers, using a fixed set of per-group FIFO queues to increase parallelism while preserving ordering within a subscription/group.Introduces
queue_consumer.pywith queue workers plus anOffsetTrackerand background commit loop to only commit Kafka offsets once all prior offsets in a partition are completed, and wires this strategy intoResultsStrategyFactory(including lifecycle shutdown handling). CLI uptime consumer options now expose the new mode and clarify--max-workerssemantics; tests are expanded with unit coverage for ordering/offset behavior and an integration test that verifies committed offsets against a real Kafka consumer group.Written by Cursor Bugbot for commit 33f4fdd. Configure here.