Pr 95633 - #1
Open
linxia0415 wants to merge 8 commits into
Open
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
📝 WalkthroughWalkthroughThis PR introduces a new thread-queue-parallel processing mode for the Sentry remote subscriptions consumer. The implementation adds queue-based per-group message handling with safe Kafka offset tracking, integrates it into the existing factory pattern, and includes comprehensive unit and integration tests validating ordering, concurrency, and offset commit behavior. ChangesThread-Queue-Parallel Processing Implementation
Sequence DiagramsequenceDiagram
participant Decoder
participant GroupingFn
participant FixedQueuePool
participant OrderedQueueWorker
participant ResultProcessor
participant OffsetTracker
participant CommitLoop
Decoder->>GroupingFn: decode & group payload
GroupingFn->>FixedQueuePool: submit WorkItem to group queue
FixedQueuePool->>OrderedQueueWorker: enqueue WorkItem
OrderedQueueWorker->>ResultProcessor: process result in transaction
ResultProcessor->>OffsetTracker: mark offset completed
CommitLoop->>OffsetTracker: get committable offsets
CommitLoop->>ResultProcessor: commit per-partition offsets
OffsetTracker->>OffsetTracker: mark committed
🎯 3 (Moderate) | ⏱️ ~25 minutes
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 inconclusive)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (1)
tests/sentry/uptime/consumers/test_results_consumer.py (1)
2142-2146: 💤 Low valueConsider logging the exception in cleanup.
While silently ignoring cleanup failures is often acceptable in tests, logging helps debug CI issues when topic deletion fails.
Suggested improvement
finally: try: admin_client.delete_topics([test_topic]) - except Exception: - pass + except Exception as e: + # Best effort cleanup - log but don't fail the test + import logging + logging.getLogger(__name__).warning(f"Failed to delete test topic {test_topic}: {e}")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/sentry/uptime/consumers/test_results_consumer.py` around lines 2142 - 2146, In the finally block where admin_client.delete_topics([test_topic]) is currently swallowed, catch the exception as e and log it instead of passing; e.g., replace "except Exception: pass" with "except Exception as e: logger.exception('Failed to delete test topic %s', test_topic, exc_info=e)" (or use the test module's existing logger) so failures in admin_client.delete_topics and the test_topic value and traceback are recorded for CI debugging.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@tests/sentry/remote_subscriptions/consumers/test_queue_consumer.py`:
- Around line 232-235: The commit handler commit_function currently sets
commit_event on every commit, causing tests to wake on intermediate commits;
change the handler so it only sets commit_event when the committed_offsets reach
the expected target for that test (e.g., accept an expected_offsets or
expected_partition_offset value supplied by the test, or compare the updated
self.committed_offsets against a provided expected_offsets dict) while still
holding process_lock; update tests to set the expected_offsets before submitting
and wait on commit_event only when the condition (committed_offsets contains/>=
expected_offsets) is met, rather than unconditionally setting the event on any
commit.
- Around line 98-107: The test test_different_groups_distributed is flaky
because asserting exactly three distinct indices from only 20 hashed keys is
probabilistic; replace the strict equality with a deterministic / weaker check:
ensure get_queue_for_group consistently maps the same group_key to the same
queue (call self.pool.get_queue_for_group twice for a sample key) and validate
index bounds (each returned queue_index is within 0..self.pool.num_queues-1),
and then assert that at least two distinct queue indices were observed
(len(queue_indices) >= 2) or increase the corpus size to a large fixed set to
make distribution stable; refer to the test function name and the method
self.pool.get_queue_for_group when making changes.
- Around line 199-203: The immediate assertion on stats["total_items"] is racy
because workers may consume items before the check; replace the direct assert
with a short polling loop that calls self.pool.get_stats() until
stats["total_items"] > 0 or a small timeout elapses (e.g. 1–2s), then assert it;
keep the existing assertion that "queue_depths" exists and
len(stats["queue_depths"]) == 3, and continue to use
self.process_complete_event.wait(timeout=5.0) to ensure processing finishes. Use
the existing symbols self.pool.get_stats, stats, and self.process_complete_event
when implementing the polling to stabilize the test.
---
Nitpick comments:
In `@tests/sentry/uptime/consumers/test_results_consumer.py`:
- Around line 2142-2146: In the finally block where
admin_client.delete_topics([test_topic]) is currently swallowed, catch the
exception as e and log it instead of passing; e.g., replace "except Exception:
pass" with "except Exception as e: logger.exception('Failed to delete test topic
%s', test_topic, exc_info=e)" (or use the test module's existing logger) so
failures in admin_client.delete_topics and the test_topic value and traceback
are recorded for CI debugging.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: bb79b98c-be2a-4748-a870-256c04a70f2c
📒 Files selected for processing (7)
src/sentry/consumers/__init__.pysrc/sentry/remote_subscriptions/consumers/queue_consumer.pysrc/sentry/remote_subscriptions/consumers/result_consumer.pytests/sentry/remote_subscriptions/__init__.pytests/sentry/remote_subscriptions/consumers/__init__.pytests/sentry/remote_subscriptions/consumers/test_queue_consumer.pytests/sentry/uptime/consumers/test_results_consumer.py
Comment on lines
+98
to
+107
| def test_different_groups_distributed(self): | ||
| """Test that different groups are distributed across queues.""" | ||
| queue_indices = set() | ||
| for i in range(20): | ||
| group_key = f"group{i}" | ||
| queue_index = self.pool.get_queue_for_group(group_key) | ||
| queue_indices.add(queue_index) | ||
|
|
||
| assert len(queue_indices) == 3 | ||
|
|
There was a problem hiding this comment.
Distribution assertion is probabilistic and can flake.
assert len(queue_indices) == 3 depends on hash outcomes for just 20 keys; this can occasionally miss one bucket and fail nondeterministically. Prefer asserting deterministic properties (same key maps consistently, index bounds valid) or using a larger fixed corpus and weaker expectation (>= 2).
Suggested stabilization
- assert len(queue_indices) == 3
+ # Avoid probabilistic all-buckets requirement; still verifies spread.
+ assert 1 < len(queue_indices) <= 3📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def test_different_groups_distributed(self): | |
| """Test that different groups are distributed across queues.""" | |
| queue_indices = set() | |
| for i in range(20): | |
| group_key = f"group{i}" | |
| queue_index = self.pool.get_queue_for_group(group_key) | |
| queue_indices.add(queue_index) | |
| assert len(queue_indices) == 3 | |
| def test_different_groups_distributed(self): | |
| """Test that different groups are distributed across queues.""" | |
| queue_indices = set() | |
| for i in range(20): | |
| group_key = f"group{i}" | |
| queue_index = self.pool.get_queue_for_group(group_key) | |
| queue_indices.add(queue_index) | |
| # Avoid probabilistic all-buckets requirement; still verifies spread. | |
| assert 1 < len(queue_indices) <= 3 |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/sentry/remote_subscriptions/consumers/test_queue_consumer.py` around
lines 98 - 107, The test test_different_groups_distributed is flaky because
asserting exactly three distinct indices from only 20 hashed keys is
probabilistic; replace the strict equality with a deterministic / weaker check:
ensure get_queue_for_group consistently maps the same group_key to the same
queue (call self.pool.get_queue_for_group twice for a sample key) and validate
index bounds (each returned queue_index is within 0..self.pool.num_queues-1),
and then assert that at least two distinct queue indices were observed
(len(queue_indices) >= 2) or increase the corpus size to a large fixed set to
make distribution stable; refer to the test function name and the method
self.pool.get_queue_for_group when making changes.
Comment on lines
+199
to
+203
| stats = self.pool.get_stats() | ||
| assert stats["total_items"] > 0 | ||
| assert "queue_depths" in stats | ||
| assert len(stats["queue_depths"]) == 3 | ||
| assert self.process_complete_event.wait(timeout=5.0), "Processing did not complete in time" |
There was a problem hiding this comment.
Immediate queue-depth assertion races worker consumption.
stats["total_items"] > 0 right after enqueue is timing-sensitive; workers may drain queues before this check. This makes CI flaky.
Suggested stabilization
- stats = self.pool.get_stats()
- assert stats["total_items"] > 0
+ # Validate schema first; depth is inherently racey under active workers.
+ stats = self.pool.get_stats()
assert "queue_depths" in stats
assert len(stats["queue_depths"]) == 3
assert self.process_complete_event.wait(timeout=5.0), "Processing did not complete in time"📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| stats = self.pool.get_stats() | |
| assert stats["total_items"] > 0 | |
| assert "queue_depths" in stats | |
| assert len(stats["queue_depths"]) == 3 | |
| assert self.process_complete_event.wait(timeout=5.0), "Processing did not complete in time" | |
| # Validate schema first; depth is inherently racey under active workers. | |
| stats = self.pool.get_stats() | |
| assert "queue_depths" in stats | |
| assert len(stats["queue_depths"]) == 3 | |
| assert self.process_complete_event.wait(timeout=5.0), "Processing did not complete in time" |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/sentry/remote_subscriptions/consumers/test_queue_consumer.py` around
lines 199 - 203, The immediate assertion on stats["total_items"] is racy because
workers may consume items before the check; replace the direct assert with a
short polling loop that calls self.pool.get_stats() until stats["total_items"] >
0 or a small timeout elapses (e.g. 1–2s), then assert it; keep the existing
assertion that "queue_depths" exists and len(stats["queue_depths"]) == 3, and
continue to use self.process_complete_event.wait(timeout=5.0) to ensure
processing finishes. Use the existing symbols self.pool.get_stats, stats, and
self.process_complete_event when implementing the polling to stabilize the test.
Comment on lines
+232
to
+235
| def commit_function(offsets: dict[Partition, int]): | ||
| with self.process_lock: | ||
| self.committed_offsets.update(offsets) | ||
| self.commit_event.set() |
There was a problem hiding this comment.
Commit synchronization tracks “any commit”, not the expected commit.
commit_event is set on every commit, so waits can unblock on an intermediate offset and assertions against final offsets can race. This affects multiple commit-related tests.
Suggested stabilization pattern
- self.commit_event = threading.Event()
+ self.commit_event = threading.Event()
+ self.expected_commit: tuple[Partition, int] | None = None
def commit_function(offsets: dict[Partition, int]):
with self.process_lock:
self.committed_offsets.update(offsets)
- self.commit_event.set()
+ if self.expected_commit is not None:
+ p, o = self.expected_commit
+ if self.committed_offsets.get(p, -1) >= o:
+ self.commit_event.set()And in each test before submit:
+ self.expected_commit = (partition, 104)
self.commit_event.clear()Also applies to: 296-301, 347-350, 363-373
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/sentry/remote_subscriptions/consumers/test_queue_consumer.py` around
lines 232 - 235, The commit handler commit_function currently sets commit_event
on every commit, causing tests to wake on intermediate commits; change the
handler so it only sets commit_event when the committed_offsets reach the
expected target for that test (e.g., accept an expected_offsets or
expected_partition_offset value supplied by the test, or compare the updated
self.committed_offsets against a provided expected_offsets dict) while still
holding process_lock; update tests to set the expected_offsets before submitting
and wait on commit_event only when the condition (committed_offsets contains/>=
expected_offsets) is met, rather than unconditionally setting the event on any
commit.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Legal Boilerplate
Look, I get it. The entity doing business as "Sentry" was incorporated in the State of Delaware in 2015 as Functional Software, Inc. and is gonna need some rights from me in order to utilize my contributions in this here PR. So here's the deal: I retain all rights, title and interest in and to my contributions, and by keeping this boilerplate intact I confirm that Sentry can use, modify, copy, and redistribute my contributions, under Sentry's choice of terms.
Summary by CodeRabbit
Release Notes
New Features
thread-queue-parallelas a new message processing mode option.--modeoption to support the new processing mode.--max-workersparameter description for better clarity.Tests