Multiple subscribers with flow control

[osmanbaskaya]

Hello,

I'd like to use two subscribers to process messages from one specific topic. These two subscribers run on different compute engine instances. I also want each subscriber to have only one message to pull from the topic. In other words, if the subscriber has already a message to work on (haven't acked yet), it shouldn't pull another one.

When I set the max_messages in flow-control settings to 1, there is only one subscriber working actively even though multiple subscribers are running from different instances. When I set the max_messages=2, then from time to time the same subscriber took 2 messages and starves the other subscriber. I am pretty sure I can achieve what I described above and missing something. Could you help me to figure this out?

Env details and code snippets are below:

Environment details

1. Pubsub related
2. Mac OS Catalina 10.15.7
   Python 3.8.0
   Name: google-cloud-pubsub
   Version: 2.1.0

Code example

from google.cloud import pubsub_v1
import time


class Consumer:
    def __init__(self, project, subscription_id=None, consumer_fn=None):
        if subscription_id is None:
            raise ValueError("Subscription ID cannot be None.")

        self.project = project
        self.subscriber_client = pubsub_v1.SubscriberClient()

        self.subscription_id = subscription_id

        self.subscription_path = self.subscriber_client.subscription_path(
            self.project, self.subscription_id
        )
        self.consumer_fn = consumer_fn or self.consumer_fn

    @staticmethod
    def consumer_fn(points):
        # do something here
        total = sum(points)
        time.sleep(10)

    def process_message(self, message):
        data = message.data.decode("utf-8")
        points = [int(point) for point in data.split()]
        start, end = points
        self.consumer_fn(range(start, end))
        message.ack()

    def consume(self, max_messages_in_flow=1):
        while True:
            flow_control = pubsub_v1.types.FlowControl(
                max_messages=max_messages_in_flow
            )
            streaming_pull_future = self.subscriber_client.subscribe(
                self.subscription_path,
                callback=self.process_message,
                flow_control=flow_control,
            )
            try:
                logger.info(f"Listening for messages on {self.subscription_path}")
                streaming_pull_future.result()
            except TimeoutError:
                pass
            except Exception as e:
                logger.error(f"Got error while processing the the message {e}")
            except KeyboardInterrupt:
                streaming_pull_future.cancel()
                self.subscriber_client.close()
                logger.info("Consumer will stop gracefully.")
                break

Thanks!

[osmanbaskaya]

Note that I tried this with different processes on the same local machine as well as on separate machines. I observed in both cases that one subscriber process takes all the messages that FlowControl allows: If the max number of messages in the flow set to 2, both messages are pulled by one process. If it's 1, the message is taken by one of the processes and another one is waiting (starving).

[anguillanneuf]

👋 @osmanbaskaya, thanks for your patience!

I'd like to use two subscribers to process messages from one specific topic. These two subscribers run on different compute engine instances. I also want each subscriber to have only one message to pull from the topic. In other words, if the subscriber has already a message to work on (haven't acked yet), it shouldn't pull another one.

It sounds like what you need is synchronous pull with lease management.

both cases that one subscriber process takes all the messages that FlowControl allows

Your observed behavior is expected when using multiple subscribers to pull asynchronously from the same subscription, because the flow control settings (in your code as well as in our samples) are being applied at the client level.

Unless the flow control is being applied at the gRPC level, messages will be buffered in the gRPC stream, where more than the specified number of messages can be present due to the Pub/Sub server sending out messages in packets ≤ 10 MiB in size. We currently don't support server-side flow control.

[osmanbaskaya]

Thanks for the answer, @anguillanneuf!

