ref(taskworker): Mark raw-mode namespaces explicitly

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:

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,
 )

src/sentry/replays/tasks.py

@@ -5,7 +5,6 @@
 
 import sentry_sdk
 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
@@ -68,9 +67,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:

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:

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(