pubsub: Expired ACKs for most of the time - wasting resources and time

[marrrcin]

(delete this for feature requests)

Client

PubSub, latest version in Go.
From Gopkg.lock:

[[projects]]
  digest = "1:cab4bd3c2a4dfa2a4a04c72db04fb47dfcffdd55e131fcf0a81ff5379b07d181"
  name = "cloud.google.com/go"
  packages = [
    ".",
    "compute/metadata",
    "iam",
    "internal",
    "internal/optional",
    "internal/trace",
    "internal/version",
    "pubsub",
    "pubsub/apiv1",
    "pubsub/internal/distribution",
    "storage",
  ]
  pruneopts = "UT"
  revision = "264def2dd949cdb8a803bb9f50fa29a67b798a6a"
  version = "v0.46.3"

Describe Your Environment

Docker container based on golang:1.12.6-alpine3.10 running in GKE.

Setup

I have published ~500k messages to PubSub, each message is JSON of size ~ 400-500 bytes. Subscription deadline is 10min.

Now, I launch go pubsub client with the configuration as follows:

ctx := context.Background()
pubsubClient, err := pubsub.NewClient(ctx, settings.ProjectId)
subscriber := pubsubClient.Subscription("not-relevant")
maxWorkers := 4
subscriber.ReceiveSettings.NumGoroutines = maxWorkers
subscriber.ReceiveSettings.MaxOutstandingMessages = maxWorkers
subscriber.ReceiveSettings.MaxOutstandingBytes = maxWorkers * 512

err = subscriber.Receive(ctx, func(ctx context.Context, m *pubsub.Message) {
	defer m.Ack()
         // do something, on average it takes 500ms, never more than 5 seconds
})

Expected Behavior

At most 4 parallel go routines are processing the messages, pulling at most 4 messages at a time from PubSub. OR keeping only 4 * 512B stand-by messages in memory All messages are processed properly with ACKs in-time.

Actual Behavior

I suspected that my callback will be called in at most 4 parallel go routines and it actually is. The problem is that messages are processed in inconsistent rate and after 10min (coincidence? I don't think so! ), metrics show that most of messages that are processed result with expired ACK and the processing is stuck in a loop with old messages, yielding my processing a waste of time and resources.
Moreover, you can see spikes in the StreamingPull Operations. Those spikes happend when my container was deployed to the cluster (with the exact configuration showed above). How do the MaxOutstandingBytes and MaxOutstandingBytes parameters influence processing? The documentation for those parameters leaves much to be desired.

GREEN is expired ACK, ORANGE is valid ACK

This is such a trivial case, why the go library makes it so difficult to implement?

[hongalex]

Thanks for bringing this to our attention! Definitely, the documentation around MaxOutstandingMessages and MaxOutstandingBytes require work. It sounds like what you are experiencing is actually "intended" behavior, albeit confusing. With streaming pull (using the default ReceiveSettings.Synchronous = false), MaxOutstandingMessages tells the client to stop pulling after that many messages (or more) are received. If your publish batch is too high (in this case 500k messages), the client will pull a large number of those all at once, and then stop. Since your numGoroutines is set to 4, a large number of messages will inevitably expire.

If you're processing a large number of messages at once, you will need to use ReceiveSettings.Synchronous = true. This uses a different RPC under the hood that can actually specify how many messages to pull, such that you only pull MaxOutstandingMessages at a time. Can you to use synchronous pull that to see if that fixes it?

[hongalex]

Closing for now. Feel free to comment below if the issue isn't solved after setting ReceiveSettings.Synchronous = true.

[marrrcin]

OK, so I've tested the approach with Synchronous=true and it seems to work. Now I just spawn a few goroutines with Synchronous=true and MaxOutstandingMessages=1 and the behaviour is predictable, ACKs are not timing out. Thanks!

[hongalex]

Glad that worked. You can probably set MaxOutstandingMessages to a higher value without worrying about messages expiring. In synchronous mode, the client will pause once MaxOutstandingMessage are in memory (aka waiting to be processed).

[marrrcin]

OK, will do!

[epsniff]

@hongalex regarding your comment:

the client will pull a large number of those all at once and then stop. 
Since your numGoroutines is set to 4, a large number of messages 
will inevitably expire.

Will the iterator's keep-alive process keep these messages alive until ReceiveSettings.MaxExtension is reached? That is, would an alternative be to increase the MaxExtension time?

[hongalex]

@epsniff Yeah, we actually call that "automatic lease extension", aka managing messages that have been pulled by the client library. While increasing MaxExtension will help alleviate the issue to a certain extent, if you are consistently pulling a large number of messages, it's likely your subscribers might get overwhelmed. In that case, it is recommended to increase the number of subscriber clients processing messages.

Alternatively, if you expect spikes in message publishing, it is also recommended to break up publish batches (by decreasing the publish settings thresholds), so that the messages can be distributed across subscriber clients more efficiently.

[ajroetker]

@hongalex we (@epsniff and I) also seem to see much bigger spikes in this issue when the subscription has a significant backlog, are the the sizes of batches affected by backlogs on the server side?