fix(taskbroker): Override Topic and Namespace in Task Sender Locally

src/sentry/runner/commands/run.py

@@ -346,15 +346,15 @@ def taskbroker_send_tasks(
     namespace: str,
     extra_arg_bytes: int | None,
 ) -> None:
-    from sentry import options
     from sentry.conf.server import KAFKA_CLUSTERS
+    from sentry.taskworker.adapters import set_route_overrides
     from sentry.utils.imports import import_string
 
     if bootstrap_servers:
         KAFKA_CLUSTERS["default"]["common"]["bootstrap.servers"] = bootstrap_servers
 
     if kafka_topic and namespace:
-        options.set("taskworker.route.overrides", {namespace: kafka_topic})
+        set_route_overrides({namespace: kafka_topic})
 
     try:
         func = import_string(task_function_path)

src/sentry/taskworker/adapters.py

@@ -39,6 +39,19 @@
 
 logger = logging.getLogger(__name__)
 
+# Map from namespaces to topics that ONLY applies to the current process (no global options)
+_route_overrides: dict[str, str] = {}
+
+
+#
+def set_route_overrides(overrides: dict[str, str]) -> None:
+    """
+    Replace route overrides (`_route_overrides`) with the provided map.
+    This change is local to the running application and does not impact global options.
+    """
+    _route_overrides.clear()
+    _route_overrides.update(overrides)
+
 
 class DjangoCacheAtMostOnceStore(AtMostOnceStore):
     """
@@ -149,11 +162,21 @@ def __init__(self) -> None:
         )
 
     def route_namespace(self, name: str) -> str:
+        # Check local overrides
+        if name in _route_overrides:
+            return Topic(_route_overrides[name]).value
+
+        # Check global overrides
         overrides = options.get("taskworker.route.overrides")
+
         if name in overrides:
             return Topic(overrides[name]).value
+
+        # Check for configured mapping
         if name in self._route_map:
             return Topic(self._route_map[name]).value
+
+        # Fall back onto the default topic
         return self._default_topic.value