Repository: mercadona/rele

Issue:
Worker should not crash when a topic does not exist
When a worker starts it tries to create subscriptions for all topics. If a topic does not exist the worker will crash.

Proposed change:
catch the `Rendezvous` exception (404 Resource not found), log an error but gracefully create and consume the rest of the defined subscriptions

Discussion:

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a99b266..1c24dc9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,10 @@
 Changelog
 =========
 
+`0.9.1` (2020-1-2)
+* Ack messages when data not json serializable (#141)
+* Use ThreadScheduler instead of ThreadPoolExecutor (#145)
+
 `0.9.0` (2019-12-20)
 * Flask support via middleware (#127)
 * Add message attributes to metrics log (#128)
diff --git a/docs/api/settings.rst b/docs/api/settings.rst
index 31c8d75..73b8f64 100644
--- a/docs/api/settings.rst
+++ b/docs/api/settings.rst
@@ -123,6 +123,8 @@ Ack deadline for all subscribers in seconds.
     passes, the message is no longer considered outstanding, and Cloud Pub/Sub will attempt
     to redeliver the message.*
 
+.. _settings_publisher_timeout:
+
 ``PUBLISHER_TIMEOUT``
 ---------------------
 
@@ -136,7 +138,7 @@ Timeout that the publishing result will wait on the future to publish successful
 <https://googleapis.dev/python/pubsub/1.1.0/publisher/api/futures.html?highlight=result#google.cloud.pubsub_v1.publisher.futures.Future.result>`_
 
 ``THREADS_PER_SUBSCRIPTION``
-------------------
+----------------------------
 
 **Optional**
 
diff --git a/rele/__init__.py b/rele/__init__.py
index 9f2820e..71852af 100644
--- a/rele/__init__.py
+++ b/rele/__init__.py
@@ -1,4 +1,4 @@
-__version__ = "0.9.0"
+__version__ = "0.9.1"
 default_app_config = "rele.apps.ReleConfig"
 
 from .client import Publisher, Subscriber  # noqa
diff --git a/rele/client.py b/rele/client.py
index 53a02d1..062a82e 100644
--- a/rele/client.py
+++ b/rele/client.py
@@ -67,11 +67,14 @@ class Subscriber:
         topic_path = self._client.topic_path(self._gc_project_id, topic)
 
         with suppress(exceptions.AlreadyExists):
-            self._client.create_subscription(
-                name=subscription_path,
-                topic=topic_path,
-                ack_deadline_seconds=self._ack_deadline,
-            )
+            try:
+                self._client.create_subscription(
+                    name=subscription_path,
+                    topic=topic_path,
+                    ack_deadline_seconds=self._ack_deadline,
+                )
+            except exceptions.NotFound:
+                logger.error("Cannot subscribe to a topic that does not exist.")
 
     def consume(self, subscription_name, callback, scheduler):
         """Begin listening to topic from the SubscriberClient.
@@ -104,7 +107,7 @@ class Publisher:
     :param gc_project_id: string Google Cloud Project ID.
     :param credentials: string Google Cloud Credentials.
     :param encoder: A valid `json.encoder.JSONEncoder subclass <https://docs.python.org/3/library/json.html#json.JSONEncoder>`_  # noqa
-    :param timeout: float
+    :param timeout: float, default :ref:`settings_publisher_timeout`
     """
 
     def __init__(self, gc_project_id, credentials, encoder, timeout):
@@ -116,7 +119,7 @@ class Publisher:
         else:
             self._client = pubsub_v1.PublisherClient(credentials=credentials)
 
-    def publish(self, topic, data, blocking=False, **attrs):
+    def publish(self, topic, data, blocking=False, timeout=None, **attrs):
         """Publishes message to Google PubSub topic.
 
         Usage::
@@ -145,6 +148,7 @@ class Publisher:
         :param topic: string topic to publish the data.
         :param data: dict with the content of the message.
         :param blocking: boolean
+        :param timeout: float, default None fallsback to :ref:`settings_publisher_timeout`
         :param attrs: Extra parameters to be published.
         :return: `Future <https://googleapis.github.io/google-cloud-python/latest/pubsub/subscriber/api/futures.html>`_  # noqa
         """
@@ -157,6 +161,6 @@ class Publisher:
         if not blocking:
             return future
 
-        future.result(timeout=self._timeout)
+        future.result(timeout=timeout or self._timeout)
         run_middleware_hook("post_publish", topic)
         return future


Test patch:
diff --git a/tests/test_client.py b/tests/test_client.py
index 954ff06..2e39828 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -3,6 +3,7 @@ import concurrent
 from unittest.mock import ANY, patch
 
 import pytest
+from google.api_core import exceptions
 from google.cloud.pubsub_v1 import SubscriberClient
 
 
@@ -63,6 +64,21 @@ class TestPublisher:
         )
         mock_future.assert_called_once_with(timeout=100)
 
+    @patch.object(concurrent.futures.Future, "result")
+    def test_publishes_data_with_client_timeout_when_blocking_and_timeout_specified(
+        self, mock_future, publisher
+    ):
+        publisher._timeout = 100.0
+        publisher.publish(
+            topic="order-cancelled", data={"foo": "bar"}, blocking=True, timeout=50
+        )
+
+        publisher._client.publish.return_value = mock_future
+        publisher._client.publish.assert_called_with(
+            ANY, b'{"foo": "bar"}', published_at=ANY
+        )
+        mock_future.assert_called_once_with(timeout=50)
+
 
 class TestSubscriber:
     @patch.object(SubscriberClient, "create_subscription")
@@ -92,3 +108,34 @@ class TestSubscriber:
         _mocked_client.assert_called_once_with(
             ack_deadline_seconds=100, name=expected_subscription, topic=expected_topic
         )
+
+    @patch.object(
+        SubscriberClient,
+        "create_subscription",
+        side_effect=exceptions.AlreadyExists("Subscription already exists"),
+    )
+    def test_does_not_raise_when_subscription_already_exists(
+        self, _mocked_client, project_id, subscriber
+    ):
+        subscriber.create_subscription(
+            subscription="test-topic", topic=f"{project_id}-test-topic"
+        )
+
+        _mocked_client.assert_called()
+
+    @patch.object(
+        SubscriberClient,
+        "create_subscription",
+        side_effect=exceptions.NotFound("Subscription topic does not exist"),
+    )
+    def test_logs_error_when_subscription_topic_does_not_exist(
+        self, _mocked_client, project_id, subscriber, caplog
+    ):
+        subscriber.create_subscription(
+            subscription="test-topic", topic=f"{project_id}-test-topic"
+        )
+
+        _mocked_client.assert_called()
+        log = caplog.records[0]
+        assert log.message == "Cannot subscribe to a topic that does not exist."
+        assert log.levelname == "ERROR"