feat(pubsub): subscriber-side changes for ordering keys #10201

pradn · 2020-01-24T19:48:53Z

Implements the other half of #9928

…_publish(). Do throw if it is called without enabling message ordering.

…deringkeys2

pubsub/google/cloud/pubsub_v1/subscriber/_protocol/dispatcher.py

kamalaboulhosn · 2020-01-28T20:32:30Z

pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

+
+        There is enough load capacity to send one message per ordering key
+        because each key is the result of an ack or nack. Since the load went
+        down by one message, it's safe to send the user another message for the


Is this true? If so, how do messages for different ordering keys get in? Let's say load capacity allows for 10 messages and I have 10 ordering keys that have built up a queue. Now, a message for an 11th ordering key comes in. What happens?

Similar remark here - what if the load is too high due to the total message size, not count?
This seems to break the assumption in the docstring...

Re: "what happens" - the 11th ordering keys message is normally put into MessagesOnHold (in the streaming pull manager's _on_response() method). Then they just wait there to be released until the load allows for it.

Response to Kamal's question:
When a message is dropped (see the dispatcher's drop() method) as a result of ack, nack, or drop by the user, messages get released in two ways:

dispatcher.drop() calls activate_ordering_keys(), which releases the next message for each ordering key (if they exist). Since at least one message was dropped for each key, it's safe to release a message for each key.

dispatcher.drop() then calls maybe_resume_consumer(), which tells the MessagesOnHold class to release the next messages in the queue (within load parameters). These messages may be ordered or unordered.

In your example, the message with the 11th ordering key would be released in step 2.

Response to Peter's question:
You're right that it's possible that the next ordered message pulled for the same key is too big, and that load would exceed the max. I should document this in the docstring.

I think this is permissible because the alternative would complicate MessagesOnHold even more. To solve this issue, we'd have to keep track of ordering keys that need to be "activated" next. When MessagesOnHold.get() is called, we'd need to check that side list, and also the main queue of held messages.

The reason why I went with this way of storing ordered messages is to prevent a situation where we have to choose between various ordering keys and the general unprocessed/on-hold queue. If we always gave the user ordered messages first, we might be starving the user of unordered messages, and vice versa. Fairness would only be guaranteed by random weighting, which is usually linear or, in a more complicated form, log(n) in the number of weights (keys). We'd run into this same issue with this little side list of "ordering keys that need to be activated".

Can't say if treating FlowControl.max_bytes only as a "guideline" is permissible or not, but yes, let's at least document that possibility here.

Sometimes releasing some of the message a bit too early is IMO still a lesser evil than starving.

kamalaboulhosn · 2020-01-28T20:57:50Z

pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

@@ -80,6 +80,20 @@ def _wrap_callback_errors(callback, on_callback_error, message):
        on_callback_error(exc)


+# TODO
+# check if dropping behavior in Leaser is going to be a problem. (maybe disbale


Is it possible for a message to be dropped that was receipt acked, but not yet sent to the user's callback? If so, then drop could break ordered delivery, yes.

No, the leaser only drops ordered messages that have been 1) sent to the user and 2) have gone past the lease expiry timer.

The lease expiry timer starts only when the message has been sent to the user.

kamalaboulhosn · 2020-01-28T22:32:07Z

