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

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

@@ -50,15 +50,28 @@ def publish(
     ) -> Future:
         if isinstance(topic, str):
             topic = TopicPath.parse(topic)
-        publisher = self._multiplexer.get_or_create(
-            topic, lambda: self._publisher_factory(topic).__enter__()
-        )
+        try:
+            publisher = self._multiplexer.get_or_create(
+                topic, lambda: self._create_and_start_publisher(topic)
+            )
+        except GoogleAPICallError as e:
+            failed = Future()
+            failed.set_exception(e)
+            return failed
         future = publisher.publish(data=data, ordering_key=ordering_key, **attrs)
         future.add_done_callback(
             lambda fut: self._on_future_completion(topic, publisher, fut)
         )
         return future
 
+    def _create_and_start_publisher(self, topic: Union[TopicPath, str]):
+        publisher = self._publisher_factory(topic)
+        try:
+          publisher.__enter__()
+        except GoogleAPICallError:
+          publisher.__exit__(None, None, None)
+          raise
+
     def _on_future_completion(
         self, topic: TopicPath, publisher: SinglePublisher, future: "Future[str]"
     ):

google/cloud/pubsublite/internal/wire/partition_count_watching_publisher.py

@@ -56,8 +56,9 @@ async def __aenter__(self):
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
-        self._partition_count_poller.cancel()
-        await wait_ignore_cancelled(self._partition_count_poller)
+        if hasattr(self, '_partition_count_poller'):
+            self._partition_count_poller.cancel()
+            await wait_ignore_cancelled(self._partition_count_poller)
         await self._watcher.__aexit__(exc_type, exc_val, exc_tb)
         for publisher in self._publishers.values():
             await publisher.__aexit__(exc_type, exc_val, exc_tb)