René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"PubSub: Unexpected behavior for 2 or more subscribes with the same scheduler","articleBody":"#### Environment details\r\n\r\nOS: Ubuntu 18.04.3 LTS\r\nPython version: 3.7.4\r\ngoogle-cloud-pubsub version: 1.0.2\r\n\r\n#### Steps to reproduce\r\n\r\n  1. Set up a GCP project with enabled Pub/Sub and set up local environment so that the scripts can connect to it.\r\n  2. Run `server.py` (in code examples below). It should create 2 topics (`topic_a` and `topic_b`) and subscriptions to them and start listening to those subscriptions. Notice that scheduler is passed as an argument to receive messages (documentation does not indicate that this should cause any troubles)\r\n 3. Open a new terminal window and run `python3.7 client.py topic_a \u0026` and `python3.7 client.py topic_b \u0026`. This should start sending messages to both topics. (code in the examples below)\r\n 4. The server will output some details about all published messages on both topics. Look at the unacked messages on both subscriptions in the GCP console or stack driver. You will see that some messages are still in the pubsub (which ones is quite random). If you try to pull them, nothing is returned.\r\n 5. Terminate the running server.\r\n 6. Pull messages from either subscription - you should be able to retrieve all of them. Notice that these messages have been processed by the server, but it is as if the `ack` did nothing.\r\n 7. If you restart the server, it will process some of the messages again and some may even be acked out of the subscription. \r\n\r\nWhole time there are no error logs or exceptions thrown, messages are just stuck in the subscription even after being acked. If you change the server code to instantiate thread scheduler inside of the `receive_messages` function, everything works as expected, therefore the shared scheduler is the problem.\r\n\r\n#### Code example\r\n\r\nserver.py\r\n```python\r\nimport concurrent.futures.thread\r\nimport os\r\nimport time\r\n\r\nfrom google.api_core.exceptions import AlreadyExists\r\nfrom google.cloud import pubsub_v1\r\nfrom google.cloud.pubsub_v1.subscriber.scheduler import ThreadScheduler\r\n\r\n\r\ndef create_subscription(project_id, topic_name, subscription_name):\r\n    \"\"\"Create a new pull subscription on the given topic.\"\"\"\r\n    subscriber = pubsub_v1.SubscriberClient()\r\n    topic_path = subscriber.topic_path(project_id, topic_name)\r\n    subscription_path = subscriber.subscription_path(\r\n        project_id, subscription_name)\r\n\r\n    subscription = subscriber.create_subscription(\r\n        subscription_path, topic_path)\r\n\r\n    print('Subscription created: {}'.format(subscription))\r\n\r\n\r\ndef receive_messages(project_id, subscription_name, t_scheduler):\r\n    \"\"\"Receives messages from a pull subscription.\"\"\"\r\n    subscriber = pubsub_v1.SubscriberClient()\r\n    subscription_path = subscriber.subscription_path(\r\n        project_id, subscription_name)\r\n\r\n    def callback(message):\r\n        print('Received message: {}'.format(message.data))\r\n        message.ack()\r\n\r\n    subscriber.subscribe(subscription_path, callback=callback, scheduler=t_scheduler)\r\n    print('Listening for messages on {}'.format(subscription_path))\r\n\r\n\r\nproject_id = os.getenv(\"PUBSUB_PROJECT_ID\")\r\n\r\npublisher = pubsub_v1.PublisherClient()\r\nproject_path = publisher.project_path(project_id)\r\n\r\n# Create both topics\r\ntry:\r\n    topics = [topic.name.split('/')[-1] for topic in publisher.list_topics(project_path)]\r\n    if 'topic_a' not in topics:\r\n        publisher.create_topic(publisher.topic_path(project_id, 'topic_a'))\r\n    if 'topic_b' not in topics:\r\n        publisher.create_topic(publisher.topic_path(project_id, 'topic_b'))\r\nexcept AlreadyExists:\r\n    print('Topics already exists')\r\n\r\n# Create subscriptions on both topics\r\nsub_client = pubsub_v1.SubscriberClient()\r\nproject_path = sub_client.project_path(project_id)\r\n\r\ntry:\r\n    subs = [sub.name.split('/')[-1] for sub in sub_client.list_subscriptions(project_path)]\r\n    if 'topic_a_sub' not in subs:\r\n        create_subscription(project_id, 'topic_a', 'topic_a_sub')\r\n    if 'topic_b_sub' not in subs:\r\n        create_subscription(project_id, 'topic_b', 'topic_b_sub')\r\nexcept AlreadyExists:\r\n    print('Subscriptions already exists')\r\n\r\nscheduler = ThreadScheduler(concurrent.futures.thread.ThreadPoolExecutor(10))\r\n\r\nreceive_messages(project_id, 'topic_a_sub', scheduler)\r\nreceive_messages(project_id, 'topic_b_sub', scheduler)\r\n\r\nwhile True:\r\n    time.sleep(60)\r\n```\r\n\r\nclient.py\r\n```python\r\nimport datetime\r\nimport os\r\nimport random\r\nimport sys\r\nfrom time import sleep\r\n\r\nfrom google.cloud import pubsub_v1\r\n\r\n\r\ndef publish_messages(pid, topic_name):\r\n    \"\"\"Publishes multiple messages to a Pub/Sub topic.\"\"\"\r\n    publisher = pubsub_v1.PublisherClient()\r\n    topic_path = publisher.topic_path(pid, topic_name)\r\n\r\n    for n in range(1, 10):\r\n        data = '[{} - {}] Message number {}'.format(datetime.datetime.now().isoformat(), topic_name, n)\r\n        data = data.encode('utf-8')\r\n        publisher.publish(topic_path, data=data)\r\n        sleep(random.randint(10, 50) / 10.0)\r\n\r\n\r\nproject_id = os.getenv(\"PUBSUB_PROJECT_ID\")\r\npublish_messages(project_id, sys.argv[1])\r\n```\r\n\r\nI have created the [stack overflow thread](https://stackoverflow.com/questions/58289804/gcp-message-stays-in-the-pub-sub-after-acknowledge/58313340#58313340) first, then did my own testing. \r\n\r\n\r\nI would expect that using the same scheduler would not cause any difference in behavior or at the very least some exceptions or warnings to be raised that would indicate that something is wrong. \r\n\r\nAlso, it might be a good idea to warn people not to use the same scheduler for multiple subscriptions (even to different topics) somewhere in the documentation of the library. Due to the nature of the issue (no obvious indication that something is wrong), it might get to production quite easily. ","author":{"url":"https://github.com/LukasSlouka","@type":"Person","name":"LukasSlouka"},"datePublished":"2019-10-17T12:35:15.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":4},"url":"https://github.com/11/python-pubsub/issues/11"}