Google Pub/Sub Python订阅器运行10分钟后报ALREADY_EXISTS错误

Hey Casper, sorry to hear you're stuck with this frustrating ALREADY_EXISTS error—let's walk through the most likely causes and fixes to get your subscriber back on track.

Possible Causes & Fixes #

1. Residual Subscriber Processes Not Fully Terminated #

Since your subscriber restarts daily, it’s possible the previous instance didn’t shut down cleanly. If a leftover process is still holding onto Pub/Sub resources (like lease locks or client connections), the new instance will throw an ALREADY_EXISTS error when it tries to claim those same resources.

• How to check: On Linux/macOS, run ps aux | grep python to look for duplicate instances of your subscriber script. On Windows, use Task Manager to spot lingering Python processes tied to your app.
• Fix: Update your restart script to force-kill old processes before launching the new one. For example:

  pkill -f "your_subscriber_script.py"
  python your_subscriber_script.py
  

  Also, add signal handling in your Python code to ensure clean shutdown when the process receives a termination signal:

  import signal
  from google.cloud import pubsub_v1
  
  subscriber = pubsub_v1.SubscriberClient()
  subscription_path = subscriber.subscription_path("your-project", "your-subscription")
  
  def shutdown_handler(signal_num, frame):
      print("Shutting down subscriber...")
      subscriber.close()
      exit(0)
  
  # Register signal handlers for SIGTERM and SIGINT
  signal.signal(signal.SIGTERM, shutdown_handler)
  signal.signal(signal.SIGINT, shutdown_handler)
  
  # Rest of your subscriber logic
  

2. Fixed Client ID Causing Resource Conflicts #

If your subscriber uses a hardcoded client_id when initializing the SubscriberClient, each restart will try to create a client instance with the same ID—this can clash with existing resources that haven’t been released yet.

• How to check: Look for code like this in your subscriber:

  subscriber = pubsub_v1.SubscriberClient(client_options={"client_id": "fixed-id"})
  

• Fix: Either remove the fixed client_id (letting the client auto-generate a unique ID each time) or generate a unique ID dynamically on startup (e.g., using the process ID or timestamp):

  import os
  import time
  from google.cloud import pubsub_v1
  
  unique_client_id = f"subscriber-{os.getpid()}-{int(time.time())}"
  subscriber = pubsub_v1.SubscriberClient(client_options={"client_id": unique_client_id})
  

3. Unreleased Pub/Sub Lease Resources #

Pub/Sub subscribers hold leases on subscription partitions to manage message delivery. If your subscriber crashes unexpectedly (instead of shutting down cleanly), these leases might not release immediately (default lease timeout is 60 seconds, but delays can happen). When your new instance starts up, it may try to claim a lease that’s still marked as active.

• How to check: Head to the Google Cloud Console, navigate to your Pub/Sub subscription, and check the "Active subscribers" count. If it’s higher than 1 when only one instance should be running, you have lingering leases.
• Fix: Ensure your subscriber always calls subscriber.close() or cancels the subscription future when exiting—even if an error occurs. Wrap your main logic in a try/finally block to guarantee cleanup:

  future = subscriber.subscribe(subscription_path, callback=your_message_callback)
  try:
      future.result()
  except Exception as e:
      print(f"Subscriber encountered an error: {e}")
  finally:
      future.cancel()
      subscriber.close()
  

4. Unhandled Exceptions Leading to Zombie Processes #

Even if you don’t see messages being processed when the crash happens, an uncaught exception might be causing the process to hang (instead of exiting) and leave resources in use.

• How to check: Add detailed logging to track every step of your subscriber’s lifecycle, including errors. Use Python’s logging module to capture stack traces:

  import logging
  logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
  logger = logging.getLogger(__name__)
  
  def your_message_callback(message):
      try:
          # Process message logic
          message.ack()
      except Exception as e:
          logger.error(f"Error processing message: {e}", exc_info=True)
          message.nack()
  
  # Rest of your subscriber setup
  

• Fix: Catch all exceptions in your callback and main loop, and ensure the subscriber cleans up resources before exiting. This prevents zombie processes from holding onto Pub/Sub resources.

Next Steps #

Start with checking for residual processes—this is the most common culprit in daily restart scenarios. If that doesn’t fix it, move on to verifying your client ID configuration and ensuring proper resource cleanup. Adding detailed logs will also help you pinpoint exactly when and why the error is triggering.

内容的提问来源于stack exchange，提问作者Casper Alant