Assert fails with duplicate AckRequests

[acocuzzo]

Steps to reproduce

1. Ack the same message multiple times on a streaming pull subscription.

This assertion fails when ack duplication occurs:

python-pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

Line 588 in 20d661c

assert len(ack_ids) == len(ack_reqs_dict)

There is no de-duplication of acks in the dispatcher, and because ack_reqs_dict is a Dict[str, AckRequest], the latest AckRequest will replace an earlier request.

The solution is either to change ack_req_dict to allow for multiple AckRequests for the same message, or to de-duplicate acks in the dispatcher.

[simonvanderveldt]

I think we're running into the same issue as this

Acking message 4497448654197332
ERROR:google.cloud.pubsub_v1.subscriber._protocol.helper_threads:Error in queue callback worker:
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/helper_threads.py", line 118, in __call__
    self._callback(items)
  File "/usr/local/lib/python3.7/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/dispatcher.py", line 159, in dispatch_callback
    self.ack(ack_requests)
  File "/usr/local/lib/python3.7/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/dispatcher.py", line 192, in ack
    ack_reqs_dict=ack_reqs_dict,
  File "/usr/local/lib/python3.7/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py", line 588, in send_unary_ack
    assert len(ack_ids) == len(ack_reqs_dict)
AssertionError

I'm not aware that we're acking multiple times. I mean the message is still there so we try again when we rerun the application that acks it, but this also happend the first time we encountered this message as far as I know. It does only happen every now and then for us, not sure how to reproduce.
Bug was introduced by #550, we've reverted back to 2.9.0 which was the last version before that change and that's working fine.