Pub/sub: amount of streaming pull operations way higher than streaming ack operations

[krelst]

We are using python 2.7.12 with google-cloud-pubsub==0.30.1 and grpcio==1.9.1 and have a very basic subscriber script:

import random
import time

from google.cloud import pubsub

GCLOUD_PROJECT_ID = <our_project_id>
SUBSCRIPTION_NAME = <subscription_name>

def pubsub_message_callback(message):
    time.sleep(random.random())
    message.ack()

def main():
    sub_client = pubsub.SubscriberClient()
    full_subscription_name = 'projects/{project_id}/subscriptions/{subscription}'.format(
        project_id=GCLOUD_PROJECT_ID,
        subscription=SUBSCRIPTION_NAME,
    )
    pubsub_subscription = sub_client.subscribe(
        full_subscription_name,
        flow_control=pubsub.types.FlowControl(max_messages=1)
    )
    pubsub_subscription.open(
        pubsub_message_callback
    )
    while True:
        pass

if __name__ == '__main__':
    main()

When having a large backlog and running this script for 20 minutes, the amount of streaming pull operations is 100 times the amount of streaming ack operations. (see image)

As the the flow control is set to max_messages=1 I would expect that it will always wait to ack the received message, before receiving another one. And even when flow control was not set to max_messages=1 it still looks like unintended behaviour as the total amount of pull operations is still way higher than the amount of ack operations (so what happens with those received/unacked messages?). Is this a bug or is it intended behaviour?

Also asked this as a question on stackoverflow

[theacodes]

@lukesneeringer or @dhermes can you take a look?

[lukesneeringer]

To be honest, I am not really sure what constitutes a "streaming pull message operation" or an "ack operation". I agree it looks odd -- are you having a discernable issue, or is this just that your metrics look strange?

[krelst]

Hi @lukesneeringer, sorry for the late response! This is the pubsub.googleapis.com/subscription/streaming_pull_ack_message_operation_count metric in stackdriver see image. I know the pubsub.googleapis.com/subscription/pull_ack_message_operation_count metric is the amount of messages that are ack'ed by a regular (non-streaming) api call, so I suppose that the other metric is the streaming equivalent. Same goes for the streaming pull operations. I couldn't really determine whether it was just the metrics that were incorrect or that there was indeed were way more messages pulled than ack'ed. Could you think of a way so that I could determine this?

[theacodes]

@kir-titievsky are these metrics supposed to match up in the streaming pull case?

I couldn't really determine whether it was just the metrics that were incorrect or that there was indeed were way more messages pulled than ack'ed. Could you think of a way so that I could determine this?

@krelst I would imagine your "undelivered" count in stackdriver would be fairly high and you would see a large number of duplicates in your workers.

[TonyRoussel]

Similar to @krelst, but the amount of streaming pull operations is 50 times the amount of streaming ack operations.
We're running google-cloud-pubsub==0.30.1 and grpcio==1.9.1 with python 3.6.3, FlowControl.max_messages at 1
Once the topic start to "grow large" (more than 0.5M undelivered) we're seeing around 25% of duplicates in our workers (using Message.message_id), and the "undelivered" count goes down
abnormally slow compared to our previous code with google-cloud-pubsub==0.27.0 and grpcio==1.4.0.

Would it be possible that the consumer pull a large number of messages, causing him to pause and thereby stop sending ack request until it resume (or modify ack deadline) ?

[kir-titievsky]

This may be related to a server-side issue under active investigation. The reason you might be seeing slower ack rates is that you are spending more time processing duplicates. Could you offer any more detail on how different you ack & pull rates are?

…

On Thu, Feb 15, 2018 at 6:58 AM TonyRoussel ***@***.***> wrote: Similar to @krelst <https://github.com/krelst>, but the amount of streaming pull operations is 50 times the amount of streaming ack operations. We're running google-cloud-pubsub==0.30.1 and grpcio==1.9.1 with python 3.6.3, FlowControl.max_messages at 1 Once the topic start to "grow large" (more than 0.5M undelivered) we're seeing around 25% of duplicates in our workers (using Message.message_id), and the "undelivered" count goes down abnormally slow compared to our previous code with google-cloud-pubsub==0.27.0 and grpcio==1.4.0. Would it be possible that the consumer pull a large number of messages, causing him to pause and thereby stop sending ack request until it resume (or modify ack deadline) ? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#4841 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ARrMFuBRm2NpSzNtEWFsP-iVgHs1gDkHks5tVEYtgaJpZM4R60pP> .

-- Kir Titievsky | Product Manager | Google Cloud Pub/Sub <https://cloud.google.com/pubsub/overview>

[TonyRoussel]

I'm not sure about what you need, so here is a graph similar to the one provided @krelst

and from our logs, how many time our callback got trigger, and how many message.ack() we called (the time window is restrained to the period where we tried pubsub 0.30.1 (we had to rollback))

+----+--------+    +----+--------+
| h  | callbk |    | h	 |  ack	  |
+----+--------+    +----+--------+
| 11 | 144471 |    | 11 | 143680 |
| 12 | 126435 |    | 12 | 125071 |
| 13 | 109496 |    | 13 | 110254 |
| 14 | 266072 |    | 14 | 266046 |
| 15 | 126582 |    | 15 | 125779 |
| 16 | 100407 |    | 16 | 101160 |
| 17 | 128546 |    | 17 | 127810 |
+----+--------+    +----+--------+

[theacodes]

I believe the change proposed at #4792 (comment) might completely solve this issue.

Do you mind trying out the version of pubsub on the pubsub-request-load branch? This version adds backpressure when there are too many outstanding requests on the stream, which should make the acks only trail behind the message count by 100+/-100.

You can install this using:

git clone -b pubsub-request-load https://github.com/GoogleCloudPlatform/google-cloud-python.git
pip install --upgrade google-cloud-python/pubsub

If this works or improves the situation at all for you, I'll get it merged and released.

[TonyRoussel]

Ok I'll give it a try soon