Repository: googleapis/python-api-core (python)

Issue:
1.17.0 breaking pubsub
Hi all,
Just a heads-up - 1.17.0 appears to be breaking the current pubsub client. Lots of time spent debugging this, hopefully it will help some other poor bastards out there:

 ```
DEBUG:google.api_core.retry:Retrying due to 503 The service was unable to fulfill your request. Please try again. [code=8a75], sleeping 0.2s ...
 ERROR:grpc._channel:Exception iterating requests!
 Traceback (most recent call last):
   File "/usr/local/lib/python3.7/site-packages/grpc/_channel.py", line 195, in consume_request_iterator
     request = next(request_iterator)
 ValueError: generator already executing
 DEBUG:google.api_core.bidi:Thread-ConsumeBidirectionalStream caught error None Exception iterating requests! and will exit. Generally this is due to the RPC itself being cancelled and the error will be surfaced to the calling code.
 Traceback (most recent call last):
   File "/usr/local/lib/python3.7/site-packages/google/api_core/grpc_helpers.py", line 144, in error_remapped_callable
     return _StreamingResponseIterator(result)
   File "/usr/local/lib/python3.7/site-packages/google/api_core/grpc_helpers.py", line 72, in __init__
     self._stored_first_result = six.next(self._wrapped)
   File "/usr/local/lib/python3.7/site-packages/grpc/_channel.py", line 416, in __next__
     return self._next()
   File "/usr/local/lib/python3.7/site-packages/grpc/_channel.py", line 706, in _next
     raise self
 grpc._channel._MultiThreadedRendezvous: <_MultiThreadedRendezvous of RPC that terminated with:
     status = StatusCode.UNKNOWN
     details = "Exception iterating requests!"
     debug_error_string = "None"
 >
 The above exception was the direct cause of the following exception:
 Traceback (most recent call last):
   File "/usr/local/lib/python3.7/site-packages/google/api_core/bidi.py", line 637, in _thread_main
     self._bidi_rpc.open()
   File "/usr/local/lib/python3.7/site-packages/google/api_core/bidi.py", line 280, in open
     call = self._start_rpc(iter(request_generator), metadata=self._rpc_metadata)
   File "/usr/local/lib/python3.7/site-packages/google/cloud/pubsub_v1/gapic/subscriber_client.py", line 1076, in streaming_pull
     requests, retry=retry, timeout=timeout, metadata=metadata
   File "/usr/local/lib/python3.7/site-packages/google/api_core/gapic_v1/method.py", line 143, in __call__
     return wrapped_func(*args, **kwargs)
   File "/usr/local/lib/python3.7/site-packages/google/api_core/retry.py", line 286, in retry_wrapped_func
     on_error=on_error,
   File "/usr/local/lib/python3.7/site-packages/google/api_core/retry.py", line 184, in retry_target
     return target()
   File "/usr/local/lib/python3.7/site-packages/google/api_core/timeout.py", line 214, in func_with_timeout
     return func(*args, **kwargs)
   File "/usr/local/lib/python3.7/site-packages/google/api_core/grpc_helpers.py", line 146, in error_remapped_callable
     six.raise_from(exceptions.from_grpc_error(exc), exc)
   File "<string>", line 3, in raise_from
 google.api_core.exceptions.Unknown: None Exception iterating requests!
 INFO:google.api_core.bidi:Thread-ConsumeBidirectionalStream exiting
 INFO:google.cloud.pubsub_v1.subscriber._protocol.heartbeater:Thread-Heartbeater exiting.
 INFO:google.cloud.pubsub_v1.subscriber._protocol.leaser:Thread-LeaseMaintainer exiting.
```

Pinning to 1.16.0 in requirements.txt fixes the issue.

Pull request description:
feat: allow disabling response stream pre-fetch
Closes #25.

