René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Resolve Incompatibility with API Core \u003e= 1.17.0","articleBody":"Changes in 1.17.0 are causing an incompatibility with pubsub. \r\n\r\n# Related Issues: \r\n\r\nhttps://github.com/googleapis/python-api-core/issues/25\r\nhttps://github.com/googleapis/python-pubsub/issues/74\r\n\r\n# Suspected changes that caused the behavior change:\r\n\r\nhttps://github.com/googleapis/python-api-core/commit/2b103b60ece16a1e1bc98cfda7ec375191a90f75#diff-d97e81006eaaf29e33270a85aead6aafR146-R148\r\n\r\n# Repro:\r\n\r\n```\r\n\r\nfrom google.cloud import pubsub_v1\r\nimport threading\r\nfrom time import sleep\r\n\r\nimport logging\r\n\r\ndef main():\r\n    logging.basicConfig(level=logging.DEBUG)\r\n    logger = logging.getLogger('sample')\r\n    logger.setLevel(logging.DEBUG)\r\n    PROJECT_ID = \"crwilcox-test\"\r\n    TOPIC_PATH = \"my-topic\"\r\n    SUBSCRIPTION_PATH = \"my-sub\"\r\n\r\n    publisher_thread = threading.Thread(target=publisher, args=(PROJECT_ID, TOPIC_PATH))\r\n    publisher_thread.daemon = True\r\n    publisher_thread.start()\r\n\r\n    subscriber(PROJECT_ID, SUBSCRIPTION_PATH)\r\n\r\ndef publisher(\r\n    project_id, topic_name\r\n):\r\n    # TODO project_id = \"Your Google Cloud Project ID\"\r\n    # TODO topic_name = \"Your Pub/Sub topic name\"\r\n\r\n\r\n    publisher = pubsub_v1.PublisherClient()\r\n    # The `topic_path` method creates a fully qualified identifier\r\n    # in the form `projects/{project_id}/topics/{topic_name}`\r\n    topic_path = publisher.topic_path(project_id, topic_name)\r\n    #publisher.create_topic(topic_path)\r\n\r\n    for n in range(1, 100000):\r\n        data = u\"Message number {}\".format(n)\r\n        # Data must be a bytestring\r\n        data = data.encode(\"utf-8\")\r\n        # When you publish a message, the client returns a future.\r\n        future = publisher.publish(topic_path, data=data)\r\n        print(future.result())\r\n        sleep(5)\r\n\r\n    print(\"Published messages.\")\r\n\r\n\r\n\r\ndef subscriber(\r\n    project_id, subscription_name, timeout=None\r\n):\r\n    # TODO project_id = \"Your Google Cloud Project ID\"\r\n    # TODO subscription_name = \"Your Pub/Sub subscription name\"\r\n    # TODO timeout = 5.0  # \"How long the subscriber should listen for\r\n    # messages in seconds\"\r\n\r\n    subscriber = pubsub_v1.SubscriberClient()\r\n    # The `subscription_path` method creates a fully qualified identifier\r\n    # in the form `projects/{project_id}/subscriptions/{subscription_name}`\r\n    subscription_path = subscriber.subscription_path(\r\n        project_id, subscription_name\r\n    )\r\n\r\n    def callback(message):\r\n        print(\"Received message: {}\".format(message.data))\r\n        message.ack()\r\n\r\n    streaming_pull_future = subscriber.subscribe(\r\n        subscription_path, callback=callback\r\n    )\r\n    print(\"Listening for messages on {}..\\n\".format(subscription_path))\r\n\r\n    # Wrap subscriber in a 'with' block to automatically call close() when done.\r\n    with subscriber:\r\n        try:\r\n            # When `timeout` is not set, result() will block indefinitely,\r\n            # unless an exception is encountered first.\r\n            streaming_pull_future.result(timeout=timeout)\r\n        except:  # noqa\r\n            streaming_pull_future.cancel()\r\n\r\n\r\nmain()\r\n\r\n```\r\n\r\n# Result:\r\n\r\nEventually, subscription will fail and stop processing.\r\n\r\nThe failure exists in bidi on getting the next item via `__iter__`\r\n\r\n```\r\nDEBUG:google.cloud.pubsub_v1.publisher._batch.thread:gRPC Publish took 0.034303903579711914 seconds.\r\n1162512773129262\r\nDEBUG:google.cloud.pubsub_v1.subscriber._protocol.streaming_pull_manager:Sent request(s) over unary RPC.\r\nDEBUG:google.cloud.pubsub_v1.subscriber._protocol.streaming_pull_manager:Released held message, scheduling callback for it, still on hold 0 (bytes 0).\r\nDEBUG:google.api_core.bidi:waiting for recv.\r\nReceived message: b'Message number 117'\r\nDEBUG:google.cloud.pubsub_v1.subscriber._protocol.dispatcher:Handling 1 batched requests\r\nDEBUG:google.cloud.pubsub_v1.subscriber._protocol.streaming_pull_manager:Sent request(s) over unary RPC.\r\niter except: \u003c_MultiThreadedRendezvous of RPC that terminated with:\r\n        status = StatusCode.DEADLINE_EXCEEDED\r\n        details = \"Deadline Exceeded\"\r\n        debug_error_string = \"{\"created\":\"@1588195797.686297000\",\"description\":\"Error received from peer ipv4:216.58.217.42:443\",\"file\":\"src/core/lib/surface/call.cc\",\"file_line\":1056,\"grpc_message\":\"Deadline Exceeded\",\"grpc_status\":4}\"\r\n\u003e\r\nINFO:google.cloud.pubsub_v1.subscriber._protocol.streaming_pull_manager:Observed non-terminating stream error 504 Deadline Exceeded\r\nINFO:google.cloud.pubsub_v1.subscriber._protocol.streaming_pull_manager:Observed recoverable stream error 504 Deadline Exceeded\r\nDEBUG:google.api_core.bidi:Re-opening stream from gRPC callback.\r\nDEBUG:google.api_core.bidi:Empty queue and inactive call, exiting request generator.\r\nDEBUG:google.cloud.pubsub_v1.subscriber._protocol.leaser:The current deadline value is 10 seconds.\r\n```\r\n\r\n```\r\nPYTHON_PATH/lib/python3.8/threading.py(890)_bootstrap()\r\n-\u003e self._bootstrap_inner()\r\n  PYTHON_PATH/lib/python3.8/threading.py(932)_bootstrap_inner()\r\n-\u003e self.run()\r\n  PYTHON_PATH/lib/python3.8/threading.py(870)run()\r\n-\u003e self._target(*self._args, **self._kwargs)\r\n  /workspace/googleapis/python-pubsub/venv/lib/python3.8/site-packages/google/api_core/bidi.py(655)_thread_main()\r\n-\u003e response = self._bidi_rpc.recv()\r\n  /workspace/googleapis/python-pubsub/venv/lib/python3.8/site-packages/google/api_core/bidi.py(562)recv()\r\n-\u003e return self._recoverable(self._recv)\r\n  /workspace/googleapis/python-pubsub/venv/lib/python3.8/site-packages/google/api_core/bidi.py(505)_recoverable()\r\n-\u003e return method(*args, **kwargs)\r\n  /workspace/googleapis/python-pubsub/venv/lib/python3.8/site-packages/google/api_core/bidi.py(559)_recv()\r\n-\u003e return next(call)\r\n\u003e /workspace/googleapis/python-pubsub/venv/lib/python3.8/site-packages/google/api_core/grpc_helpers.py(109)next()\r\n-\u003e six.raise_from(exceptions.from_grpc_error(exc), exc)\r\n```","author":{"url":"https://github.com/crwilcox","@type":"Person","name":"crwilcox"},"datePublished":"2020-04-29T21:33:15.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":7},"url":"https://github.com/93/python-pubsub/issues/93"}