Core: Mitigate busy reopen loop in ResumableBidiRpc consuming 100% CPU (#8193)

* Add bidi._Throttle helper class

* Add optional reopen throttling to ResumableBidiRpc

* Enable Bidi reopen throttling in SPM

* Change bidi._Throttle signature

The commit renames the entry_cap parameter to access_limit, and
changes the type of the time_window argument from float to timedelta.

Author: plamut

google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

@@ -383,6 +383,7 @@ def open(self, callback, on_callback_error):
             start_rpc=self._client.api.streaming_pull,
             initial_request=self._get_initial_request,
             should_recover=self._should_recover,
+            throttle_reopen=True,
         )
         self._rpc.add_done_callback(self._on_rpc_done)
 

tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py

@@ -428,6 +428,7 @@ def test_open(heartbeater, dispatcher, leaser, background_consumer, resumable_bi
         start_rpc=manager._client.api.streaming_pull,
         initial_request=manager._get_initial_request,
         should_recover=manager._should_recover,
+        throttle_reopen=True,
     )
     resumable_bidi_rpc.return_value.add_done_callback.assert_called_once_with(
         manager._on_rpc_done