fix: consume part of StreamingResponseIterator to support failure while under a retry context

api_core/google/api_core/grpc_helpers.py

@@ -64,6 +64,18 @@ def error_remapped_callable(*args, **kwargs):
 class _StreamingResponseIterator(grpc.Call):
     def __init__(self, wrapped):
         self._wrapped = wrapped
+        self._stored_first_result = False
+
+        # 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._first_result = six.next(self._wrapped)
+            self._stored_first_result = True
+        except StopIteration:
+            # ignore stop iteration at this time. This should be handled outside of retry.
+            pass
+
 
     def __iter__(self):
         """This iterator is also an iterable that returns itself."""
@@ -76,8 +88,12 @@ def next(self):
             protobuf.Message: A single response from the stream.
         """
         try:
-            return six.next(self._wrapped)
+            if self._stored_first_result:
+                self._stored_first_result = False
+                return self._first_result
+           return six.next(self._wrapped)
         except grpc.RpcError as exc:
+            # If the stream has already returned data, we cannot recover here. 
             six.raise_from(exceptions.from_grpc_error(exc), exc)
 
     # Alias needed for Python 2/3 support.