pubsub/google/cloud/pubsub_v1/subscriber/_protocol/dispatcher.py

        """
        self._manager.leaser.remove(items)
+        ordering_keys = (k.ordering_key for k in items if k.ordering_key)
+        self._manager.activate_ordering_keys(ordering_keys)


This might be later than it needs to be for delivering the next message for the ordering key. We only need to wait for the user callback to run to completion, we do not need to wait for ack or nack. Maybe in Python it doesn't matter so much because the likelihood of someone doing asynchronous processing outside of the callback is small.

My assumption was that most users would ack their message inside the user callback itself. This would be before the callback finishes, though presumably near the end. I think this scenario is more likely than offloading the message to an async process. We can revisit the question if the async scenario arises in practice. The change itself would be small - modifying _wrap_callback_errors in streaming_pull_manager.py.

pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

plamut

This is the first review pass. I did not check the tests yet, nor have I run the code, as the aim was mostly to get the big picture and verify a few assumptions.

pubsub/google/cloud/pubsub_v1/subscriber/_protocol/leaser.py

pubsub/google/cloud/pubsub_v1/subscriber/_protocol/messages_on_hold.py

plamut · 2020-01-29T15:15:58Z

pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

+
+        There is enough load capacity to send one message per ordering key
+        because each key is the result of an ack or nack. Since the load went
+        down by one message, it's safe to send the user another message for the


Similar remark here - what if the load is too high due to the total message size, not count?
This seems to break the assumption in the docstring...

Re: "what happens" - the 11th ordering keys message is normally put into MessagesOnHold (in the streaming pull manager's _on_response() method). Then they just wait there to be released until the load allows for it.

pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

plamut · 2020-01-31T11:39:21Z

Starting the repo split, please do not merge.

pubsub/google/cloud/pubsub_v1/subscriber/_protocol/messages_on_hold.py

plamut · 2020-02-03T14:02:14Z

pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

-        activating the key.
+        Since the load went down by one message, it's probably safe to send the
+        user another message for the same key. Since the released message may be
+        bigger than the previous one, this may increase the load above the maximum.


Just to double check - it might also happen that we release a message while the current load is still above 1.0. Please confirm that this is fine, too, if it happens due to message size.

An example how this could happen (please correct me if I'm wrong):

FlowControl.max_messages is set to "a lot" and FlowControl.max_bytes is set to 10. The user code is currently processing three messages with sizes 1, 1, and 1, respectively (load == 0.3).

The client then receives several message with sizes 9, 8, and 7. Since the current load allows for it, we dispatch the first message (size 9). The user code is now processing messages of sizes 1, 1, 1, and 9, the load is 1.2, and two messages with sizes 8 and 7 are on hold.

The user code ACKs a small message, that message is dropped from the lease management. The load drops to 1.1.

Since the load went down by one message, we (incorrectly) assume that we now have capacity to release another message, thus we release the message of size 8.

The user code is now processing messages of sizes 1, 1, 9, and 8, and the load actually jumps to 1.9.

In unlucky cases, the issue can compound - the user code ACKs another small message (size 1), which makes the client to release another (bigger) message - size 7 this time. The user code is then processing messages of sizes 1, 9, 8, 7, and the load is 2.5 ...

Yes, you're right that this can happen.

My assumption is that messages are of roughly the same size and that user byte size limits are significantly higher than the messages themselves. The default byte size limit is 100mb and the max user-message size is 10mb, which I believe is rare.

Hopefully, users wont hit upon a pathological case often.

pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

plamut

I still have a sanity check question if we are indeed fine with temporarily not taking FlowControl.max_bytes into account in some circumstances, but otherwise it seems like we're getting there. 👍

Now that we have both the subscriber and the publisher sides, we should consider adding at least one system tests to cover the ordering keys feature.

The rest are a few minor remarks and a merge conflict to resolve.

plamut · 2020-02-03T14:34:42Z

pubsub/tests/unit/pubsub_v1/subscriber/test_messages_on_hold.py

@@ -0,0 +1,274 @@
+# Copyright 2018, Google LLC


New file, should have the current year in the Copyright line (AFAIK).

Suggested change

# Copyright 2018, Google LLC

# Copyright 2020, Google LLC

…t change.

plamut · 2020-02-10T07:22:03Z

Merged in the new repo, closing.

pradn added 17 commits December 6, 2019 17:51

Ordering keys - publish-side

bc5156c

fix

4295df5

Merge branch 'master' into orderingkeys2

bc18d59

Add sequencer tests files, left out accidentally.

01f354b

Get code coverage back up to 💯.

930a441

Remove debug code.

9da7406

Small fixes and more test coverage.

acab6d8

Changes based on Peter's comments.

508eaa4

Ordering keys - publish-side

d722889

merged

65df845

remove errant comment

64c39d8

Fix docstring.

91201cb

Do not throw when an unknown topic-key pair passed into client.resume…

2d5d751

…_publish(). Do throw if it is called without enabling message ordering.

Fix deadlock between publisher client and ordered sequencer.

a301196

Merge branch 'master' of github.com:pradn/google-cloud-python into or…

156f746

…deringkeys2

Minor refactor using tuple destructuring to improve clarity.

d374de1

Reduce thread-related test flakiness.

6e7e7a1

pradn requested review from anguillanneuf and plamut as code owners January 24, 2020 19:48

googlebot added the cla: yes This human has signed the Contributor License Agreement. label Jan 24, 2020

pradn requested a review from kamalaboulhosn January 24, 2020 19:58

Fix lint/coverage.

b3a6546

pradn force-pushed the orderingkeys3 branch from 0d0002e to ffdc7ef Compare January 24, 2020 20:30

Ordering keys - subscriber-side

ae46e6c

pradn force-pushed the orderingkeys3 branch from 13ff3c8 to ae46e6c Compare January 24, 2020 21:12

kamalaboulhosn suggested changes Jan 28, 2020

View reviewed changes

plamut added the api: pubsub Issues related to the Pub/Sub API. label Jan 29, 2020

plamut reviewed Jan 29, 2020

View reviewed changes

pradn added 2 commits January 29, 2020 12:09

Address Kamal's comments.

faaebc3

Address Peter's comments.

1c432c1

plamut added the do not merge Indicates a pull request not ready for merge, due to either quality or timing. label Jan 31, 2020

plamut reviewed Feb 3, 2020

View reviewed changes

Peter's comments: Change copyright year for newly added files. Commen…

d876ef1

…t change.

kamalaboulhosn mentioned this pull request Feb 5, 2020

feat(pubsub): ordering keys googleapis/python-pubsub#26

Merged

plamut closed this Feb 10, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(pubsub): subscriber-side changes for ordering keys #10201

feat(pubsub): subscriber-side changes for ordering keys #10201

pradn commented Jan 24, 2020

kamalaboulhosn Jan 28, 2020

plamut Jan 29, 2020

pradn Jan 29, 2020 •

edited

pradn Jan 30, 2020 •

edited

plamut Feb 3, 2020 •

edited

kamalaboulhosn Jan 28, 2020

pradn Jan 29, 2020

kamalaboulhosn Jan 28, 2020

pradn Jan 29, 2020

plamut left a comment

plamut Jan 29, 2020

plamut commented Jan 31, 2020

plamut Feb 3, 2020

pradn Feb 3, 2020

plamut left a comment

plamut Feb 3, 2020

plamut commented Feb 10, 2020

feat(pubsub): subscriber-side changes for ordering keys #10201

feat(pubsub): subscriber-side changes for ordering keys #10201

Conversation

pradn commented Jan 24, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pradn Jan 29, 2020 • edited

Choose a reason for hiding this comment

pradn Jan 30, 2020 • edited

Choose a reason for hiding this comment

plamut Feb 3, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

plamut left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

plamut commented Jan 31, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

plamut left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

plamut commented Feb 10, 2020

pradn Jan 29, 2020 •

edited

pradn Jan 30, 2020 •

edited

plamut Feb 3, 2020 •

edited