Subscriber future hangs with large number of messages

[Spawn]

Environment details

• OS type and version: python:3.6-alpine
• Python version: 3.6
• pip version: 20.2.3
• google-cloud-pubsub 1.7.0

Problem

We have an high-loaded worker in kubernetes which listens pub/sub queue. Problem is that when pub/sub queue receive several thousands messages worker is stopping processing without any errors (just freezes). Other workers whose processing time is short are working normally but problem worker sometimes could process one message up to several minutes (http requests, database requests) and after processing several hundreds messages logs are stop printing anything.

So to reproduce issue you should have pub/sub subscription with several thousands messages and create python subscriber which will process one message up to 1 minute and longer sometimes.

Regarding longer processing time I specified timeout=60 for future so if processing time taking so long it should throw TimeoutError where I'm calling future.cancel() and here is the problem - cancel() is waiting for graceful shutdown of consuming messages thread and when I did some debugging for it I found out that sometimes cancel() never ends. For instance on server this kind of freeze took 10 days without any logs/errors/etc.

I'm wondering if there way to force future canceling?

Subscription config

Code example

def run():
    subscriber = pubsub_v1.SubscriberClient()
    subscription_path = subscriber.subscription_path(
        project_id, subscription_name
    )

    future = subscriber.subscribe(subscription_path, callback=some_long_working_code)
    logger.info('Subscriber initialized, listening...')

    with subscriber:
        try:
            future.result(timeout=60)
        except TimeoutError:  # from concurrent.futures import TimeoutError
            logger.warning('Timeout reached, resubscribing...')
            future.cancel()

Thank you for your time!

[pradn]

The fact that cancel() is hanging probably indicates its a threading issue - perhaps your work item is holding up a thread indefinitely and cancel() can't finish waiting for it? The subscriber runs callbacks on a set of threads, and perhaps they are all being used up? You can pass in a scheduler here. The default thread pool uses a max_workers/threads of just 10. Try raising this limit and see if it helps. If the number of long running tasks is greater than the number of threads, you might still run into starvation issues. Perhaps add a metric to see how long each work item is taking and also to check if some never finish? That might be the root cause.

[plamut]

@Spawn
While developing a graceful (blocking) streaming pull shutdown in #292 I realized that streaming_future.cancel() already blocks until all currently executing callbacks are done processing. I fixed that along the way, so that by default the cancel() method will actually return as soon as streaming pull is shut down (e.g. helper threads terminated). The already executing message callbacks will keep running, but cancel() will not block on them anymore, unless explicitly told to do so.

As @pradn mentioned, the hangs observed are most likely caused by some of the currently executing message callbacks blocking indefinitely. BTW, have you identified any potentially problematic spots in the code in the meantime?

[Spawn]

Hey @plamut thank you for update.

BTW, have you identified any potentially problematic spots in the code in the meantime?

Yes, I did some investigations regarding this. By the fact only place which could block execution is database commit() call. I put log messages before and after commit and noticed that both are appearing in logs. Log message after commit is last line of code in current callback and this message ends logs. So it seems runtime freezes elsewhere.

[plamut]

Log message after commit is last line of code in current callback and this message ends logs. So it seems runtime freezes elsewhere.

Noted, thanks.

[acocuzzo]

@Spawn are you still experiencing this issue with the latest version of google-cloud-pubsub?

[acocuzzo]

Closing as no response, please re-open if you are still experiencing