PubSub: dispatch_callback errors after cancel() of a subscription

[udim]

OS: Linux 4.17.0-3rodete2-amd64, Debian 4.17.17-1rodete2
Python 2.7.13
google-cloud-pubsub 0.35.4

Stacktrace 1

ERROR:google.cloud.pubsub_v1.subscriber._protocol.helper_threads:Error in queue callback worker: 'NoneType' object has no attribute 'is_paused'
Traceback (most recent call last):
  File "/usr/local/google/home/ehudm/virtualenvs/beamenv/local/lib/python2.7/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/helper_threads.py", line 112, in __call__
    self._callback(items)
  File "/usr/local/google/home/ehudm/virtualenvs/beamenv/local/lib/python2.7/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/dispatcher.py", line 101, in dispatch_callback
    self.ack(batched_commands.pop(requests.AckRequest))
  File "/usr/local/google/home/ehudm/virtualenvs/beamenv/local/lib/python2.7/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/dispatcher.py", line 124, in ack
    self.drop(items)
  File "/usr/local/google/home/ehudm/virtualenvs/beamenv/local/lib/python2.7/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/dispatcher.py", line 133, in drop
    self._manager.maybe_resume_consumer()
  File "/usr/local/google/home/ehudm/virtualenvs/beamenv/local/lib/python2.7/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py", line 224, in maybe_resume_consumer
    if not self._consumer.is_paused:
AttributeError: 'NoneType' object has no attribute 'is_paused'

Stacktrace 2

ERROR:google.cloud.pubsub_v1.subscriber._protocol.helper_threads:Error in queue callback worker: 'NoneType' object has no attribute 'remove'
Traceback (most recent call last):
  File "/usr/local/google/home/ehudm/virtualenvs/beamenv/local/lib/python2.7/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/helper_threads.py", line 112, in __call__
    self._callback(items)
  File "/usr/local/google/home/ehudm/virtualenvs/beamenv/local/lib/python2.7/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/dispatcher.py", line 101, in dispatch_callback
    self.ack(batched_commands.pop(requests.AckRequest))
  File "/usr/local/google/home/ehudm/virtualenvs/beamenv/local/lib/python2.7/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/dispatcher.py", line 124, in ack
    self.drop(items)
  File "/usr/local/google/home/ehudm/virtualenvs/beamenv/local/lib/python2.7/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/dispatcher.py", line 132, in drop
    self._manager.leaser.remove(items)
AttributeError: 'NoneType' object has no attribute 'remove'

Steps to reproduce

event = threading.Event()
def message_cb(msg):
  msg.ack()
  event.set()
sub_client = pubsub.SubscriberClient()
sub_future = sub_client.subscribe(sub_name, message_cb)
event.wait()
sub_future.cancel()

This subscription should be canceled approximately after the first message. In my test there are two messages sent to the topic and event.set() is called after the second.

I realize that calling cancel() so soon prevents the ack from being sent. Is there a cleaner way to stop the subscription?

[tseaver]

@theacodes I'm not sure how cancellation should work. Can you clairify?

[theacodes]

Cancellation tries to be as "fast" as possible, as waiting for all currently held messages to resolve is basically the halting problem. We should probably update the docs there to mention that.

More serious is the fact that there doesn't seem to be a "clean" shutdown here. I'm going to assign over to @crwilcox to handle, but really the fix should be straightforward (be more defensive when trying to invoke things on the streaming pull manager that could be in the process of shutting down)

[tseaver]

@theacodes, @crwilcox Should this be categorized as a bug, then?

[sduskis]

@plamut, can you please take a look at this?

[plamut]

The first manifestation of the issue (stacktrace 1) is the issue #5751, which has been fixed and released. However, I was able to reproduce the second issue (stacktrace 2) even with the latest PubSub version (0.41.0 at the time of writing) - an exception can occur in the background threads upon shutdown, particularly in the callback dispatcher thread.

It appears that some extra checks need to be added when interacting with the streaming pull manager, because the latter could be in the process of shutting down (as @theacodes already mentioned).

Working on a fix.

Update: Fixed, opened a PR.