googleapis / googleapis/google-cloud-python

Background Thread Not Exiting on Pub/Sub Subscriber Shutdown, Causing ValueError: 'Cannot invoke RPC: Channel closed!'

Đang mở
#15,639 2 bình luận 1 reaction 1 người được giao Được giao cho @abbrowne126 Xem trên GitHub
api: pubsub priority: p3 type: bug
Ngôn ngữ chính
Python
Star
5.4k
Fork
1.8k
Merge trung bình
3 ngày 4 giờ
Pull request đã merge (30 ngày)
122

Mô tả

#### Environment details

- `OS type and version`: MacOS 15.1.1
- `Python version`: 3.11.6
- `pip version`: pip 24.0
- `google-cloud-pubsub`: 2.27.1

### Description
I'm encountering an issue where my FastAPI application does not terminate gracefully when shutting down the Pub/Sub subscriber. This results in lingering background threads and the following error:
```scss
ValueError('Cannot invoke RPC: Channel closed!')
```
This behavior aligns with the [GitHub issue googleapis/google-cloud-python#15651](https://github.com/googleapis/google-cloud-python/issues/15651) related to incomplete shutdown handling in the `google-cloud-pubsub` library.

## Steps to Reproduce

1. **Set Up a FastAPI Application with Pub/Sub Subscriber:**
- Initialize a FastAPI application.
- Integrate a Pub/Sub subscriber using the `google-cloud-pubsub` library.
- Implement startup and shutdown logic using an async context manager to manage the subscriber lifecycle.

2. **Run the Application:**
- Start the FastAPI server.
- Ensure the Pub/Sub subscriber is actively listening to messages.

3. **Initiate Shutdown:**
- Terminate the application using `Ctrl+C` or another shutdown signal.
- Observe the shutdown process and logs.

4. **Verify Shutdown Behavior:**
- Notice that the application does not terminate cleanly.
- Check for lingering background threads and error logs indicating issues with shutting down the Pub/Sub subscriber.

## Code Example

Below is a minimalistic example that reproduces the issue. This setup initializes a FastAPI application with a Pub/Sub subscriber and attempts to shut it down gracefully.

```python
# main.py
from fastapi import FastAPI
import uvicorn
from contextlib import asynccontextmanager
import asyncio
from google.cloud import pubsub_v1
import concurrent.futures
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class PubSubSubscriber:
def __init__(self, project_id: str, subscription_name: str):
self.project_id = project_id
self.subscription_name = subscription_name
self.subscriber = pubsub_v1.SubscriberClient()
self.streaming_pull_future = None

def _get_subscription_path(self) -> str:
return self.subscriber.subscription_path(self.project_id, self.subscription_name)

async def _callback(self, message):
"""Asynchronous callback for processing messages."""
try:
data = message.data.decode('utf-8')
logger.info(f"Received message: {data}")
message.ack()
except Exception as e:
logger.error(f"Error processing message: {e}")
message.nack()

async def start(self):
"""Starts the Pub/Sub subscriber."""
subscription_path = self._get_subscription_path()
self.streaming_pull_future = self.subscriber.subscribe(
subscription_path, callback=self._callback_wrapper
)
logger.info("Pub/Sub subscriber started.")
await self._monitor_streaming_future()

async def _callback_wrapper(self, message):
"""Wrapper to ensure the callback is asynchronous."""
await self._callback(message)

async def _monitor_streaming_future(self):
"""Monitors the streaming pull for errors."""
loop = asyncio.get_running_loop()
try:
await loop.run_in_executor(None, self.streaming_pull_future.result)
except Exception as e:
logger.error(f"Streaming pull future threw an exception: {e}")
self.stop()

def stop(self):
"""Stops the Pub/Sub subscriber."""
logger.info("Stopping Pub/Sub subscriber...")
if self.streaming_pull_future:
self.streaming_pull_future.cancel()
try:
self.streaming_pull_future.result(timeout=5.0)
except concurrent.futures.TimeoutError:
logger.warning("Timed out waiting for streaming pull future to terminate.")
except Exception as e:
logger.debug(f"Ignoring exception after cancel: {e}")
self.subscriber.close()
logger.info("Pub/Sub subscriber stopped.")

@asynccontextmanager
async def lifespan(app: FastAPI):
"""Defines startup and shutdown events for the FastAPI application."""
project_id = "your-google-project-id"
subscription_name = "your-subscription-name"

# Startup logic
subscriber = PubSubSubscriber(project_id, subscription_name)
app.state.pubsub_task = asyncio.create_task(subscriber.start())
app.state.pubsub_instance = subscriber
logger.info("Application startup complete.")

yield # Application is running

# Shutdown logic
logger.info("Shutting down Pub/Sub subscriber.")
pubsub_instance = getattr(app.state, "pubsub_instance", None)
pubsub_task = getattr(app.state, "pubsub_task", None)

if pubsub_instance:
try:
pubsub_instance.stop()
except Exception as e:
logger.error(f"Error while stopping Pub/Sub subscriber: {e}")

if pubsub_task:
pubsub_task.cancel()
try:
await pubsub_task
except asyncio.CancelledError:
pass

logger.info("Pub/Sub subscriber shutdown complete.")

app = FastAPI(title="MailFlowAI", lifespan=lifespan)

@app.get("/")
async def read_root():
return {"Hello": "World"}

if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000)
```

---

## Stack Trace

Upon initiating shutdown (e.g., pressing `Ctrl+C`), the following logs and errors are produced:

```
^CINFO: Shutting down
INFO: Waiting for application shutdown.
INFO:main:Shutting down Pub/Sub subscriber.
INFO:main:Stopping Pub/Sub subscriber...
WARNING:google.api_core.bidi:Background thread did not exit.
ERROR:root:Exception in callback >: ValueError('Cannot invoke RPC: Channel closed!')
INFO:main:Pub/Sub subscriber stopped.
INFO:main:Pub/Sub subscriber shutdown complete.
INFO: Application shutdown complete.
INFO: Finished server process [7040]
```

---

- **Additional Context:**
- The issue appears to be linked to the `google-cloud-pubsub` library's handling of shutdown sequences, where the background gRPC threads do not terminate as expected, leading to `ValueError` exceptions and lingering threads that prevent the application from exiting gracefully.

---

**Thank you for your time and assistance!**

---

## Notes for Maintainer

- **Replace Placeholder Values:**
- E.g. "your-google-project-id"` and `"your-subscription-name"` with your actual Google Cloud project ID and Pub/Sub subscription name in the code example.

---

Thanks!

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Đánh giá

Issue này chưa được đánh giá.

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.