René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"`bidi.BackgroundConsumer.start()` eats exception when `_bidi_rpc.open()` fails","articleBody":"#### Environment details\r\n\r\n  - OS type and version:\r\n  - Python version: `python --version`\r\n\r\nPython 3.9.5\r\n\r\n  - pip version: `pip --version`\r\n\r\n```\r\nconda 4.9.2\r\npip 21.2.1 from /usr/local/Caskroom/miniconda/base/envs/dev-3.9/lib/python3.9/site-packages/pip (python 3.9)\r\n```\r\n\r\n  - `google-api-core` version: `pip show google-api-core`\r\n\r\n```\r\nconda list google-api-core        # packages in environment at /usr/local/Caskroom/miniconda/base/envs/dev-3.9:\r\n#\r\n# Name                    Version                   Build  Channel\r\ngoogle-api-core           2.0.1                     dev_0    \u003cdevelop @ dcb6ebd9994fddcb1729150df1675ebf8c503a73\u003e\r\n```\r\n\r\n\r\n\r\n#### Steps to reproduce\r\n\r\n1. Construct a BackgroundConsumer and BidiRpc.\r\n2. Intentionally make the initial request one that will fail.\r\n3. Observe that RPC \"done callback\" is never called. Nor is any notification sent to `on_response`.\r\n\r\n\r\n#### Code example\r\n\r\n```python\r\n        start_time = time.monotonic()\r\n        self._inital_request = initial_request\r\n        self._stream_name = initial_request.write_stream\r\n\r\n        inital_response_future = AppendRowsFuture()\r\n        self._futures_queue.put(inital_response_future)\r\n\r\n        self._rpc = bidi.BidiRpc(\r\n            self._client.append_rows,\r\n            initial_request=self._inital_request,\r\n            # TODO: pass in retry and timeout. Blocked by\r\n            # https://github.com/googleapis/python-api-core/issues/262\r\n            metadata=tuple(\r\n                itertools.chain(\r\n                    self._metadata,\r\n                    # This header is required so that the BigQuery Storage API\r\n                    # knows which region to route the request to.\r\n                    ((\"x-goog-request-params\", f\"write_stream={self._stream_name}\"),),\r\n                )\r\n            ),\r\n        )\r\n        self._rpc.add_done_callback(self._on_rpc_done)\r\n\r\n        self._consumer = bidi.BackgroundConsumer(self._rpc, self._on_response)\r\n        self._consumer.start()\r\n\r\n        # Make sure RPC has started before returning.\r\n        # Without this, consumers may get:\r\n        #\r\n        # ValueError: Can not send() on an RPC that has never been open()ed.\r\n        #\r\n        # when they try to send a request.\r\n        while not self._rpc.is_active and self._consumer.is_active:\r\n            # Avoid 100% CPU while waiting for RPC to be ready.\r\n            time.sleep(_WRITE_OPEN_INTERVAL)\r\n\r\n            # TODO: Check retry.deadline instead of (per-request) timeout.\r\n            # Blocked by\r\n            # https://github.com/googleapis/python-api-core/issues/262\r\n            if timeout is None:\r\n                continue\r\n            current_time = time.monotonic()\r\n            if current_time - start_time \u003e timeout:\r\n                break\r\n\r\n        return inital_response_future\r\n```\r\n\r\n```\r\n...\r\n\r\n    request = types.AppendRowsRequest()\r\n    request.write_stream = \"invalid\"\r\n    proto_data = types.AppendRowsRequest.ProtoData()\r\n    proto_data.rows = proto_rows\r\n\r\n...\r\n\r\n    request.offset = 0\r\n    # requests = generate_sample_data(write_stream.name)\r\n    append_rows_stream, response_future_1 = write_client.append_rows(request)\r\n```\r\n\r\nSee https://github.com/googleapis/python-bigquery-storage/pull/284\r\n\r\n#### Stack trace\r\n\r\nThe problem is that there isn't a stacktrace, but here's the output when I turn on DEBUG logging:\r\n\r\n```\r\n$ pytest 'samples/snippets/append_rows_proto2_test.py::test_append_rows_proto2[US]' --log-cli-level=DEBUG\r\n================================ test session starts ================================\r\nplatform darwin -- Python 3.9.5, pytest-6.2.4, py-1.10.0, pluggy-0.13.1\r\nrootdir: /Users/swast/src/github.com/googleapis/python-bigquery-storage\r\nplugins: cov-2.12.1, asyncio-0.15.1, requests-mock-1.9.3\r\ncollected 1 item                                                                    \r\n\r\nsamples/snippets/append_rows_proto2_test.py::test_append_rows_proto2[US] \r\n---------------------------------- live log setup -----------------------------------\r\nDEBUG    google.auth._default:_default.py:206 Checking None for explicit credentials as part of auth process...\r\nDEBUG    google.auth._default:_default.py:181 Checking Cloud SDK credentials as part of auth process...\r\nDEBUG    google.cloud.bigquery.opentelemetry_tracing:opentelemetry_tracing.py:66 This service is instrumented using OpenTelemetry. OpenTelemetry could not be imported; please add opentelemetry-api and opentelemetry-instrumentation packages in order to get BigQuery Tracing data.\r\nDEBUG    urllib3.util.retry:retry.py:333 Converted retries value: 3 -\u003e Retry(total=3, connect=None, read=None, redirect=None, status=None)\r\nDEBUG    google.auth.transport.requests:requests.py:182 Making request: POST https://oauth2.googleapis.com/token\r\nDEBUG    urllib3.connectionpool:connectionpool.py:971 Starting new HTTPS connection (1): oauth2.googleapis.com:443\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://oauth2.googleapis.com:443 \"POST /token HTTP/1.1\" 200 None\r\nDEBUG    urllib3.connectionpool:connectionpool.py:971 Starting new HTTPS connection (1): bigquery.googleapis.com:443\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://bigquery.googleapis.com:443 \"GET /bigquery/v2/projects/swast-scratch/datasets?prettyPrint=false HTTP/1.1\" 200 None\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://bigquery.googleapis.com:443 \"POST /bigquery/v2/projects/swast-scratch/datasets?prettyPrint=false HTTP/1.1\" 200 None\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://bigquery.googleapis.com:443 \"POST /bigquery/v2/projects/swast-scratch/datasets?prettyPrint=false HTTP/1.1\" 200 None\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://bigquery.googleapis.com:443 \"POST /bigquery/v2/projects/swast-scratch/datasets/python_bigquery_storage_samples_snippets_20210831210924_9f81bc/tables?prettyPrint=false HTTP/1.1\" 200 None\r\n----------------------------------- live log call -----------------------------------\r\nDEBUG    google.auth._default:_default.py:206 Checking None for explicit credentials as part of auth process...\r\nDEBUG    google.auth._default:_default.py:181 Checking Cloud SDK credentials as part of auth process...\r\nDEBUG    google.auth.transport.requests:requests.py:182 Making request: POST https://oauth2.googleapis.com/token\r\nDEBUG    urllib3.connectionpool:connectionpool.py:971 Starting new HTTPS connection (1): oauth2.googleapis.com:443\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://oauth2.googleapis.com:443 \"POST /token HTTP/1.1\" 200 None\r\nDEBUG    google.api_core.bidi:bidi.py:695 Started helper thread Thread-ConsumeBidirectionalStream\r\nDEBUG    google.api_core.bidi:bidi.py:660 Thread-ConsumeBidirectionalStream caught error 400 Invalid stream name. Entity: invalid and will exit. Generally this is due to the RPC itself being cancelled and the error will be surfaced to the calling code.\r\nTraceback (most recent call last):\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/grpc_helpers.py\", line 156, in error_remapped_callable\r\n    return _StreamingResponseIterator(\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/grpc_helpers.py\", line 82, in __init__\r\n    self._stored_first_result = next(self._wrapped)\r\n  File \"/usr/local/Caskroom/miniconda/base/envs/dev-3.9/lib/python3.9/site-packages/grpc/_channel.py\", line 426, in __next__\r\n    return self._next()\r\n  File \"/usr/local/Caskroom/miniconda/base/envs/dev-3.9/lib/python3.9/site-packages/grpc/_channel.py\", line 826, in _next\r\n    raise self\r\ngrpc._channel._MultiThreadedRendezvous: \u003c_MultiThreadedRendezvous of RPC that terminated with:\r\n        status = StatusCode.INVALID_ARGUMENT\r\n        details = \"Invalid stream name. Entity: invalid\"\r\n        debug_error_string = \"{\"created\":\"@1630444168.199761000\",\"description\":\"Error received from peer ipv6:[2607:f8b0:4009:819::200a]:443\",\"file\":\"src/core/lib/surface/call.cc\",\"file_line\":1067,\"grpc_message\":\"Invalid stream name. Entity: invalid\",\"grpc_status\":3}\"\r\n\u003e\r\n\r\nThe above exception was the direct cause of the following exception:\r\n\r\nTraceback (most recent call last):\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/bidi.py\", line 636, in _thread_main\r\n    self._bidi_rpc.open()\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/bidi.py\", line 279, in open\r\n    call = self._start_rpc(iter(request_generator), metadata=self._rpc_metadata)\r\n  File \"/Users/swast/src/github.com/googleapis/python-bigquery-storage/google/cloud/bigquery_storage_v1beta2/services/big_query_write/client.py\", line 522, in append_rows\r\n    response = rpc(requests, retry=retry, timeout=timeout, metadata=metadata,)\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/gapic_v1/method.py\", line 142, in __call__\r\n    return wrapped_func(*args, **kwargs)\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/retry.py\", line 283, in retry_wrapped_func\r\n    return retry_target(\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/retry.py\", line 190, in retry_target\r\n    return target()\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/grpc_helpers.py\", line 160, in error_remapped_callable\r\n    raise exceptions.from_grpc_error(exc) from exc\r\ngoogle.api_core.exceptions.InvalidArgument: 400 Invalid stream name. Entity: invalid\r\nINFO     google.api_core.bidi:bidi.py:676 Thread-ConsumeBidirectionalStream exiting\r\nFAILED                                                                        [100%]\r\n--------------------------------- live log teardown ---------------------------------\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://bigquery.googleapis.com:443 \"DELETE /bigquery/v2/projects/swast-scratch/datasets/python_bigquery_storage_samples_snippets_20210831210924_9f81bc/tables/append_rows_proto2_9096?prettyPrint=false HTTP/1.1\" 200 None\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://bigquery.googleapis.com:443 \"DELETE /bigquery/v2/projects/swast-scratch/datasets/python_bigquery_storage_samples_snippets_20210831210924_5c5f49?deleteContents=true\u0026prettyPrint=false HTTP/1.1\" 200 None\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://bigquery.googleapis.com:443 \"DELETE /bigquery/v2/projects/swast-scratch/datasets/python_bigquery_storage_samples_snippets_20210831210924_9f81bc?deleteContents=true\u0026prettyPrint=false HTTP/1.1\" 200 None\r\n\r\n\r\n===================================== FAILURES ======================================\r\n____________________________ test_append_rows_proto2[US] ____________________________\r\n\r\ncapsys = \u003c_pytest.capture.CaptureFixture object at 0x110c67370\u003e\r\nbigquery_client = \u003cgoogle.cloud.bigquery.client.Client object at 0x110c6de20\u003e\r\nsample_data_table = 'swast-scratch.python_bigquery_storage_samples_snippets_20210831210924_9f81bc.append_rows_proto2_9096'\r\n\r\n\u003e   ???\r\n\r\n/Users/swast/src/python-bigquery-storage/samples/snippets/append_rows_proto2_test.py:58: \r\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _\r\nsamples/snippets/append_rows_proto2.py:180: in append_rows_proto2\r\n    response_future_2 = append_rows_stream.send(request)\r\ngoogle/cloud/bigquery_storage_v1beta2/writer.py:206: in send\r\n    self._rpc.send(request)\r\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _\r\n\r\nself = \u003cgoogle.api_core.bidi.BidiRpc object at 0x110d7b760\u003e\r\nrequest = offset {\r\n  value: 6\r\n}\r\nproto_rows {\r\n  rows {\r\n    serialized_rows: \"0\\243\\223\\001\\200\\001\\007\"\r\n    serialized_rows: \":\\0...ed_rows: \"Z\\01711:07:48.123456\\200\\001\\013\"\r\n    serialized_rows: \"`\\206\\341\\307\\343\\357\\253\\362\\002\\200\\001\\014\"\r\n  }\r\n}\r\n\r\n\r\n    def send(self, request):\r\n        \"\"\"Queue a message to be sent on the stream.\r\n    \r\n        Send is non-blocking.\r\n    \r\n        If the underlying RPC has been closed, this will raise.\r\n    \r\n        Args:\r\n            request (protobuf.Message): The request to send.\r\n        \"\"\"\r\n        if self.call is None:\r\n\u003e           raise ValueError(\"Can not send() on an RPC that has never been open()ed.\")\r\nE           ValueError: Can not send() on an RPC that has never been open()ed.\r\n\r\n../python-api-core/google/api_core/bidi.py:315: ValueError\r\n-------------------------------- Captured log setup ---------------------------------\r\nDEBUG    google.auth._default:_default.py:206 Checking None for explicit credentials as part of auth process...\r\nDEBUG    google.auth._default:_default.py:181 Checking Cloud SDK credentials as part of auth process...\r\nDEBUG    google.cloud.bigquery.opentelemetry_tracing:opentelemetry_tracing.py:66 This service is instrumented using OpenTelemetry. OpenTelemetry could not be imported; please add opentelemetry-api and opentelemetry-instrumentation packages in order to get BigQuery Tracing data.\r\nDEBUG    urllib3.util.retry:retry.py:333 Converted retries value: 3 -\u003e Retry(total=3, connect=None, read=None, redirect=None, status=None)\r\nDEBUG    google.auth.transport.requests:requests.py:182 Making request: POST https://oauth2.googleapis.com/token\r\nDEBUG    urllib3.connectionpool:connectionpool.py:971 Starting new HTTPS connection (1): oauth2.googleapis.com:443\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://oauth2.googleapis.com:443 \"POST /token HTTP/1.1\" 200 None\r\nDEBUG    urllib3.connectionpool:connectionpool.py:971 Starting new HTTPS connection (1): bigquery.googleapis.com:443\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://bigquery.googleapis.com:443 \"GET /bigquery/v2/projects/swast-scratch/datasets?prettyPrint=false HTTP/1.1\" 200 None\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://bigquery.googleapis.com:443 \"POST /bigquery/v2/projects/swast-scratch/datasets?prettyPrint=false HTTP/1.1\" 200 None\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://bigquery.googleapis.com:443 \"POST /bigquery/v2/projects/swast-scratch/datasets?prettyPrint=false HTTP/1.1\" 200 None\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://bigquery.googleapis.com:443 \"POST /bigquery/v2/projects/swast-scratch/datasets/python_bigquery_storage_samples_snippets_20210831210924_9f81bc/tables?prettyPrint=false HTTP/1.1\" 200 None\r\n--------------------------------- Captured log call ---------------------------------\r\nDEBUG    google.auth._default:_default.py:206 Checking None for explicit credentials as part of auth process...\r\nDEBUG    google.auth._default:_default.py:181 Checking Cloud SDK credentials as part of auth process...\r\nDEBUG    google.auth.transport.requests:requests.py:182 Making request: POST https://oauth2.googleapis.com/token\r\nDEBUG    urllib3.connectionpool:connectionpool.py:971 Starting new HTTPS connection (1): oauth2.googleapis.com:443\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://oauth2.googleapis.com:443 \"POST /token HTTP/1.1\" 200 None\r\nDEBUG    google.api_core.bidi:bidi.py:695 Started helper thread Thread-ConsumeBidirectionalStream\r\nDEBUG    google.api_core.bidi:bidi.py:660 Thread-ConsumeBidirectionalStream caught error 400 Invalid stream name. Entity: invalid and will exit. Generally this is due to the RPC itself being cancelled and the error will be surfaced to the calling code.\r\nTraceback (most recent call last):\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/grpc_helpers.py\", line 156, in error_remapped_callable\r\n    return _StreamingResponseIterator(\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/grpc_helpers.py\", line 82, in __init__\r\n    self._stored_first_result = next(self._wrapped)\r\n  File \"/usr/local/Caskroom/miniconda/base/envs/dev-3.9/lib/python3.9/site-packages/grpc/_channel.py\", line 426, in __next__\r\n    return self._next()\r\n  File \"/usr/local/Caskroom/miniconda/base/envs/dev-3.9/lib/python3.9/site-packages/grpc/_channel.py\", line 826, in _next\r\n    raise self\r\ngrpc._channel._MultiThreadedRendezvous: \u003c_MultiThreadedRendezvous of RPC that terminated with:\r\n        status = StatusCode.INVALID_ARGUMENT\r\n        details = \"Invalid stream name. Entity: invalid\"\r\n        debug_error_string = \"{\"created\":\"@1630444168.199761000\",\"description\":\"Error received from peer ipv6:[2607:f8b0:4009:819::200a]:443\",\"file\":\"src/core/lib/surface/call.cc\",\"file_line\":1067,\"grpc_message\":\"Invalid stream name. Entity: invalid\",\"grpc_status\":3}\"\r\n\u003e\r\n\r\nThe above exception was the direct cause of the following exception:\r\n\r\nTraceback (most recent call last):\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/bidi.py\", line 636, in _thread_main\r\n    self._bidi_rpc.open()\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/bidi.py\", line 279, in open\r\n    call = self._start_rpc(iter(request_generator), metadata=self._rpc_metadata)\r\n  File \"/Users/swast/src/github.com/googleapis/python-bigquery-storage/google/cloud/bigquery_storage_v1beta2/services/big_query_write/client.py\", line 522, in append_rows\r\n    response = rpc(requests, retry=retry, timeout=timeout, metadata=metadata,)\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/gapic_v1/method.py\", line 142, in __call__\r\n    return wrapped_func(*args, **kwargs)\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/retry.py\", line 283, in retry_wrapped_func\r\n    return retry_target(\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/retry.py\", line 190, in retry_target\r\n    return target()\r\n  File \"/Users/swast/src/github.com/googleapis/python-api-core/google/api_core/grpc_helpers.py\", line 160, in error_remapped_callable\r\n    raise exceptions.from_grpc_error(exc) from exc\r\ngoogle.api_core.exceptions.InvalidArgument: 400 Invalid stream name. Entity: invalid\r\nINFO     google.api_core.bidi:bidi.py:676 Thread-ConsumeBidirectionalStream exiting\r\n------------------------------- Captured log teardown -------------------------------\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://bigquery.googleapis.com:443 \"DELETE /bigquery/v2/projects/swast-scratch/datasets/python_bigquery_storage_samples_snippets_20210831210924_9f81bc/tables/append_rows_proto2_9096?prettyPrint=false HTTP/1.1\" 200 None\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://bigquery.googleapis.com:443 \"DELETE /bigquery/v2/projects/swast-scratch/datasets/python_bigquery_storage_samples_snippets_20210831210924_5c5f49?deleteContents=true\u0026prettyPrint=false HTTP/1.1\" 200 None\r\nDEBUG    urllib3.connectionpool:connectionpool.py:452 https://bigquery.googleapis.com:443 \"DELETE /bigquery/v2/projects/swast-scratch/datasets/python_bigquery_storage_samples_snippets_20210831210924_9f81bc?deleteContents=true\u0026prettyPrint=false HTTP/1.1\" 200 None\r\n============================== short test summary info ==============================\r\nFAILED samples/snippets/append_rows_proto2_test.py::test_append_rows_proto2[US] - ...\r\n================================= 1 failed in 7.33s =================================\r\n```\r\n","author":{"url":"https://github.com/tswast","@type":"Person","name":"tswast"},"datePublished":"2021-08-31T21:09:59.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":3},"url":"https://github.com/268/python-api-core/issues/268"}