Understanding message ordering in Google PubSub
Message Ordering is a guarantee that is a very sought after feature from any message broker, queue or eventing system. If a message broker can guarantee delivery in the same order as they were published, the downstream applications can rely on this guarantee to build their features on top of it.
However, achieving a total message ordering in a distributed system is a hard problem, so each message broker provides a different slightly weaker guarantee of ordering.
For example, Kafka guarantees message ordering per partition, but doesn’t guarantee ordering across different partitions. Similarly Google PubSub offers ordering guarantees across same ordering key and same region and not across different ordering keys and different regions.
In this post we’ll understand how the message ordering works in PubSub across different scenarios.
Lets get started.
Prerequisites
Before we begin let us setup a few things which we’ll need during our testing.
- A PubSub topic and subscription with ordering enabled. Note ordering is a subscription level attribute.
- A producer application which will attach ordering key to each message.
- A consumer application to simulate various scenarios and see how the message ordering guarantee holds across these scenarios.
Creating Topic and Subscription
Let us create a topic first.
gcloud pubsub topics create ordering-demo-topicNext create a two subscriptions for this topic one with message ordering enabled and another without it.
gcloud pubsub subscriptions create ordering-demo-subscription --topic projects/<YOUR-PROJECT-ID>/topics/ordering-demo-topic --enable-message-ordering
gcloud pubsub subscriptions create unordered-demo-subscription --topic projects/<YOUR-PROJECT-ID>/topics/ordering-demo-topic Verify from console that both of the resources are created with desired properties.
We’ll use gcloud cli to publish messages to our topic and have a consumer written in Java running on our local machine. We’ll test out various conditions and see where the ordering guarantees stands.
We’ll use following script to keep publishing message to our topic while we run our various tests. This will publish 10 messages with content from 1 to 10 and same ordering key john .
for i in {1..10}; do gcloud pubsub topics publish projects/<YOUR-PROJECT-ID>/topics/ordering-demo-topic --message=$i --ordering-key=john; doneAlso I’m using Spring Boot to create my consumer and the setup looks like this.
@Configuration
@ComponentScan("xyz.sushil.pubsubordering.configurations")
public class PubSubConfig {
@Value("${spring.cloud.gcp.pubsub.project-id}")
public String projectId;
@Bean
public PubSubTemplate pubSubTemplate(PubSubPublisherTemplate pubSubPublisherTemplate,
PubSubSubscriberTemplate pubSubSubscriberTemplate) {
return new PubSubTemplate(pubSubPublisherTemplate, pubSubSubscriberTemplate);
}
@Bean
public PubSubPublisherTemplate pubSubPublisherTemplate() {
var factory = new DefaultPublisherFactory(() -> projectId);
// without setting this you wont be able to publish to the topic.
factory.setEnableMessageOrdering(true);
return new PubSubPublisherTemplate(factory);
}
@Bean
public PubSubSubscriberTemplate pubSubSubscriberTemplate(){
return new PubSubSubscriberTemplate(new DefaultSubscriberFactory(() -> projectId));
}
}@Slf4j
@Component
public class OrderedConsumer implements Runnable{
@Autowired
private PubSubTemplate pubSubTemplate;
@Value("${pubsub.subscription}")
private String subscription;
Map<String, List<String>> messageOrder = new HashMap<>();
@Override
public void run() {
log.info("Staring consumer");
Subscriber subscriber = pubSubTemplate.subscribe(subscription, (message) -> {
/*
DO SOMETHING WITH YOUR MESSAGE
*/
});
subscriber.startAsync();
}
}Let the testing begin
Reading Unordered Subscription
First we’ll read from an unordered subscription. As expected the messages can be delivered out of order. You may get the messages in order but its not guaranteed. If you run a consumer multiple times you’ll see messages getting delivered out of order.
private Consumer<BasicAcknowledgeablePubsubMessage> unorderedDelivery = new Consumer<BasicAcknowledgeablePubsubMessage>() {
@Override
public void accept(BasicAcknowledgeablePubsubMessage message) {
String orderingId = message.getPubsubMessage().getOrderingKey();
String messageStr = message.getPubsubMessage().getData().toStringUtf8();
log.info("ACK-ing message with id {}, Key : {}, Value : {}", message.getPubsubMessage().getMessageId(), orderingId, messageStr);
message.ack();
}
};Once you start the consumer you’ll see messages starts flowing in.
ACK-ing message with id 7890438094365211, Key : john, Value : 1
ACK-ing message with id 7890450513434510, Key : john, Value : 6
ACK-ing message with id 7890399903809843, Key : john, Value : 4
ACK-ing message with id 7890450043510575, Key : john, Value : 8
ACK-ing message with id 7890422600707415, Key : john, Value : 5
ACK-ing message with id 7890459988102076, Key : john, Value : 3
ACK-ing message with id 7890437354343453, Key : john, Value : 7
ACK-ing message with id 7890449925095907, Key : john, Value : 10
ACK-ing message with id 8106031466911181, Key : john, Value : 9
ACK-ing message with id 7890423148514558, Key : john, Value : 2You can see there is no order in the delivery of message.
Reading from ordered subscription but there are no messages when you start the consumer
In this scenario we’ll start the consumer first and then publish the messages. In this case if you ACK the messages PubSub will deliver all the messages in order.
private Consumer<BasicAcknowledgeablePubsubMessage> orderedDeliveryOrderedAck = new Consumer<BasicAcknowledgeablePubsubMessage>() {
@Override
public void accept(BasicAcknowledgeablePubsubMessage message) {
String orderingId = message.getPubsubMessage().getOrderingKey();
String messageStr = message.getPubsubMessage().getData().toStringUtf8();
log.info("ACK-ing message with id {}, Key : {}, Value : {}", message.getPubsubMessage().getMessageId(), orderingId, messageStr);
message.ack();
}
};As expected, in this case you’ll see that messages are being delivered in order.
ACK-ing message with id 7890449377495427, Key : john, Value : 1
ACK-ing message with id 7890474587005750, Key : john, Value : 2
ACK-ing message with id 7890473939298025, Key : john, Value : 3
ACK-ing message with id 7890483998759232, Key : john, Value : 4
ACK-ing message with id 6846110610467839, Key : john, Value : 5
ACK-ing message with id 7890473876964595, Key : john, Value : 6
ACK-ing message with id 7890500293796919, Key : john, Value : 7
ACK-ing message with id 7890474375205828, Key : john, Value : 8
ACK-ing message with id 7890498651097446, Key : john, Value : 9
ACK-ing message with id 7890473485180211, Key : john, Value : 10Now let us try same scenario but with ACK-ing only first 3 messages.
private Consumer<BasicAcknowledgeablePubsubMessage> orderedDeliveryFirst3Ack = new Consumer<BasicAcknowledgeablePubsubMessage>() {
@Override
public void accept(BasicAcknowledgeablePubsubMessage message) {
String orderingId = message.getPubsubMessage().getOrderingKey();
String messageStr = message.getPubsubMessage().getData().toStringUtf8();
if(Integer.parseInt(messageStr) <= 3) {
log.info("ACK-ing message with id {}, Key : {}, Value : {}", message.getPubsubMessage().getMessageId(),
orderingId, messageStr);
message.ack();
} else {
log.info("Got a message with id {} , Key : {}, Value : {}", message.getPubsubMessage().getMessageId(),
orderingId, messageStr);
}
}
};You’ll see that PubSub will keep re-delivering the 4th message again and again until you ack it. It won’t move to next message for a particular key.
ACK-ing message with id 7890473415523129, Key : john, Value : 1
ACK-ing message with id 7890462031353832, Key : john, Value : 2
ACK-ing message with id 6846109807642545, Key : john, Value : 3
Got a message with id 7890511988942602 , Key : john, Value : 4
Got a message with id 7890511988942602 , Key : john, Value : 4
Got a message with id 7890511988942602 , Key : john, Value : 4
Got a message with id 7890511988942602 , Key : john, Value : 4
Got a message with id 7890511988942602 , Key : john, Value : 4Reading from ordered subscription but there are already messages to consume
If there are already messages in the ordered subscription to be read and you start the consumer you’ll see interesting behaviour.
Now assume you were to not ack any message while reading, then the consumer will be stuck on first message.
private final Consumer<BasicAcknowledgeablePubsubMessage> orderedDeliveryNoAck =
new Consumer<BasicAcknowledgeablePubsubMessage>() {
@Override
public void accept(BasicAcknowledgeablePubsubMessage message) {
String orderingId = message.getPubsubMessage().getOrderingKey();
String messageStr = message.getPubsubMessage().getData().toStringUtf8();
log.info("Got a message with id {} , Key : {}, Value : {}", message.getPubsubMessage().getMessageId(),
orderingId, messageStr);
}
};You’ll see in most cases your consumer will be stuck on first message.
Got a message with id 7890525630116112 , Key : john, Value : 1
Got a message with id 7890525630116112 , Key : john, Value : 1
Got a message with id 7890525630116112 , Key : john, Value : 1
Got a message with id 7890525630116112 , Key : john, Value : 1
Got a message with id 7890525630116112 , Key : john, Value : 1
Got a message with id 7890525630116112 , Key : john, Value : 1Next case is an interesting one, let us assume you would ack first few messages. What I have seen is that if you acknowledge first few messages, PubSub will optimistically deliver more messages to you without acknowledging previous messages. In such cases, you can ack some later messages and then ack the interim ones.
In order to fulfil the ordered delivery guarantee PubSub will re-deliver messages that you have acknowledged out of order.
Let us see an example.
private final Consumer<BasicAcknowledgeablePubsubMessage> orderedDeliveryUnorderedAck =
new Consumer<BasicAcknowledgeablePubsubMessage>() {
final AtomicInteger count = new AtomicInteger(0);
final AtomicInteger below3Acks = new AtomicInteger(0);
final AtomicInteger message8Ack = new AtomicInteger(0);
@Override
public void accept(BasicAcknowledgeablePubsubMessage message) {
String orderingId = message.getPubsubMessage().getOrderingKey();
String messageStr = message.getPubsubMessage().getData().toStringUtf8();
log.info("Got a message with id {} , Key : {}, Value : {}", message.getPubsubMessage().getMessageId(),
orderingId, messageStr);
int i = count.incrementAndGet();
if (below3Acks.get() < 3 && Integer.parseInt(messageStr) <= 3) {
log.info("ACK-ing message with id {}, Key : {}, Value : {}", message.getPubsubMessage().getMessageId(),
orderingId, messageStr);
message.ack();
below3Acks.incrementAndGet();
} else if (message8Ack.get() == 0 && Integer.parseInt(messageStr) == 8) {
log.info("ACK-ing message with id {}, Key : {}, Value : {}", message.getPubsubMessage().getMessageId(),
orderingId, messageStr);
message.ack();
message8Ack.incrementAndGet();
} else if (below3Acks.get() == 3 && message8Ack.get() == 1) {
log.info("ACK-ing message with id {}, Key : {}, Value : {}", message.getPubsubMessage().getMessageId(),
orderingId, messageStr);
message.ack();
}
}
};In the above consumer, we ack the first 3 messages (1,2,3), and then only acknowledge the 8th message onwards out of order if they are optimistically delivered. Only after these conditions are met the consumer will ack messages from 4 onwards. In this case you’ll see even though you have acknowledged some messages out of order, they will be redelivered when you start acknowledging message from 4th.
Since PubSub guarantees at-least once delivery this semantic of re-delivering the previously acknowledged messages is within the guaranteed scope.
ACK-ing message with id 7890764139912289, Key : john, Value : 1
ACK-ing message with id 7890738123739016, Key : john, Value : 2
ACK-ing message with id 7890764320170883, Key : john, Value : 3
Got a message with id 7890749767954460 , Key : john, Value : 4
Got a message with id 7890769921519689 , Key : john, Value : 5
Got a message with id 7890763069044883 , Key : john, Value : 6
Got a message with id 7890762964904641 , Key : john, Value : 7
ACK-ing message with id 7890739016587500, Key : john, Value : 8
ACK-ing message with id 7890732995748874, Key : john, Value : 9
ACK-ing message with id 7890748516302371, Key : john, Value : 10
ACK-ing message with id 7890749767954460, Key : john, Value : 4
ACK-ing message with id 7890769921519689, Key : john, Value : 5
ACK-ing message with id 7890763069044883, Key : john, Value : 6
ACK-ing message with id 7890762964904641, Key : john, Value : 7
ACK-ing message with id 7890739016587500, Key : john, Value : 8
ACK-ing message with id 7890732995748874, Key : john, Value : 9
ACK-ing message with id 7890748516302371, Key : john, Value : 10Does PubSub ordering guarantee always holds ?
We saw quite a few examples in this post, and saw that in all of those cases the ordered delivery guarantee is held by PubSub. Even in cases when you acknowledge some of the messages out of order they are redelivered to fulfil the ordered at-least once delivery.
If you read the PubSub docs on message ordering, you’ll see they have mentioned this behaviour. A small excerpt from the doc is :
When the Pub/Sub service redelivers a message with an ordering key, the Pub/Sub service also redelivers every subsequent message with the same ordering key, including acknowledged messages.
In continuation to this, the doc further mentions:
If both message ordering and a dead-letter topic are enabled on a subscription, this may not be true, as Pub/Sub forwards messages to dead-letter topics on a best-effort basis.
This means if you have dead-lettering enabled, in some edge cases your ordering might be affected when a message is retried N times before being moved to dead letter. You can try this example yourself by creating a subscription with dead letter topic and not acknowledging messages.
So is PubSub ordering a panacea to all the ordering woes ?
Well for most cases, the ordering works as expected, however docs do mention a case where if you don’t have dead letter topic attached to your subscription and the consumer fails to acknowledge the message and die, the other consumers might get stuck.
The inability to acknowledge a message might hold up the delivery of messages for other ordering keys. This issue is possible when servers restart unexpectedly or there are changes in the set of servers used due to traffic changes. To preserve order across such events, all messages published to the old server must be acknowledged before messages from the new server are delivered, even if they are for different ordering keys. If you cannot ensure timely acknowledgment of all messages, consider attaching a dead-letter topic to the subscription. Order of messages might not be preserved when they are written to the dead-letter topic.
I couldn’t re-create this scenario for myself, but I’ll keep trying and will update the post if I succeed.
If you find any bug in my code snippets or have any questions, feel free to drop a comment below.
Till then Happy Coding :)
