feat: exactly-once delivery support

[pradn]

• Add new field to SubscriptionProperties and modify the GAPIC layer with just this one change. This change will come from the auto-GAPIC generated code once the proto change is public.
• Add async ack/modack/nack methods to the Message class. Exactly-once users have to use this to get the status of ack/modack/nack requests. The futures returned by these methods may be completed by the AcknowledgeError exception (with detailed AcknowledgeStatus status code) or a success AcknowledgeStatus code. These futures are of a new type, one used only on the subscriber-side for propagating exactly-once errors. This future mirrors the publishing-side future that already exists. The futures are completed immediately if exactly-once delivery is not enabled, upon successful ack/modack completion, and when errors/exceptions occur.
• Split unary ack/modack methods into two in the subscriber manager. Some repeated code but it's simpler this way.
• Keep track of whether exactly-once is enabled for a subscription by looking at the subscription's SubscriptionProperties, returned in the streaming response.
• Set the minimum ack deadline to 60 secs if exactly-once is known to be turned on. This deadline is used by the leaser.
• Add new min-lease-extension parameter. If the user sets this, it overrides the auto-set param based on whether exactly-once is enabled or not.
• Retry unary ack/modacks if we get errors in the GRPC call's ErrorInfo (no retry delays as of yet since the server will eventually fail them if the ack expires). This is done in the dispatcher, which retries only the acks/modacks that have failed temporarily. Permanent failures are surfaced to the user as future exceptions. These retries use an exponential backoff with a max of 10 min, which is sufficient since the server allows ack/modacks for 10 mins max. These retries are triggered on a new thread to prevent the dispatcher thread from being blocked by retry-wait sleeps.
• Retry lease modacks, both immediately upon receipt of messages and also in the background. If retries fail or if permanent errors arise, we log the errors but don't otherwise disrupt user code.
• Send a new ack deadline via stream_ack_deadline_seconds whenever the exactly_once setting for a stream changes.
• Add tests ensuring 100% coverage
• Usage sample for this feature added at samples: sample for receiving messages with exactly-once delivery enabled #588

[plamut]

This is only a partial feedback, I have not gone through the entire PR yet.

The think we need to be careful about is to not block any worker threads for too long, e.g. when retrying failed requests.

[plamut]

Round two, added some additional comments.

Will review the streaming pull manager tests later.

[plamut]

Went through the rest of the PR, this should be it for the first review round. :)

[plamut]

@pradn I see a lot of comments marked as resolved, but no changes in the code - is there a git push missing? :)

[pradn]

Pushed

[snippet-bot]

No region tags are edited in this PR.

This comment is generated by snippet-bot.
If you find problems with this result, please file an issue at:
https://github.com/googleapis/repo-automation-bots/issues.
To update this comment, add snippet-bot:force-run label or use the checkbox below:

• Refresh this comment

[acocuzzo]

After conversation: https://github.com/googleapis/python-pubsub/pull/550/files#r769774809
is resolved, and all checks pass, LGTM.