Pub/Sub: Send modify_ack_deadline requests in chunks

[mwytock]

Currently leaser is limited in total number of messages (<2000?) which is a problem for use cases with many small messages.

Same issue was solved in nodejs library here: googleapis/nodejs-pubsub#65

[mwytock]

More context: lease tries to ack all messages at the same time which results in the following error

Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py", line 335, in send
    self._send_unary_request(request)
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py", line 322, in _send_unary_request
    ack_deadline_seconds=deadline,
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/_gapic.py", line 40, in <lambda>
    fx = lambda self, *a, **kw: wrapped_fx(self.api, *a, **kw)  # noqa
  File "/usr/local/lib/python2.7/dist-packages/google/cloud/pubsub_v1/gapic/subscriber_client.py", line 821, in modify_ack_deadline
    request, retry=retry, timeout=timeout, metadata=metadata
  File "/usr/local/lib/python2.7/dist-packages/google/api_core/gapic_v1/method.py", line 143, in __call__
    return wrapped_func(*args, **kwargs)
  File "/usr/local/lib/python2.7/dist-packages/google/api_core/retry.py", line 273, in retry_wrapped_func
    on_error=on_error,
  File "/usr/local/lib/python2.7/dist-packages/google/api_core/retry.py", line 182, in retry_target
    return target()
  File "/usr/local/lib/python2.7/dist-packages/google/api_core/timeout.py", line 214, in func_with_timeout
    return func(*args, **kwargs)
  File "/usr/local/lib/python2.7/dist-packages/google/api_core/grpc_helpers.py", line 59, in error_remapped_callable
    six.raise_from(exceptions.from_grpc_error(exc), exc)
  File "/usr/local/lib/python2.7/dist-packages/six.py", line 737, in raise_from
    raise value
InvalidArgument: 400 Request payload size exceeds the limit: 524288 bytes.

[plamut]

@mwytock Thanks for the report!

By default the streaming pull is configured to lease a maximum of 100 messages, or 100 MiB of messages, whichever limit is hit first.
https://github.com/googleapis/google-cloud-python/blob/master/pubsub/google/cloud/pubsub_v1/types.py#L71-L73

Any messages received on top of that are put in a waiting buffer, and the message stream gets paused. Sending 100 or less ACK IDs should work fine, thus I'm wondering if you maybe use custom FlowControl settings? Or send ACK and mofiyAckDeadline requests manually?

Also, are there any other interesting entries in the log referring to pausing/resuming the message stream? Things like the following:

• "Message backlog over load at X.YZ, pausing."
• "Current load is X.YZ, resuming consumer."
• "Did not resume, current load is X.YZ."

Any such info would be useful, thanks!

[mwytock]

Hi Peter, Yes you are right - 100 ack IDs is no problem, this only happens when you modify flow control to allow more simultaneous messages in flight, 2000+. Perhaps this isn't the intended use for the async pull code path; we actually worked around the issue by switching to sync pull and managing the the modify ack request messages ourselves which seems to work a lot better. If that's the case feel free to close with working as intended - but I might suggest raising if some crazy person comes along and tries to set max messages too high in Flow control as we did!

…

On Tue, Aug 27, 2019, 6:39 AM Peter Lamut ***@***.***> wrote: @mwytock <https://github.com/mwytock> Thanks for the report! By default the streaming pull is configured to lease a maximum of 100 messages, or 100 MiB of messages, whichever limit is hit first. https://github.com/googleapis/google-cloud-python/blob/master/pubsub/google/cloud/pubsub_v1/types.py#L71-L73 Any messages received on top of that are put in a waiting buffer, and the message stream gets paused. Sending 100 or less ACK IDs should work fine, thus I'm wondering if you maybe use custom FlowControl settings? Or send ACK and mofiyAckDeadline requests manually? Also, are there any other interesting entries in the log referring to pausing/resuming the message stream? Things like the following: - *"Message backlog over load at X.YZ, pausing."* - *"Current load is X.YZ, resuming consumer."* - *"Did not resume, current load is X.YZ."* Any such info would be useful, thanks! — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#9103?email_source=notifications&email_token=AAEQPVW2NEYIXU7JVC3LWODQGUVBDA5CNFSM4IPUQOKKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOD5HYPKA#issuecomment-525305768>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAEQPVVYIXRSI3P7C2J43RDQGUVBDANCNFSM4IPUQOKA> .

[plamut]

@mwytock Thanks for the quick response.

It seems that the defaults are fine then, but I agree that if a user can customize this to a high number, the resulting error message should be more descriptive, or the error should not occur at all.

Since the max request size limit is imposed by the backend, a quick way around it would indeed be raising an error if the FlowControl.max_messages is set to to high a number.

Alternatively, sending the message(s) in smaller batches could be implemented, similarly to what the Node.js client did.

[mwytock]

Sounds right to me! On a more subjective note, I would add that throwing an error if max_messages is too high is probably the right thing to do. We later discovered other info in the docs (https://cloud.google.com/pubsub/docs/pull "StreamingPull: Dealing with large backlogs of small messages") which convinced us that we should switch away from StreamingPull and to use Pull for our use case.

…

On Tue, Aug 27, 2019 at 7:45 AM Peter Lamut ***@***.***> wrote: @mwytock <https://github.com/mwytock> Thanks for the quick response. It seems that the defaults are fine then, but I agree that if a user can customize this to a high number, the resulting error message should be more descriptive, or the error should not occur at all. Since the max request size limit is imposed by the backend, a quick way around it would indeed be raising an error if the FlowControl.max_messages is set to to high a number. Alternatively, sending the message in smaller batches could be implemented, similarly to what the Node.js client did. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#9103?email_source=notifications&email_token=AAEQPVVWDDIUBNNHJYJ7IC3QGU4ZVA5CNFSM4IPUQOKKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOD5H7SEQ#issuecomment-525334802>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAEQPVRZOYEAGNSCLHOAGJLQGU4ZVANCNFSM4IPUQOKA> .

[plamut]

I will definitely put this up for consideration to the product team. There is a threshold beyond which the max_messages setting is not really sensible anymore for practical purposes, although on the other hand, consistency in client implementations across different languages is also high on the list.

It's good that we now have this issue documented, thus thanks again for reporting it!