René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"tests.system.TestStreamingPull: test_streaming_pull_blocking_shutdown[grpc-grpc] failed","articleBody":"This test failed!\n\nTo configure my behavior, see [the Flaky Bot documentation](https://github.com/googleapis/repo-automation-bots/tree/main/packages/flakybot).\n\nIf I'm commenting on this issue too often, add the `flakybot: quiet` label and\nI will stop commenting.\n\n---\n\ncommit: 1dcff30f6254cb6429565fe74b9c8cdc3e0338bb\nbuildURL: [Build Status](https://source.cloud.google.com/results/invocations/69291d8b-3212-412a-bd6f-d953dea8c4a1), [Sponge](http://sponge2/69291d8b-3212-412a-bd6f-d953dea8c4a1)\nstatus: failed\n\u003cdetails\u003e\u003csummary\u003eTest output\u003c/summary\u003e\u003cbr\u003e\u003cpre\u003eself = \u003ctests.system.TestStreamingPull object at 0x7fd8dca2b520\u003e\npublisher = \u003cgoogle.cloud.pubsub_v1.PublisherClient object at 0x7fd8b0700c70\u003e\ntopic_path = 'projects/precise-truck-742/topics/t-1678138055460'\nsubscriber = \u003cgoogle.cloud.pubsub_v1.SubscriberClient object at 0x7fd8dcaf3850\u003e\nsubscription_path = 'projects/precise-truck-742/subscriptions/s-1678138055462'\ncleanup = [(\u003cbound method PublisherClient.delete_topic of \u003cgoogle.cloud.pubsub_v1.PublisherClient object at 0x7fd8b0700c70\u003e\u003e, ()...erClient object at 0x7fd8dcaf3850\u003e\u003e, (), {'subscription': 'projects/precise-truck-742/subscriptions/s-1678138055462'})]\n\n    def test_streaming_pull_blocking_shutdown(\n        self, publisher, topic_path, subscriber, subscription_path, cleanup\n    ):\n        # Make sure the topic and subscription get deleted.\n        cleanup.append((publisher.delete_topic, (), {\"topic\": topic_path}))\n        cleanup.append(\n            (subscriber.delete_subscription, (), {\"subscription\": subscription_path})\n        )\n    \n        # The ACK-s are only persisted if *all* messages published in the same batch\n        # are ACK-ed. We thus publish each message in its own batch so that the backend\n        # treats all messages' ACKs independently of each other.\n        publisher.create_topic(name=topic_path)\n        subscriber.create_subscription(name=subscription_path, topic=topic_path)\n        _publish_messages(publisher, topic_path, batch_sizes=[1] * 10)\n    \n        # Artificially delay message processing, gracefully shutdown the streaming pull\n        # in the meantime, then verify that those messages were nevertheless processed.\n        processed_messages = []\n    \n        def callback(message):\n            time.sleep(15)\n            processed_messages.append(message.data)\n            message.ack()\n    \n        # Flow control limits should exceed the number of worker threads, so that some\n        # of the messages will be blocked on waiting for free scheduler threads.\n        flow_control = pubsub_v1.types.FlowControl(max_messages=5)\n        executor = concurrent.futures.ThreadPoolExecutor(max_workers=3)\n        scheduler = pubsub_v1.subscriber.scheduler.ThreadScheduler(executor=executor)\n        subscription_future = subscriber.subscribe(\n            subscription_path,\n            callback=callback,\n            flow_control=flow_control,\n            scheduler=scheduler,\n            await_callbacks_on_shutdown=True,\n        )\n    \n        try:\n            subscription_future.result(timeout=10)  # less than the sleep in callback\n        except exceptions.TimeoutError:\n            subscription_future.cancel()\n            subscription_future.result()  # block until shutdown completes\n    \n        # Blocking om shutdown should have waited for the already executing\n        # callbacks to finish.\n        assert len(processed_messages) == 3\n    \n        # The messages that were not processed should have been NACK-ed and we should\n        # receive them again quite soon.\n        all_done = threading.Barrier(7 + 1, timeout=5)  # +1 because of the main thread\n        remaining = []\n    \n        def callback2(message):\n            remaining.append(message.data)\n            message.ack()\n            all_done.wait()\n    \n        subscription_future = subscriber.subscribe(\n            subscription_path, callback=callback2, await_callbacks_on_shutdown=False\n        )\n    \n        try:\n\u003e           all_done.wait()\n\ntests/system.py:685: \n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \n\nself = \u003cthreading.Barrier object at 0x7fd8d898bf10\u003e, timeout = 5\n\n    def wait(self, timeout=None):\n        \"\"\"Wait for the barrier.\n    \n        When the specified number of threads have started waiting, they are all\n        simultaneously awoken. If an 'action' was provided for the barrier, one\n        of the threads will have executed that callback prior to returning.\n        Returns an individual index number from 0 to 'parties-1'.\n    \n        \"\"\"\n        if timeout is None:\n            timeout = self._timeout\n        with self._cond:\n            self._enter() # Block while the barrier drains.\n            index = self._count\n            self._count += 1\n            try:\n                if index + 1 == self._parties:\n                    # We release the barrier\n                    self._release()\n                else:\n                    # We wait until someone releases us\n\u003e                   self._wait(timeout)\n\n/usr/local/lib/python3.10/threading.py:668: \n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \n\nself = \u003cthreading.Barrier object at 0x7fd8d898bf10\u003e, timeout = 5\n\n    def _wait(self, timeout):\n        if not self._cond.wait_for(lambda : self._state != 0, timeout):\n            #timed out.  Break the barrier\n            self._break()\n\u003e           raise BrokenBarrierError\nE           threading.BrokenBarrierError\n\n/usr/local/lib/python3.10/threading.py:706: BrokenBarrierError\n\nDuring handling of the above exception, another exception occurred:\n\nself = \u003ctests.system.TestStreamingPull object at 0x7fd8dca2b520\u003e\npublisher = \u003cgoogle.cloud.pubsub_v1.PublisherClient object at 0x7fd8b0700c70\u003e\ntopic_path = 'projects/precise-truck-742/topics/t-1678138055460'\nsubscriber = \u003cgoogle.cloud.pubsub_v1.SubscriberClient object at 0x7fd8dcaf3850\u003e\nsubscription_path = 'projects/precise-truck-742/subscriptions/s-1678138055462'\ncleanup = [(\u003cbound method PublisherClient.delete_topic of \u003cgoogle.cloud.pubsub_v1.PublisherClient object at 0x7fd8b0700c70\u003e\u003e, ()...erClient object at 0x7fd8dcaf3850\u003e\u003e, (), {'subscription': 'projects/precise-truck-742/subscriptions/s-1678138055462'})]\n\n    def test_streaming_pull_blocking_shutdown(\n        self, publisher, topic_path, subscriber, subscription_path, cleanup\n    ):\n        # Make sure the topic and subscription get deleted.\n        cleanup.append((publisher.delete_topic, (), {\"topic\": topic_path}))\n        cleanup.append(\n            (subscriber.delete_subscription, (), {\"subscription\": subscription_path})\n        )\n    \n        # The ACK-s are only persisted if *all* messages published in the same batch\n        # are ACK-ed. We thus publish each message in its own batch so that the backend\n        # treats all messages' ACKs independently of each other.\n        publisher.create_topic(name=topic_path)\n        subscriber.create_subscription(name=subscription_path, topic=topic_path)\n        _publish_messages(publisher, topic_path, batch_sizes=[1] * 10)\n    \n        # Artificially delay message processing, gracefully shutdown the streaming pull\n        # in the meantime, then verify that those messages were nevertheless processed.\n        processed_messages = []\n    \n        def callback(message):\n            time.sleep(15)\n            processed_messages.append(message.data)\n            message.ack()\n    \n        # Flow control limits should exceed the number of worker threads, so that some\n        # of the messages will be blocked on waiting for free scheduler threads.\n        flow_control = pubsub_v1.types.FlowControl(max_messages=5)\n        executor = concurrent.futures.ThreadPoolExecutor(max_workers=3)\n        scheduler = pubsub_v1.subscriber.scheduler.ThreadScheduler(executor=executor)\n        subscription_future = subscriber.subscribe(\n            subscription_path,\n            callback=callback,\n            flow_control=flow_control,\n            scheduler=scheduler,\n            await_callbacks_on_shutdown=True,\n        )\n    \n        try:\n            subscription_future.result(timeout=10)  # less than the sleep in callback\n        except exceptions.TimeoutError:\n            subscription_future.cancel()\n            subscription_future.result()  # block until shutdown completes\n    \n        # Blocking om shutdown should have waited for the already executing\n        # callbacks to finish.\n        assert len(processed_messages) == 3\n    \n        # The messages that were not processed should have been NACK-ed and we should\n        # receive them again quite soon.\n        all_done = threading.Barrier(7 + 1, timeout=5)  # +1 because of the main thread\n        remaining = []\n    \n        def callback2(message):\n            remaining.append(message.data)\n            message.ack()\n            all_done.wait()\n    \n        subscription_future = subscriber.subscribe(\n            subscription_path, callback=callback2, await_callbacks_on_shutdown=False\n        )\n    \n        try:\n            all_done.wait()\n        except threading.BrokenBarrierError:  # PRAGMA: no cover\n\u003e           pytest.fail(\"The remaining messages have not been re-delivered in time.\")\nE           Failed: The remaining messages have not been re-delivered in time.\n\ntests/system.py:687: Failed\u003c/pre\u003e\u003c/details\u003e","author":{"url":"https://github.com/flaky-bot[bot]","@type":"Person","name":"flaky-bot[bot]"},"datePublished":"2023-03-06T21:29:58.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":1},"url":"https://github.com/879/python-pubsub/issues/879"}