Subscriber shutdown should optionally wait for the currently executing callbacks to complete

[rslinckx]

Environment details

• OS type and version: osx
• Python version: 3.8
• google-cloud-pubsub version: 1.6.1

Steps to reproduce

When shutting down the subscriber, in-flight messages cannot be acked because the consumer stops first, and the dispatcher checks consumer.is_active before sending any requests. The in-flight tasks are correctly waited for by the shutdown sequence, but the ack() never gets sent

1. Start subscriber
2. Shutdown subscriber by calling cancel() on the returned future
3. In-flight messages are processed and the shutdown wait for them to return
4. Calling ack() on the messages will not be sent to pubsub as the consumer is already shutting down

Code example

import time
from google.cloud import pubsub_v1

project_id = "test"
subscription_id = "sub1"

def on_message(message):
    time.sleep(5)
    message.ack()


with pubsub_v1.SubscriberClient() as subscriber:
    subscription_path = subscriber.subscription_path(project_id, subscription_id)
    streaming_pull_future = subscriber.subscribe(subscription_path, callback=on_message)

    try:
        streaming_pull_future.result()
    except KeyboardInterrupt:
        streaming_pull_future.cancel()

Just after receiving the message, hit ctrl-c, the process will shutdown, wait for the 5 seconds sleep() in the on_message callback, and then exit.

The message.ack() has no effect as it is not sent to pubsub server.

Is this expected ? This means there is no way to gracefully shutdown the subscriber without loosing currently executing callbacks?

I tried to switch the shutdown sequence to first stopping the scheduler, then the consumer in streaming_pull_manager.py:510 and it seems to correctly allow the callbacks to terminate and process the ack(), but i fear this is only a partial workaround because it means new messages may still come in.

Ideally, the consumer would stop receiving new messages, nack() all held messages, and allow for currently executing callbacks to terminate, process the ack/nack/drop requests, then continue shutting down ?

[plamut]

Ideally, the consumer would stop receiving new messages, nack() all held messages, and allow for currently executing callbacks to terminate, process the ack/nack/drop requests, then continue shutting down ?

nack-ing all held messages should be doable, but I'm not so sure about the currently executing callbacks - these are run in a scheduler that uses an internal thread pool. Waiting on all the callbacks to complete is generally not desirable, as it can block the shutdown process for quite awhile.

We could provide an optional flag to the streaming pull future, however, that would allow users to explicitly request to wait for the scheduler shutdown before proceeding.

The shutdown could take a lot of time in this case, but that would be the responsibility of the user to use the new flag correctly and to make sure their callbacks do not block "forever"...

I think we can treat this as a feature request, will bring it up at the next meeting.

[rslinckx]

The use case is when having some kind of task queue, one would expect a graceful shutdown to stop receiving new tasks, finish processing currently executing tasks (up to an option hard limit), and exit gracefully.

A second timeout can be used to forcefully exit the executor to prevent stalled shutdowns

[pradn]

Raphaël, we discussed this idea at our internal meeting. We think it's a valid case, but we'll wait until there's additional demand from customers before adding this. It does increase complexity somewhat.

It's also possible to implement something similar by overriding the scheduler you pass in. It can go into a shutdown mode where newly added callbacks are immediately completed. When the remaining callbacks finish executing, the scheduler exits shutdown mode. This way, the scheduler can set a point beyond which new work is not accepted, allowing graceful shutdown.

We'll reopen this as necessary, based on other users' feedback. Thanks for the feature request and discussion; please feel free to send more feedback!

[plamut]

cc @pradn

Another user independently requested the same (#276).