Repository: googleapis/python-pubsub

Issue:
google.api_core.exceptions.InvalidArgument: 400 Request contains an invalid argument.
Thanks for stopping by to let us know something could be better!

**PLEASE READ**: If you have a support contract with Google, please create an issue in the [support console](https://cloud.google.com/support/) instead of filing on GitHub. This will ensure a timely response.

Please run down the following list and make sure you've tried the usual "quick fixes":

  - Search the issues already opened: https://github.com/googleapis/python-pubsub/issues
  - Search StackOverflow: https://stackoverflow.com/questions/tagged/google-cloud-platform+python

If you are still having issues, please be sure to include as much information as possible:

#### Environment details

  - OS type and version: 2.0.42-debian10
  - Python version: `python --version`: 3.9.1
  - pip version: `pip --version`: 22.0.3
  - `google-cloud-pubsub` version: `pip show google-cloud-pubsub`: 2.12.1
 
#### Code example

```python
# Wrap subscriber in a 'with' block to automatically call close() when done.
with subscriber:
    try:
        # When `timeout` is not set, result() will block indefinitely,
        # unless an exception is encountered first.
        streaming_pull_future.result(timeout=timeout)
    except TimeoutError:
        streaming_pull_future.cancel()  # Trigger the shutdown.
        streaming_pull_future.result()  # Block until the shutdown is complete.
```

#### Stack trace
```
File "/usr/local/lib/python3.9/concurrent/futures/_base.py", line 433, in result
    return self.__get_result()
  File "/usr/local/lib/python3.9/concurrent/futures/_base.py", line 389, in __get_result
    raise self._exception
google.api_core.exceptions.InvalidArgument: 400 Request contains an invalid argument.
```

This is a bit weird, my code is basically very similar to the one i copied from the [GCP docs](https://cloud.google.com/pubsub/docs/pull), and i cannot reproduce this issue locally, that means locally i have run many times, it is all ok, but once i deployed the code to [GCP Dataproc](https://cloud.google.com/dataproc), end of each run i get the above `google.api_core.exceptions.InvalidArgument: 400 Request contains an invalid argument.` error, hitting my head on wall, pls help.


Discussion:

Patch:
diff --git a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py
index f7e44cb..f909c8e 100644
--- a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py
+++ b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py
@@ -927,20 +927,11 @@ class StreamingPullManager(object):
             A request suitable for being the first request on the stream (and not
             suitable for any other purpose).
         """
-        # Any ack IDs that are under lease management need to have their
-        # deadline extended immediately.
-        if self._leaser is not None:
-            # Explicitly copy the list, as it could be modified by another
-            # thread.
-            lease_ids = list(self._leaser.ack_ids)
-        else:
-            lease_ids = []
-
         # Put the request together.
         request = gapic_types.StreamingPullRequest(
-            modify_deadline_ack_ids=list(lease_ids),
-            modify_deadline_seconds=[self.ack_deadline] * len(lease_ids),
             stream_ack_deadline_seconds=stream_ack_deadline_seconds,
+            modify_deadline_ack_ids=[],
+            modify_deadline_seconds=[],
             subscription=self._subscription,
             client_id=self._client_id,
             max_outstanding_messages=(


Test patch:
diff --git a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py
index ab21a15..a8cbfbc 100644
--- a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py
+++ b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py
@@ -1318,8 +1318,8 @@ def test__get_initial_request():
     assert isinstance(initial_request, gapic_types.StreamingPullRequest)
     assert initial_request.subscription == "subscription-name"
     assert initial_request.stream_ack_deadline_seconds == 123
-    assert initial_request.modify_deadline_ack_ids == ["1", "2"]
-    assert initial_request.modify_deadline_seconds == [10, 10]
+    assert initial_request.modify_deadline_ack_ids == []
+    assert initial_request.modify_deadline_seconds == []
 
 
 def test__get_initial_request_wo_leaser():