Regression in v2.13.0: ConsumeBidirectionalStream caught unexpected exception and will exit

[dhendry]

tl;dr: Cause and fix

I believe v2.13.0 introduced a regression here: v2.12.0...v2.13.0#diff-769a9f2c04700ff355d77a3292b7ff3cea5ff9bda91d0b31138eeb89fcd4d02dL95-L98

v2.12.0 code:

    def _is_active(self):
        # Note: there is a possibility that this starts *before* the call
        # property is set. So we have to check if self.call is set before
        # seeing if it's active.
        if self.call is not None and not self.call.is_active():
            return False
        else:
            return True

v2.13.0 code:

    def _is_active(self):
        # Note: there is a possibility that this starts *before* the call
        # property is set. So we have to check if self.call is set before
        # seeing if it's active.
        return self.call is not None and self.call.is_active()

Note that in v2.12.0 version _is_active() will return True when self.call is None but in v2.13.0 this will return False.

This seems to very directly violate the behaviour highlighted in the comment:

        # Note: there is a possibility that this starts *before* the call
        # property is set. So we have to check if self.call is set before
        # seeing if it's active.

As such the fix should be as simple as

return self.call is None or self.call.is_active()

Environment details

• OS type and version: Mac 14.1.1
• Python version: python --version: 3.9.18
• pip version: pip --version: 22.3
• google-api-core version: pip show google-api-core

Name: google-api-core
Version: 2.14.0
Summary: Google API client core library
Home-page: https://github.com/googleapis/python-api-core
Author: Google LLC
Author-email: googleapis-packages@google.com
License: Apache 2.0
Location: /Users/dhendry/code/perpetua-pybackend-common/.venv/lib/python3.9/site-packages
Requires: google-auth, googleapis-common-protos, protobuf, requests
Required-by: firebase-admin, google-api-python-client, google-cloud-appengine-logging, google-cloud-bigquery, google-cloud-bigquery-storage, google-cloud-core, google-cloud-firestore, google-cloud-logging, google-cloud-pubsub, google-cloud-secret-manager, google-cloud-storage, google-cloud-tasks

Also relevant here:

Name: google-cloud-bigquery-storage
Version: 2.22.0
Summary: Google Cloud Bigquery Storage API client library
Home-page: https://github.com/googleapis/python-bigquery-storage
Author: Google LLC
Author-email: googleapis-packages@google.com
License: Apache 2.0
Location: /Users/dhendry/code/perpetua-pybackend-common/.venv/lib/python3.9/site-packages
Requires: google-api-core, proto-plus, protobuf
Required-by: perpetua-pybackend-common

Steps to reproduce

After upgrading google-api-core from 2.12.0 to 2.13.0 we are seeing an ConsumeBidirectionalStream caught unexpected exception and will exit. error when using the BigQueryWriteClient. I have narrowed down the issue to this library, NOT the google-cloud-bigquery-storage library

Note: you many need to be on a slow-ish network connection (<= ~50 MBit/sec) to reproduce reliably.

1. Create an AppendRowsStream
2. Call .send() with a large AppendRowsRequest request (close to the maximum, say 9 MB). Very specifically, it seems that the request must be large enough to take more than 1 second to complete (1 second because that is the default value for google.api_core.bidi._RequestQueueGenerator._period)
3. An exception will be raised

Code example

from google.cloud.bigquery_storage_v1 import BigQueryWriteClient
from google.cloud.bigquery_storage_v1.types import ProtoRows, ProtoSchema
from google.cloud.bigquery_storage_v1.types.storage import AppendRowsRequest
from google.cloud.bigquery_storage_v1.writer import AppendRowsStream

s = AppendRowsStream(
    BigQueryWriteClient(),
    AppendRowsRequest(
        write_stream="STREAM NAME",
        proto_rows=AppendRowsRequest.ProtoData(
            writer_schema=ProtoSchema(proto_descriptor="ADD PROTO DESCRIPTOR"),
        ),
    ),
)

s.send(
    AppendRowsRequest(
        proto_rows=AppendRowsRequest.ProtoData(
            rows=ProtoRows(serialized_rows=[SERIALIZED ROWS TOTALLING ~9mb])
        )
    )
)

Stack trace

2023-11-30 17:33:07,189 [DEBUG] Started helper thread Thread-ConsumeBidirectionalStream
2023-11-30 17:33:14,180 [DEBUG] Empty queue and inactive call, exiting request generator.
2023-11-30 17:33:14,266 [DEBUG] waiting for recv.
2023-11-30 17:33:14,266 [DEBUG] recved response.
2023-11-30 17:33:14,266 [DEBUG] waiting for recv.
2023-11-30 17:33:14,268 [INFO] RPC termination has signaled streaming pull manager shutdown.
2023-11-30 17:33:14,268 [ERROR] Thread-ConsumeBidirectionalStream caught unexpected exception  and will exit.
Traceback (most recent call last):
  File ".venv/lib/python3.9/site-packages/google/api_core/bidi.py", line 662, in _thread_main
    response = self._bidi_rpc.recv()
  File ".venv/lib/python3.9/site-packages/google/api_core/bidi.py", line 345, in recv
    return next(self.call)
  File ".venv/lib/python3.9/site-packages/google/api_core/grpc_helpers.py", line 115, in __next__
    return next(self._wrapped)
  File ".venv/lib/python3.9/site-packages/grpc/_channel.py", line 541, in __next__
    return self._next()
  File ".venv/lib/python3.9/site-packages/grpc/_channel.py", line 965, in _next
    raise StopIteration()
StopIteration
2023-11-30 17:33:14,268 [DEBUG] Stopping consumer.
2023-11-30 17:33:14,269 [INFO] Thread-ConsumeBidirectionalStream exiting

[parthea]

Hi @dhendry,

Thanks for reporting this issue and providing a detailed error report! This will be fixed in #562