Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

PubSub: Policy.on_exception actually used to make consumer go inactive. #4472

Merged
merged 4 commits into from
Nov 29, 2017
Merged

PubSub: Policy.on_exception actually used to make consumer go inactive. #4472

merged 4 commits into from
Nov 29, 2017

Conversation

dhermes
Copy link
Contributor

@dhermes dhermes commented Nov 28, 2017 •
edited
Loading

Towards #4463.

@dhermes dhermes added the api: pubsub Issues related to the Pub/Sub API. label Nov 28, 2017
@googlebot googlebot added the cla: yes This human has signed the Contributor License Agreement. label Nov 28, 2017
dhermes
dhermes commented Nov 28, 2017
View reviewed changes
pubsub/google/cloud/pubsub_v1/subscriber/futures.py
bool: ``True`` if this subscription is opened with this future,
``False`` otherwise.
bool: :data:`True` if this subscription is opened with this
future, :data:`False` otherwise.

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

pubsub/tests/unit/pubsub_v1/subscriber/test_consumer.py
policy.on_response.side_effect = exc

consumer = _consumer.Consumer(policy=policy)
policy.on_exception.side_effect = OnException(consumer._exiting)

This comment was marked as spam.

Sign in to view

theacodes
theacodes reviewed Nov 28, 2017
View reviewed changes
pubsub/google/cloud/pubsub_v1/subscriber/_consumer.py Outdated
@@ -244,10 +244,9 @@ def _blocking_consume(self):
self.stop_consuming()
except Exception as exc:
try:
self._policy.on_exception(exc)
self.active = self._policy.on_exception(exc)
except:
self.active = False

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

theacodes
theacodes reviewed Nov 28, 2017
View reviewed changes
pubsub/google/cloud/pubsub_v1/subscriber/_consumer.py Outdated
@@ -244,10 +244,9 @@ def _blocking_consume(self):
self.stop_consuming()
except Exception as exc:
try:
self._policy.on_exception(exc)
self.active = self._policy.on_exception(exc)

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

@dhermes
Copy link
Contributor Author

dhermes commented Nov 28, 2017 •
edited
Loading

OK I have updated the gist to be much clearer as to how things were behaving and how they behave now. Most notable:

  • Heartbeats where running=True, done=True in the logs_pypi.txt (happens immediately after thread.Policy.on_exception is called)
  • Lease management (e.g. "Snoozing lease management...") ceases in logs_local.txt immediately after the heartbeats change to running=False, done=True. Just to be sure, we have 40 seconds worth of heartbeats before stopping the main thread and there is still no lease management happening

@dhermes dhermes mentioned this pull request Nov 28, 2017
Closed
@dhermes
Dropping failure case when calling Policy.on_exception().
86ca355
@dhermes
Adding another test for _blocking_consume().
11d5973 
This keeps the thread active after one failure but
stops after a second.
@dhermes
Copy link
Contributor Author

dhermes commented Nov 29, 2017 •
edited
Loading

Remaining point of discussion (got buried below the fold):

... AFAICT thread.Policy.close() (which gets called by BasePolicy.lease()) is what cleans up (it's the only thing that can stop Consumer._blocking_consume since Consumer.stop_consuming is the only thing that calls self._exiting.set()).

@chemelnucfin chemelnucfin changed the title Policy.on_exception actually used to make consumer go inactive. PubSub: Policy.on_exception actually used to make consumer go inactive. Nov 29, 2017
@dhermes
Copy link
Contributor Author

dhermes commented Nov 29, 2017 •
edited
Loading

OK my previous remark seems to be wrong, instead it seems that BasePolicy.maintain_leases() is where the action is.

@dhermes
Copy link
Contributor Author

dhermes commented Nov 29, 2017

I am going to merge this since it "moves the bar" towards correctness even though I know it isn't 100% correct. E.g. I can still trigger "set_exception can only be called once." by using a subscription that doesn't exist.

@dhermes dhermes merged commit 905f49b into googleapis:master Nov 29, 2017
@dhermes dhermes deleted the fix-4463 branch November 29, 2017 17:58
@dhermes dhermes mentioned this pull request Dec 15, 2017
Closed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@theacodes theacodes theacodes left review comments

@lukesneeringer lukesneeringer lukesneeringer approved these changes

Assignees

@lukesneeringer lukesneeringer

@tseaver tseaver

@theacodes theacodes

Labels
api: pubsub Issues related to the Pub/Sub API. cla: yes This human has signed the Contributor License Agreement.
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.
5 participants
@dhermes @lukesneeringer @theacodes @tseaver @googlebot