When I run this code on multiple machine instances, I observed that they can process multiple messages simultaneously even if max_messages set to 1 in flow-control, which makes sense because we have multiple clients and flow control settings are applied at the client level. What surprised me is that I am seeing 15% of messages are already processed and ACKed before but still picked up by one of the other machines second times. Each message is processed in 15 seconds and I have ack deadline of 90 seconds. They all are acked before the deadline but still seeing duplicated processing. I can understand to see duplicate messages time to time (that is why it's suggested having idempotent ops) but is that normal to see this much duplicate processing (15% for 15 seconds avg processing with 90 seconds ack deadline)?

Are you still suggesting synchronous pull with lease management if I run this code on multiple machines with max_messages=1 in the flow control settings? I feel async pull solves my problem with the flow of 1 message on different machines solves my problem but what is your opinion?

[anguillanneuf]

You're welcome!

What surprised me is that I am seeing 15% of messages are already processed and ACKed before but still picked up by one of the other machines second times.

This is possible because buffered messages in one subscriber (again, server-side) can expire or get nack'ed (stream broken) and prompt the server to re-deliver them.

When you repeated the experiment, did you purge the messages in the subscription to start everything fresh?

Flow control settings don't apply to synchronous pull. You may select how many message to pull synchronously from the subscription. Note that the server may return fewer than the specified number of messages.

[osmanbaskaya]

Really helpful, thanks!

This is possible because buffered messages in one subscriber (again, server-side) can expire or get nack'ed (stream broken) and prompt the server to re-deliver them.

Is that the correct scenario: Assume that

• the flow control is 1.
• instance - i
• message - m

i1 takes m1 and start processing it. i2 gets m2 and starts processing it. since it's async m3 comes i1 and sitting in the buffer. processing m1 took enough time to m3's ack deadline is expired in i1's buffer and now i2 got the same message. i1 processed the message m3 and acked but i2 has the same message in its buffer (as you mentioned as packets ≤ 10 MiB in size). Even though I have 15 seconds average processing and 90 seconds ack deadline, there might be messages in the buffer (most likely not immediate but in the middle or end of it) may get expired and picked up by another process. Especially the fact that the size of an individual message is very very small. Is that the right kind of scenario? If this is the right scenario, is there a way to mitigate the percentage (aside from increasing the ack deadline) in async pull? I guess, no, because flow control is done on the client-side.

How can I add the flow control constraints in gRPC level?

When you repeated the experiment, did you purge the messages in the subscription to start everything fresh?

I didn't do the purge. In this case, is it still possible to see a message that is processed hours ago got processed one more time? I guess yes, since 10MB buffer can contain thousands of my messages.

Flow control settings don't apply to synchronous pull. You may select how many messages to pull synchronously from the subscription. Note that the server may return fewer than the specified number of messages.

Do you think the problem I have is more appropriately solved with sync pull? If so, could you tell me a simple problem where async is more appropriate than the sync if it makes sense to compare at all?

Thanks a lot for the help!

[anguillanneuf]

Is that the right kind of scenario?

Yes. Essentially I wanted to get across is this scenario, but I think you get it. To avoid so many duplicates in a multi-subscriber async pull scenario, you don't want messages to get stuck in the buffer space between the server and the client library's user space for too long, and you want to process and ack messages fast enough to flush the buffer relatively quickly.

If all you need is to get one message and process it, and high throughput is not a concern, sync pull can meet your need.

Generally speaking, async pull is used for high throughput scenarios. In your case, 15s processing time and 90s ack deadline sound reasonable to me for async pull, but I would not set flow control max message size to 1. When you set it to a larger number, the client library can automatically modify ack deadlines for the messages in the client library's user space, essentially telling the server not to send out duplicate copies for those messages to the same or different subscriber. (Take note of the resource consumption to modify/extend ack deadlines for the outstanding messages.)

I also want to add that if you are publishing your messages with batch settings, you want to use a batch size of 1 message (instead of using the default batch settings). This can prevent the server from delivering a large batch of messages to one subscriber and starving the rest.

How can I add the flow control constraints in gRPC level?

This would require writing your own client library with server-side changes. I would recommend waiting for future releases of this library.

[osmanbaskaya]

@anguillanneuf, sincerely thank you!

Essentially I wanted to get across is this scenario, but I think you get it. [...]

yeah, I was precisely talking about this situation. I am sorry that I missed this section somehow. I think my messages are even smaller than the scenario you shared, thus, it takes so much time to finish the messages in the client's buffer. Also, my publisher is very very fast, it publishes thousands of messages and stops, which is really a bad case for async pull :)

I think you're right that sync will solve this problem much better. Also, I think you may change your mind on this:

In your case, 15s processing time and 90s ack deadline sound reasonable to me for async pull, [...]

since I didn't mention the nature of publisher, which is like burst-mode (publish a lot quickly and wait for a while)

but I would not set flow control max message size to 1. When you set it to a larger number, the client library can automatically modify ack deadlines for the messages in the client library's user space, essentially telling the server not to send out duplicate copies for those messages to the same or different subscriber. (Take note of the resource consumption to modify/extend ack deadlines for the outstanding messages.)

but doesn't this contradict the scenario? If the ack deadlines of the messages in the user space are modified correctly, why then duplicate messages are sent to the same or different subscribers if the only problem is that messages are in the client's buffer? Also, why is this behavior related to flow control? Even though the max message size is 1, shouldn't the client library update the ack deadlines of the messages in the buffer? Are you saying messages are not getting update because the max message size is 1?

This would require writing your own client library with server-side changes. I would recommend waiting for future releases of this library.

I see. Happy to wait for it, then.

[anguillanneuf]

Actually, how fast your publisher publishes shouldn't have any effect on the subscriber side of things. One big benefit of using a messaging service like Pub/Sub is the decoupling of the message producer and the message consumer.

I think I didn't make it quite clear there are two "buffers" here: buffer in the client vs. buffer in the gRPC stream.

1. For messages in the client library's buffer, their ack deadlines are extended automatically by the client library. When you set max message size to 1 in your flow control settings, no other message is having their ack deadlines extended.
2. For messages in the gRPC stream's buffer, they expire after the preconfigured ack deadline, resulting in possible redeliveries from the server and duplicates.

For example, [m1, m2, m3, m4] are in this the gPRC buffer, and [m5] is the client buffer. The client automatically extends m5's ack deadline, but processing of m5 takes so long that [m1, m2, m3, m4] all expire but they remain in the gRPC stream's buffer. When the server sends them out again to the same subscriber using the same connection, you will see [m1, m2, m3, m4, m1, m2, m3, m4] in the gRPC buffer. If you client library lets in only one message into its own buffer at a time, you aren't taking advantage of the client library's ability to automatically extend message ack deadlines. Does this make sense?

[osmanbaskaya]

Thanks a lot for clarifying that we are talking about two buffers! Yes, it makes a lot of sense now. Thank you so much, @anguillanneuf!