Message Queues with Pulsar
When we set out to build Ubiquity we wanted to ensure that we were cloud agnostic. Why? We suspected several customers of ours would want to run the service on-prem. To keep that possible we decided that every system we built had to run directly on a Kubernetes cluster, rather than depend on discreet cloud services. To this day we can start up ubiquity on k8s on our development laptops. As we architected the data flow we knew we needed a message bus.
I had used a lot SQS on Amazon in the past, and new that several components would benefit from a robust queuing system. But we had to find solution that we could host and scale within our k8s cluster. We eventually settled on Apache Pulsar. If you are not familiar with it, its a highly scalable system with the ability to keep your message queues flowing with low latency even when scaling up to extremely large amounts of data.
We built up our in house expertise on the platform and then got it deployed into our clusters. It was running well, but we kept running into one problem. My message consumer threads were quickly holding large number of messages waiting to be processed, but when I added new threads they weren’t able to grab any of the backlog messages until I had restarted all of the consumer threads. I knew this wasn’t ideal, but sometimes you have to wait til you have time to research a problem.
Recently, I found the configuration option that I had been looking for and solved it! Really the answer was in the documentation the whole time, but it wasn’t until I knew the name of the parameter that I was actually able to find it, helpfully named, Use Pulsar as a message queue.
A bit of bakground that I was able to extract from all this. when you create a consumer using the pulsar client libraries, and call the basic receive() method, you will get a blocking call that yields one message at a time. But under the hood the client library actually fills a queue with a normally sensible default of 1000 messages, and then works it way throug them all. In my use case, though, I needed to do a lot of work per message, and the default value meant that when I added new consumers to a shared queue, I wouldn’t actually get any messages until the old consumers had let go of the ones they were waiting to process.
Get David Smith’s stories in your inbox
Join Medium for free to get updates from this writer.
TL;DR from the documentation:
If you’d like to have tight control over message dispatching across consumers, set the consumers’ receiver queue size very low (potentially even to 0 if necessary). Each consumer has a receiver queue that determines how many messages the consumer attempts to fetch at a time. For example, a receiver queue of 1000 (the default) means that the consumer attempts to process 1000 messages from the topic’s backlog upon connection. Setting the receiver queue to 0 essentially means ensuring that each consumer is only doing one thing at a time.
So I tried the param, tuning the receiver_queue_size down to 0. I pushed this to my local cluster to test, and but found my processor was now hitting 100% when the queue was empty.
Remember I had mentioned that client.receive() is a blocking call? Well that wasn’t quite true. When you set the receiver_queue_size to 0 it seems to stop acting like a blocking call, because picking 0 messages is now allowed. One quick change fixed this, receiver_queue_size=1 . Now the consumer of my queue is processing only the next message it needs, and no longer blocks the other consumers as they scale in:
from pulsar import Client, ConsumerType
SERVICE_URL = "pulsar://localhost:6650"
TOPIC = "persistent://public/default/mq-topic-1"
SUBSCRIPTION = "sub-1"
client = Client(SERVICE_URL)
consumer = client.subscribe(
TOPIC,
SUBSCRIPTION,
# If you'd like to restrict the receiver queue size
receiver_queue_size=1,
consumer_type=ConsumerType.Shared)So what does this look like in my observability layer? Here’s the screenshot:
The middle of each chart shows the faulty autoscaling issue. The autoscaler was adding new nodes, but the new nodes were not taking the load so it would cycle back and forth adding and subtracting nodes to get the load. The result was that it took much longer to process than it should have. On the right of the chart you see the correct behavior: A similar number of messages hit the queue, but the scaling worked smoothly. The new consumers took on the correct load and the backlog dropped without slowing down.
Hopefully having another link to the documentation on my blog here will make it easier for others to find this little corner of Apache’s web properties and save some time in the future.
