Use an explicit kombu.Connection while sending authority tasks

h/services/annotation_authority_queue.py

@@ -3,6 +3,7 @@
 from dataclasses import dataclass
 
 from celery import Celery
+from kombu import Connection
 
 from h.services.annotation_read import AnnotationReadService
 
@@ -54,19 +55,20 @@ def publish(self, event_action: str, annotation_id: str) -> None:
         }
 
         authority_celery = Celery(annotation.authority)
-        authority_celery.conf.broker_url = authority_queue_config.broker_url
-
-        authority_celery.send_task(
-            authority_queue_config.task_name,
-            queue=authority_queue_config.queue_name,
-            kwargs={"event": payload},
-        )
-        LOG.info(
-            "Published event %s for annotation %s to %s",
-            event_action,
-            annotation.id,
-            annotation.authority,
-        )
+        with Connection(authority_queue_config.broker_url) as connection:
+            authority_celery.send_task(
+                authority_queue_config.task_name,
+                queue=authority_queue_config.queue_name,
+                # We need to pass the connection explicitly to avoid using the default connection / broker
+                connection=connection,
+                kwargs={"event": payload},
+            )
+            LOG.info(
+                "Published event %s for annotation %s to %s",
+                event_action,
+                annotation.id,
+                annotation.authority,
+            )
 
     def _parse_authority_queue_config(
         self, config_json: str | None

tests/unit/h/services/annotation_authority_queue_test.py

@@ -36,7 +36,13 @@ def test_publish_with_no_mentions(self, svc, Celery, annotation_read_service):
         Celery.assert_not_called()
 
     def test_publish(
-        self, svc, Celery, annotation_read_service, annotation_json_service, annotation
+        self,
+        svc,
+        Celery,
+        annotation_read_service,
+        annotation_json_service,
+        annotation,
+        Connection,
     ):
         annotation_read_service.get_annotation_by_id.return_value = annotation
 
@@ -51,12 +57,13 @@ def test_publish(
             with_metadata=True,
         )
         Celery.assert_called_once_with(
-            annotation_read_service.get_annotation_by_id.return_value.authority
+            annotation_read_service.get_annotation_by_id.return_value.authority,
         )
-        Celery.return_value.conf.broker_url = "broker_url"
+        Connection.assert_called_once_with("url")
         Celery.return_value.send_task.assert_called_once_with(
             "task",
             queue="queue",
+            connection=Connection.return_value.__enter__.return_value,
             kwargs={
                 "event": {
                     "action": "create",
@@ -87,6 +94,10 @@ def test_parse_config_with_config(self, svc, valid_config):
     def Celery(self, patch):
         return patch("h.services.annotation_authority_queue.Celery")
 
+    @pytest.fixture
+    def Connection(self, patch):
+        return patch("h.services.annotation_authority_queue.Connection")
+
     @pytest.fixture
     def valid_config(self):
         return json.dumps(