Pub/Sub subscriber flow control stops ack requests

[kir-titievsky]

Pub/Sub client library stops sending Ack or modifyAckDeadline requests entirely when flow control settings are defined. See code below: when ran as it is against a subscription with a large backlog, it shows that the client thinks it's processing ~700 messages/second, while there is no outgoing traffic from the machine and the monitoring metrics show no ack | modAckDeadline traffic.

If we comment out the flow control argument, the overall processing rate is the same, acks and modifyAckDeadlines are regularly and successfully received by the server.

Expected behavior: flow control settings do not prevent the client library from acknowledging the message.

1. OS type and version
   Mac Os Sierra
2. Python version and virtual environment information python --version
   2.7
3. google-cloud-python version pip show google-cloud, pip show google-<service> or pip freeze
   pip show google-cloud-pubsub
   Name: google-cloud-pubsub
   Version: 0.28.4
4. Stacktrace if available
5. Steps to reproduce

• Run without flow control setting against a subscription with a large (10K messages) backlog
• Run without flow control setting against a subscription with a large (10K messages) backlog

6. Code example

import datetime
import sys
import time
import threading

from google.cloud import pubsub_v1 as pubsub

subscription_name = "projects/%s/subscriptions/%s"%(sys.argv[1], sys.argv[2])
sleep_time_ms = 0
try:
    sleep_time_ms = int(sys.argv[3])
except Exception:
    print "Could not parse custom sleep time."

print "Using sleep time %g ms"%sleep_time_ms
start_time = -1
messages_processed = 0.0
thread_lock = threading.Lock()


def callback(message):
    global start_time, thread_lock,messages_processed

    if start_time < 0 :
        with thread_lock: start_time = time.time()

    time.sleep(float(sleep_time_ms)/1000)
    message.ack()
    with thread_lock:
        messages_processed += 1


subscription = pubsub.SubscriberClient().subscribe(subscription_name, flow_control = pubsub.types.FlowControl(max_messages=100)).open(callback=callback)

while True:
    time.sleep(5)
    print "Acking messages at rate %.5f / second"%( messages_processed / (time.time() - start_time))

Using GitHub flavored markdown can help make your request clearer.
See: https://guides.github.com/features/mastering-markdown/

[lruslan]

Also affected by this bug in the latest release google-cloud-pubsub==0.29.1
@lukesneeringer is thee any other way to limit number of messages pulled by subscriber?
I have 2000 events per second with 300KB of average message size and this just kills my workers with OOM, thanks

[dhermes]

@lruslan This is on my list of things to resolve for the Pub / Sub client. I just pushed 0.29.1 after addressing the primary issue in #4234, so other fixes are still TBD.

[MaxDesiatov]

reproducible for me with google-cloud-pubsub==0.29.2

[dhermes]

@explicitcall I am luckily close to "progress" for fixing how the flow control pauses and resumes the consumer thread.

[dhermes]

@kir-titievsky I'm fairly certain 0.29.4 has resolved this, would you mind helping verify?

[dhermes]

OK, see https://github.com/dhermes/google-cloud-pubsub-performance/tree/master/issue-4238, I am 100% sure this issue is fixed.

@kir-titievsky I'd still love to chat about other ways I can "kick the tires" to make sure this implementation is hardened.