Pubsub: allow registering a callback for batches of messages.

[gcbirzan]

The old API allowed you to operate on batches of messages. Sometimes, this is advantageous as you can aggregate other calls. With the current API, this is non trivial.

So, I think (if the underlying code supports it), adding, either a parameter to open(), or another function open_many() that'll always return a list (there could be a limit in the flow control), even if it's just with one item would be useful.

[theacodes]

/cc @kir-titievsky @lukesneeringer

Curious to your thoughts on this. It would actually be relatively trivial to implement.

[lukesneeringer]

We could just have a batch=True in the open method on the subscription that fired the callback once per message or in the chunk.

I would be less enthusiastic about implementing our own batches (e.g. "call the callback once you get 50 messages but not before"), but sending whatever comes through the stream at a time is wholly reasonable.

[kir-titievsky]

With Luke on the they why do this and how to if we do. The “why not” here is that we now need to explain to users how batching works (why certain messages are delivered in batches but not others). And this is not straightforward. Maybe that’s OK. More interesting is that we may now need to think of other language APIs. What about a separate utility batching class (the one Luke does not want to implement :)? Sort of. This class is kind of like a set (so you can use its add method as a callback). But it knows how to call another callback that processes batches. Maybe even configurable the same way we process publish batching. This could be handy for other APIs.

On Tue, Mar 6, 2018 at 5:52 PM Luke Sneeringer ***@***.***> wrote: We could just have a batch=True in the open method on the subscription that fired the callback once per message or in the chunk. I would be less enthusiastic about implementing our own batches (e.g. "call the callback once you get 50 messages but not before"), but sending whatever comes through the stream at a time is wholly reasonable. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <https://github.com/GoogleCloudPlatform/google-cloud-python/issues/4994#issuecomment-370958201>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ARrMFkx8-hZClEUfji72SXSUDEdBMFhjks5tbxMrgaJpZM4SfcIk> .

-- Kir Titievsky | Product Manager | Google Cloud Pub/Sub

[gcbirzan]

While I'm not 100% sure of how batched messages will come from streaming API, vs the old pull (where messages accumulate as you're processing), a best effort batching should at least alleviate some of the issues, for high rate queues, and might even improve performance.

As far as documenting it, since I'm not familiar with the specifics, I cannot explain how it works, but just mentioning it's not guaranteed to batch messages would be enough for me to understand.

[theacodes]

To @kir-titievsky's point, we need to consider two separate approaches to this:

1. Simple - just remove the "send messages on at a time to the executor" behavior. We get a list of messages on every response from the server. We can just send those to your callback. Kir makes a great point that this is non-deterministic- we have no control over how many items the server sends. Your callback may get anywhere from 1 to 100.
2. Provide a batch utility. This would look more like the publisher batch - we would have a max_messages_per_batch and max_latency_per_batch config. The subscriber would receive messages from the server as normal but instead of sending one message at a time, it would hold on to them until either threshold is reached. This behavior is clearer but there is one quirk: the subscriber will hold on to overflow messages and give them to the next batch.

WDYT?

btw, the "simple" method is easily possible for users to do today with the current interface.

[krelst]

Copying in this issue as I requested the same feature in that issue: googleapis/google-cloud-python#4141

[psalaberria002]

What's the status on this? Batching of messages seems a very common requirement.

[plamut]

@psalaberria002 This is an old feature request, but nevertheless - I am working on this right now and will submit a PR shortly. The change will allow users to optionally provide batch=True argument to subscriber.open(), which will result in callbacks being invoked with batches of received messages, as opposed to a single message per callback invocation.

[sduskis]

I'm going to remove the "triaged for GA" label. The goal of GA will be parity with other GA implementations. Other GA implementations don't currently have this feature.

[plamut]

Since I was almost done with the implementation, I still opened a pull request that adds this feature, but put it on hold for now.

[plamut]

Just for the reference, this has been put on hold for the time being, because a consistency across PubSub clients for other languages is desired. Once this feature is officially accepted, the now-closed PR can be re-opened.