From 45576318c666a0b88fa7b0cbf2cddbaf117390c9 Mon Sep 17 00:00:00 2001 From: Markus Unterwaditzer Date: Wed, 5 Aug 2026 14:21:15 +0200 Subject: [PATCH 1/2] ref(taskworker): Mark raw-mode namespaces explicitly Set is_raw_mode on the namespaces backing topics that taskbroker consumes in raw mode, and drop the task options that taskbroker ignores for them. Raw-mode activations are built by taskbroker from the topic's `raw:` block in its own config, not by a producer, so processing_deadline_duration and compression_type on the decorator never reach the broker. They already disagree with what is deployed: process_profile_from_kafka_raw declares 80s while every region runs the topic at 60s, and 60 is what applies. Removing them is a no-op at runtime and makes the deployed config the only place the values live. taskbroker-client now rejects them at import for raw namespaces, so they cannot drift back in. ingest.profiling.passthrough is left alone; infra has already migrated off it and STREAM-1191 removes it. Blocked on a taskbroker-client release containing is_raw_mode. ref STREAM-1044 Co-Authored-By: Claude Opus 5 (1M context) --- src/sentry/ingest/consumer/simple_event.py | 3 --- src/sentry/profiles/task.py | 2 -- src/sentry/replays/tasks.py | 3 --- src/sentry/snuba/query_subscriptions/run.py | 1 - src/sentry/taskworker/namespaces.py | 8 ++++++++ 5 files changed, 8 insertions(+), 9 deletions(-) diff --git a/src/sentry/ingest/consumer/simple_event.py b/src/sentry/ingest/consumer/simple_event.py index d7f876d36777..fa224990a1eb 100644 --- a/src/sentry/ingest/consumer/simple_event.py +++ b/src/sentry/ingest/consumer/simple_event.py @@ -4,7 +4,6 @@ from arroyo.backends.kafka.consumer import KafkaPayload from arroyo.dlq import InvalidMessage from arroyo.types import BrokerValue, Message -from taskbroker_client.constants import CompressionType from taskbroker_client.retry import Retry from sentry import options @@ -87,9 +86,7 @@ def process_simple_event_message( @instrumented_task( name="sentry.ingest.consumer.simple_event.process_event_from_kafka", namespace=ingest_events_raw_tasks, - processing_deadline_duration=150, retry=Retry(times=2, delay=5, on=(Retriable,)), - compression_type=CompressionType.ZSTD, silo_mode=SiloMode.CELL, ) def process_event_from_kafka(message_bytes: bytes) -> None: diff --git a/src/sentry/profiles/task.py b/src/sentry/profiles/task.py index fde07d6acd4b..ee91dda5dfb8 100644 --- a/src/sentry/profiles/task.py +++ b/src/sentry/profiles/task.py @@ -190,9 +190,7 @@ def process_profile_from_kafka( @instrumented_task( name="sentry.profiles.task.process_profile_from_kafka_raw", namespace=ingest_profiling_raw_tasks, - processing_deadline_duration=80, retry=Retry(times=2, delay=5), - compression_type=CompressionType.ZSTD, silo_mode=SiloMode.CELL, pass_headers=True, ) diff --git a/src/sentry/replays/tasks.py b/src/sentry/replays/tasks.py index f36958b3c2ef..e7a2444aae3f 100644 --- a/src/sentry/replays/tasks.py +++ b/src/sentry/replays/tasks.py @@ -4,7 +4,6 @@ from typing import Any from django.utils import timezone -from taskbroker_client.constants import CompressionType from taskbroker_client.retry import Retry from taskbroker_client.state import current_task from taskbroker_client.worker.workerchild import ProcessingDeadlineExceeded @@ -67,9 +66,7 @@ def delete_replay( @instrumented_task( name=PROCESS_REPLAY_RECORDING_TASK_NAME, namespace=replays_raw_tasks, - processing_deadline_duration=90, retry=Retry(times=3, delay=5), - compression_type=CompressionType.ZSTD, silo_mode=SiloMode.CELL, ) def process_replay_recording(message_bytes: bytes) -> None: diff --git a/src/sentry/snuba/query_subscriptions/run.py b/src/sentry/snuba/query_subscriptions/run.py index 9ac386567152..1ec476d16c5f 100644 --- a/src/sentry/snuba/query_subscriptions/run.py +++ b/src/sentry/snuba/query_subscriptions/run.py @@ -78,7 +78,6 @@ def _register_subscription_tasks() -> None: @instrumented_task( name=f"sentry.snuba.query_subscriptions.run.process_{name}_subscription_from_kafka", namespace=namespace, - processing_deadline_duration=60, silo_mode=SiloMode.CELL, ) def task_fn(message_bytes: bytes, _d: Dataset = dataset) -> None: diff --git a/src/sentry/taskworker/namespaces.py b/src/sentry/taskworker/namespaces.py index a469afbb28d2..7e82ff38b1f2 100644 --- a/src/sentry/taskworker/namespaces.py +++ b/src/sentry/taskworker/namespaces.py @@ -97,6 +97,7 @@ ingest_profiling_raw_tasks = app.taskregistry.create_namespace( "ingest.profiling.raw", app_feature="profiles", + is_raw_mode=True, ) ingest_transactions_tasks = app.taskregistry.create_namespace( @@ -117,6 +118,7 @@ ingest_events_raw_tasks = app.taskregistry.create_namespace( "ingest.events.raw", app_feature="errors", + is_raw_mode=True, ) ingest_errors_tasks = app.taskregistry.create_namespace( @@ -132,26 +134,31 @@ snuba_events_subscriptions_raw_tasks = app.taskregistry.create_namespace( "snuba.subscriptions.events.raw", app_feature="errors", + is_raw_mode=True, ) snuba_transactions_subscriptions_raw_tasks = app.taskregistry.create_namespace( "snuba.subscriptions.transactions.raw", app_feature="transactions", + is_raw_mode=True, ) snuba_metrics_subscriptions_raw_tasks = app.taskregistry.create_namespace( "snuba.subscriptions.metrics.raw", app_feature="sessions", + is_raw_mode=True, ) snuba_generic_metrics_subscriptions_raw_tasks = app.taskregistry.create_namespace( "snuba.subscriptions.generic_metrics.raw", app_feature="transactions", + is_raw_mode=True, ) snuba_eap_subscriptions_raw_tasks = app.taskregistry.create_namespace( "snuba.subscriptions.eap.raw", app_feature="transactions", + is_raw_mode=True, ) issues_tasks = app.taskregistry.create_namespace( @@ -249,6 +256,7 @@ replays_raw_tasks = app.taskregistry.create_namespace( "replays.raw", app_feature="replays", + is_raw_mode=True, ) reports_tasks = app.taskregistry.create_namespace( From c160b823ccdcbbe53a5d3e5f6204057d55132f7f Mon Sep 17 00:00:00 2001 From: Markus Unterwaditzer Date: Thu, 6 Aug 2026 16:49:17 +0200 Subject: [PATCH 2/2] build: Bump taskbroker-client to 0.20.19 for is_raw_mode --- pyproject.toml | 2 +- uv.lock | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index c9336c583382..c614dbc141fc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -107,7 +107,7 @@ dependencies = [ "statsd>=3.3.0", "structlog>=22.1.0", "symbolic>=13.1.1", - "taskbroker-client>=0.20.18", + "taskbroker-client>=0.20.19", "tiktoken>=0.8.0", "tokenizers>=0.22.0", "tldextract>=5.1.2", diff --git a/uv.lock b/uv.lock index ce82aa48b855..c8cd3f7a25f8 100644 --- a/uv.lock +++ b/uv.lock @@ -2426,7 +2426,7 @@ requires-dist = [ { name = "stripe", specifier = ">=6.7.0" }, { name = "structlog", specifier = ">=22.1.0" }, { name = "symbolic", specifier = ">=13.1.1" }, - { name = "taskbroker-client", specifier = ">=0.20.18" }, + { name = "taskbroker-client", specifier = ">=0.20.19" }, { name = "tiktoken", specifier = ">=0.8.0" }, { name = "tldextract", specifier = ">=5.1.2" }, { name = "tokenizers", specifier = ">=0.22.0" }, @@ -2819,7 +2819,7 @@ wheels = [ [[package]] name = "taskbroker-client" -version = "0.20.18" +version = "0.20.19" source = { registry = "https://pypi.devinfra.sentry.io/simple" } dependencies = [ { name = "confluent-kafka", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, @@ -2836,7 +2836,7 @@ dependencies = [ { name = "zstandard", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, ] wheels = [ - { url = "https://pypi.devinfra.sentry.io/wheels/taskbroker_client-0.20.18-py3-none-any.whl", hash = "sha256:6cb7d58b955928d6f2e40a57981077bdc2265648559428cfa150a68c66a3dcb8" }, + { url = "https://pypi.devinfra.sentry.io/wheels/taskbroker_client-0.20.19-py3-none-any.whl", hash = "sha256:509d1f88bbccbdab6f0e7173aa82f38532da9dffa90aa56587a3e5fa0097c11f" }, ] [[package]]