PubSub: Release the state lock before calling the publish api

[plamut]

Closes #8036.

Supersedes #7686 (the fix here is essentially the same, but also adds tests and a few minor things).

Summary

This PR makes sure that calls to publisher.publish() are not unnecessarily blocked while the underlying API publish call is in progress - the latter can take a substantial amount of time, especially on network errors when retries kick in.

Description

The fix makes sure that the lock in the _commit() method is held for no longer than necessary. The details on why releasing the lock earlier is fine are listed in the comment on the original PR.

How to test

Steps to reproduce:

• Set the total publish timeout setting to 60,000 milliseconds (in order to not wait for too long),
• Disable the internet connection on your machine,
• Run the test publisher script:

import logging

from google.cloud import pubsub_v1


log_format = (
    "%(levelname)-8s [%(asctime)s] %(threadName)-33s "
    "[%(name)s] [%(filename)s:%(lineno)d][%(funcName)s] %(message)s"
)
logging.basicConfig(
    level=logging.DEBUG,
    format=log_format,
)
logger = logging.getLogger(__name__)

PROJECT_NAME = "some-pubsub-project"
TOPIC_NAME = "my-topic"


def main():
    publisher = pubsub_v1.PublisherClient()
    topic_path = publisher.topic_path(PROJECT_NAME, TOPIC_NAME)

    message_future = publisher.publish(topic_path, b"I'm going to timeout")

    try:
        message_future.exception(timeout=3)
    except Exception as exc:
        logger.info("\x1b[1mException raised:\x1b[0m {}".format(exc))

    logger.info("\x1b[1mAbout to publish the second message\x1b[0m")
    publisher.publish(topic_path, b"I'm going to take quite some time to return")
    logger.info("\x1b[1mReturned from second call to publish()\x1b[0m")


if __name__ == "__main__":
    main()

Actual result (before the fix):
The message "Returned from second call to publish" appears in the logs only after a minute or so after the "About to publish the second message".

Expected result (after the fix):
The second message is logged shortly after the first one. The second call to publisher.publish() is not blocked while publishing the first message is in progress and hitting the automatic retry because of the network error.