This PR adds the ability to disable automatically pre-fetching the first item of a stream returned by `*-Stream` gRPC callables. This hook will be used in PubSub to fix the [stalled stream issue](https://github.com/googleapis/python-pubsub/issues/93), while also not affecting Firestore, since the default behavior is preserved.

I realize the fix is far from ideal, but it's the least ugly among the approaches I tried, e.g. somehow passing the flag through `ResumableBidiRpc` (it's a messy rabbit hole). On the PubSub side monkeypatching the generated SubscriberClient will be needed, but it's a (relatively) clean one-liner:
```patch
diff --git google/cloud/pubsub_v1/gapic/subscriber_client.py google/cloud/pubsub_v1/gapic/subscriber_client.py
index e98a686..1d6c058 100644
--- google/cloud/pubsub_v1/gapic/subscriber_client.py
+++ google/cloud/pubsub_v1/gapic/subscriber_client.py
@@ -1169,6 +1169,8 @@ class SubscriberClient(object):
                 default_timeout=self._method_configs["StreamingPull"].timeout,
                 client_info=self._client_info,
             )
+            # TODO: explain this monkeypatch!
+            self.transport.streaming_pull._prefetch_first_result_ = False
 
         return self._inner_api_calls["streaming_pull"](
             requests, retry=retry, timeout=timeout, metadata=metadata
```

If/when we merge this, we should also release it, and then we can add `!= 1.17.0` to the `google-api-core` version pin in PubSub.

### PR checklist
- [x] Make sure to open an issue as a [bug/issue](https://github.com/googleapis/python-api-core/issues/new/choose) before writing your code!  That way we can discuss the change, evaluate designs, and agree on the general idea
- [x] Ensure the tests and linter pass
- [x] Code coverage does not decrease (if any source code was changed)
- [x] Appropriate docs were updated (if necessary)


Patch:
diff --git a/google/api_core/grpc_helpers.py b/google/api_core/grpc_helpers.py
index fde6c33..b617ddf 100644
--- a/google/api_core/grpc_helpers.py
+++ b/google/api_core/grpc_helpers.py
@@ -62,14 +62,15 @@ def _wrap_unary_errors(callable_):
 
 
 class _StreamingResponseIterator(grpc.Call):
-    def __init__(self, wrapped):
+    def __init__(self, wrapped, prefetch_first_result=True):
         self._wrapped = wrapped
 
         # This iterator is used in a retry context, and returned outside after init.
         # gRPC will not throw an exception until the stream is consumed, so we need
         # to retrieve the first result, in order to fail, in order to trigger a retry.
         try:
-            self._stored_first_result = six.next(self._wrapped)
+            if prefetch_first_result:
+                self._stored_first_result = six.next(self._wrapped)
         except TypeError:
             # It is possible the wrapped method isn't an iterable (a grpc.Call
             # for instance). If this happens don't store the first result.
@@ -141,7 +142,12 @@ def _wrap_stream_errors(callable_):
     def error_remapped_callable(*args, **kwargs):
         try:
             result = callable_(*args, **kwargs)
-            return _StreamingResponseIterator(result)
+            # Auto-fetching the first result causes PubSub client's streaming pull
+            # to hang when re-opening the stream, thus we need examine the hacky
+            # hidden flag to see if pre-fetching is disabled.
+            # https://github.com/googleapis/python-pubsub/issues/93#issuecomment-630762257
+            prefetch_first = getattr(callable_, "_prefetch_first_result_", True)
+            return _StreamingResponseIterator(result, prefetch_first_result=prefetch_first)
         except grpc.RpcError as exc:
             six.raise_from(exceptions.from_grpc_error(exc), exc)
 


Test patch:
diff --git a/tests/unit/test_grpc_helpers.py b/tests/unit/test_grpc_helpers.py
index 1fec64f..ef84514 100644
--- a/tests/unit/test_grpc_helpers.py
+++ b/tests/unit/test_grpc_helpers.py
@@ -80,6 +80,19 @@ def test_wrap_stream_okay():
     assert responses == expected_responses
 
 
+def test_wrap_stream_prefetch_disabled():
+    responses = [1, 2, 3]
+    iter_responses = iter(responses)
+    callable_ = mock.Mock(spec=["__call__"], return_value=iter_responses)
+    callable_._prefetch_first_result_ = False
+
+    wrapped_callable = grpc_helpers._wrap_stream_errors(callable_)
+    wrapped_callable(1, 2, three="four")
+
+    assert list(iter_responses) == responses  # no items should have been pre-fetched
+    callable_.assert_called_once_with(1, 2, three="four")
+
+
 def test_wrap_stream_iterable_iterface():
     response_iter = mock.create_autospec(grpc.Call, instance=True)
     callable_ = mock.Mock(spec=["__call__"], return_value=response_iter)