René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"tests.system.TestStreamingPull: test_streaming_pull_max_messages[rest-rest] 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: 26d2b39cefcc1a250f1f75db233f7383a2040b6d\nbuildURL: [Build Status](https://source.cloud.google.com/results/invocations/d84680a4-0558-4f44-a35f-e84864bbad7a), [Sponge](http://sponge2/d84680a4-0558-4f44-a35f-e84864bbad7a)\nstatus: failed\n\u003cdetails>\u003csummary>Test output\u003c/summary>\u003cbr>\u003cpre>self = \u003ctests.system.TestStreamingPull object at 0x7f55475b28f0>\npublisher = \u003cgoogle.cloud.pubsub_v1.PublisherClient object at 0x7f5544354a60>\ntopic_path_base = 'projects/precise-truck-742/topics/t-1689361995945'\nsubscription_path_base = 'projects/precise-truck-742/subscriptions/s-1689361995947'\ncleanup = [(\u003cbound method PublisherClient.delete_topic of \u003cgoogle.cloud.pubsub_v1.PublisherClient object at 0x7f5544354a60>>, ()...39ab0>>, (), {'subscription': 'projects/precise-truck-742/subscriptions/s-1689361995947-streaming-pull-max-messages'})]\n\n    def test_streaming_pull_max_messages(\n        self, publisher, topic_path_base, subscription_path_base, cleanup\n    ):\n        subscriber = pubsub_v1.SubscriberClient(transport=\"grpc\")\n        custom_str = \"-streaming-pull-max-messages\"\n        topic_path = topic_path_base + custom_str\n        subscription_path = subscription_path_base + custom_str\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        # create a topic and subscribe to it\n        publisher.create_topic(name=topic_path)\n        subscriber.create_subscription(name=subscription_path, topic=topic_path)\n    \n        batch_sizes = (7, 4, 8, 2, 10, 1, 3, 8, 6, 1)  # total: 50\n        _publish_messages(publisher, topic_path, batch_sizes=batch_sizes)\n    \n        # now subscribe and do the main part, check for max pending messages\n        total_messages = sum(batch_sizes)\n        flow_control = types.FlowControl(max_messages=5)\n        callback = StreamingPullCallback(\n            processing_time=1, resolve_at_msg_count=total_messages\n        )\n    \n        subscription_future = subscriber.subscribe(\n            subscription_path, callback, flow_control=flow_control\n        )\n    \n        # Expected time to process all messages in ideal case:\n        #     (total_messages / FlowControl.max_messages) * processing_time\n        #\n        # With total=50, max messages=5, and processing_time=1 this amounts to\n        # 10 seconds (+ overhead), thus a full minute should be more than enough\n        # for the processing to complete. If not, fail the test with a timeout.\n        try:\n>           callback.done_future.result(timeout=60)\n\ntests/system.py:654: \n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \n\nself = None, timeout = 60\n\n    def result(self, timeout=None):\n        \"\"\"Return the result of the call that the future represents.\n    \n        Args:\n            timeout: The number of seconds to wait for the result if the future\n                isn't done. If None, then there is no limit on the wait time.\n    \n        Returns:\n            The result of the call that the future represents.\n    \n        Raises:\n            CancelledError: If the future was cancelled.\n            TimeoutError: If the future didn't finish executing before the given\n                timeout.\n            Exception: If the call raised then that exception will be raised.\n        \"\"\"\n        try:\n            with self._condition:\n                if self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]:\n                    raise CancelledError()\n                elif self._state == FINISHED:\n                    return self.__get_result()\n    \n                self._condition.wait(timeout)\n    \n                if self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]:\n                    raise CancelledError()\n                elif self._state == FINISHED:\n                    return self.__get_result()\n                else:\n>                   raise TimeoutError()\nE                   concurrent.futures._base.TimeoutError\n\n/usr/local/lib/python3.10/concurrent/futures/_base.py:448: TimeoutError\n\nDuring handling of the above exception, another exception occurred:\n\nself = \u003ctests.system.TestStreamingPull object at 0x7f55475b28f0>\npublisher = \u003cgoogle.cloud.pubsub_v1.PublisherClient object at 0x7f5544354a60>\ntopic_path_base = 'projects/precise-truck-742/topics/t-1689361995945'\nsubscription_path_base = 'projects/precise-truck-742/subscriptions/s-1689361995947'\ncleanup = [(\u003cbound method PublisherClient.delete_topic of \u003cgoogle.cloud.pubsub_v1.PublisherClient object at 0x7f5544354a60>>, ()...39ab0>>, (), {'subscription': 'projects/precise-truck-742/subscriptions/s-1689361995947-streaming-pull-max-messages'})]\n\n    def test_streaming_pull_max_messages(\n        self, publisher, topic_path_base, subscription_path_base, cleanup\n    ):\n        subscriber = pubsub_v1.SubscriberClient(transport=\"grpc\")\n        custom_str = \"-streaming-pull-max-messages\"\n        topic_path = topic_path_base + custom_str\n        subscription_path = subscription_path_base + custom_str\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        # create a topic and subscribe to it\n        publisher.create_topic(name=topic_path)\n        subscriber.create_subscription(name=subscription_path, topic=topic_path)\n    \n        batch_sizes = (7, 4, 8, 2, 10, 1, 3, 8, 6, 1)  # total: 50\n        _publish_messages(publisher, topic_path, batch_sizes=batch_sizes)\n    \n        # now subscribe and do the main part, check for max pending messages\n        total_messages = sum(batch_sizes)\n        flow_control = types.FlowControl(max_messages=5)\n        callback = StreamingPullCallback(\n            processing_time=1, resolve_at_msg_count=total_messages\n        )\n    \n        subscription_future = subscriber.subscribe(\n            subscription_path, callback, flow_control=flow_control\n        )\n    \n        # Expected time to process all messages in ideal case:\n        #     (total_messages / FlowControl.max_messages) * processing_time\n        #\n        # With total=50, max messages=5, and processing_time=1 this amounts to\n        # 10 seconds (+ overhead), thus a full minute should be more than enough\n        # for the processing to complete. If not, fail the test with a timeout.\n        try:\n            callback.done_future.result(timeout=60)\n        except exceptions.TimeoutError:\n>           pytest.fail(\n                \"Timeout: receiving/processing streamed messages took too long.\"\n            )\nE           Failed: Timeout: receiving/processing streamed messages took too long.\n\ntests/system.py:656: Failed\u003c/pre>\u003c/details>","author":{"url":"https://github.com/apps/flaky-bot","@type":"Person","name":"flaky-bot"},"datePublished":"2023-07-14T19:20:37Z","url":"https://github.com/googleapis/python-pubsub/issues/953"}