fix: shutdown event loop if publisher fails to start and set exception on result future

[hannahrogers-google]

When publishing to a topic that does not exist, the publisher fails to start because the call to get topic partitions fails with 'Not found'. This change does the following:

1. Shuts down the event loop for the single publisher on failure
2. Returns a failed future, instead of throwing the 'not found' error

[dpcollins-google]

You can't actually do this. This will result in the managed loop being closed twice when a context manager is used. You shouldn't try to catch this exception- the caller still has to call exit

[hannahrogers-google]

You're right, the problem is that we are not calling exit on SinglePublisherImpl from the MultiplexedPublisherClient.

The MultiplexedPublisherClient attempts to get_or_create a SinglePublisherImpl here:

python-pubsublite/google/cloud/pubsublite/cloudpubsub/internal/multiplexed_publisher_client.py

Line 53 in aa52ef4

publisher = self._multiplexer.get_or_create(

The ClientMultiplexer will attempt to create and start the client here:

python-pubsublite/google/cloud/pubsublite/cloudpubsub/internal/client_multiplexer.py

Line 43 in aa52ef4

self._live_clients[key] = factory()

But, in the Not Found topic case, the client fails on enter because the call to getTopicPartitions fails. So, this raises an exception and the new publisher client is never assigned to live_clients[key].

When exit is eventually called on the ClientMultiplexer, we will never call exit on the 'not found' client because it was never assigned to live_clients[key]:

python-pubsublite/google/cloud/pubsublite/cloudpubsub/internal/client_multiplexer.py

Line 74 in aa52ef4

self._closer(client)

Therefore, we will never run exit on the managed_loop. I will re-work this PR to address the real issue.

[meredithslota]

@dpcollins-google Is this ok to merge? It seems like it but just want to check.