Guaranteeing reply delivery to the same instance in KeyShared subscription without per-instance topics

[StormLord07]

Hello,

I’m trying to implement a request/response pattern between two programs using Pulsar, but I want to avoid:

• Creating a large number of per-instance topics or subscriptions
• Duplicating messages to all instances

Idea

• Program A may have multiple instances.

  • All share the same KeyShared subscription on a topic.
  • Each instance has its own logical “key” (e.g., key_1, key_2, …) and consumes only messages routed to it by message key hash range.
  • When sending a request, it uses a producer and sets a message property reply_id = key_1 to indicate where to send the response.

• Program B also has multiple instances.

  • Consumes from the same request topic.
  • When replying, it publishes a message with key = key_1.

---

Problem

If Program B publishes with key = key_1, can we guarantee that the message will be routed back to the same instance of Program A that originally sent the request?
Assumptions:

• Using KeyShared subscription mode
• Key stays consistent (key_1 maps to the same consumer)
• Auto-split or sticky hash range policy is used

---

Reasoning

• ConsumerExclusive: Apps starting up are not guaranteed to have freed the previous subscription (consumer name still considered active by the broker). This forces us to recreate new subscriptions each time, which would lead to topic/subscription bloat over time. replication tends to slow things down a lot. since sometimes the long disconnected exclusive consumers tend to stick and replicate.
• If we try to use ConsumerShared, we risk the response being routed to an instance of the program that is in the process of shutting down, resulting in lost or undeliverable replies.
• ConsumerFailover seems like a good idea, but I am a bit fuzzy about its working. And it seems its even worse, since the message will most certanly be lost to the old instance since it's a master at the moment.

And the other problem is we can't exactly know which instance is being restarted, but that i may be wrong, I do not exactly know how they are being restarted.

---

Edit:

I’ve implemented STICKY consumers with a very narrow range. For example, I calculate:

int32_t range = (murmur3_32(<key>) % 65536) + 1;

Then I assign the range as {range, range}, ensuring that messages with the same key always land on the same consumer.

However, as the number of program instances increases, we run into the birthday paradox: the probability of collisions grows rapidly. When that happens, the resulting bugs will be extremely difficult to diagnose.

TLDR

The question is whether the KeyShared hash-range mapping is stable enough between producer and consumer keys to make this safe, or whether we still need per-instance topics/subscriptions for strict targeting. Or is there a better/suggested way?

[lhotari]

I’m trying to implement a request/response pattern between two programs using Pulsar, but I want to avoid:

In the first place, I would avoid request-response over Pulsar since it could lead to a very tightly coupled solution between the applications. For tightly coupled applications, it might be better to use plain HTTP or gRPC interfaces to reduce unnecessary complexity. What is the exact requirement to use request-response over Pulsar?

For a RPC use case over Pulsar, there's an example at https://github.com/streamnative/pulsar-recipes/tree/main/rpc .
That has a different type of design and relies on consuming rpc client topics using a regex topic pattern.

Creating a large number of per-instance topics or subscriptions

Why do you want to avoid this? Pulsar supports a large number of topic and subscriptions. Pulsar features such as namespace policies for TTL and automatically deleting inactive topics could address the possible concerns around maintenance.

Duplicating messages to all instances

What is the reason for avoiding duplication? Are the messages large? Just thinking that data pipelines usually do a lot of duplication with intermediate topics where messages could be transformed and/or routed. This simplifies solutions in many cases.

Auto-split or sticky hash range policy is used

Auto-split wouldn't provide a guarantee that you are looking for. You'd need to use sticky with Key_shared. An alternative for key_shared subscriptions could be partitioned topics using exclusive or failover subscriptions (there's answers in #24044 about that).

[StormLord07]

Thanks for the reply. We ended up with a split approach:

• gRPC for true request/response.
• Pulsar for everything else, like topics w/o replies, stuff that requires reading from multiple services, etc.

What is the exact requirement to use request-response over Pulsar?

A bit ashamed to tell this but we just wanted to reduce the amount of tech used in a project, which bit us hard.

For an RPC use case over Pulsar, there's an example at https://github.com/streamnative/pulsar-recipes/tree/main/rpc.
You'd need to use sticky with Key_shared.

We built something similar, though w/o the regex, didn't know it was a thing. It could have worked. But in our way the same issue about collisions of murmur hash persisted, we used the sticky range and key_shared sub

Why do you want to avoid this? Pulsar supports a large number of topic and subscriptions.

Mainly operational risk and startup cost:

• Lifecycle/cleanup worst case scenario: service A is restarted daily while service B might restart twice a year. Per-instance topics/subs require robust detection of “no more producers/consumers” to trigger cleanup. Without that, stale topics/subs accumulate in service B, which can trigger too much open files error, and
• Resource footprint: Each new producer/consumer initiates broker connections, which takes some time, might not be very big, but we still try to avoid things like that
• We missed TTL early on and it's still a pretty big number for it to not delete a good topic, we use per namespace TTL, a day or few, Which doesn't save us from some problems I'll describe in next part

What is the reason for avoiding duplication? Are the messages large?

Not really, we’re fine with duplication in pipelines when it’s intentional. Our concern was accidental fan-out during RPC-like flows, which can inflate load with no benefit.

In our dev env a small bug created a new topic per request. A smoke test then spawned a very large number of topics. Broker or bookkeeper became overloaded, and CLI tool became useless, throwing Java OOMs, we couldn't delete topics, and TTL was set at a few days, so we had to manually reset. If i remember correctly it was just bookeeper, but might have been full pulsar broker.

It was our mistake and a bad design decision overall, but it pushed us toward a design that minimizes blast radius from such bugs.

Thanks again for reply, might look into the regex solution in case, but I think we'll stick with gRPC for RPC. since it is what it's build for in a first place.

Also I've heard that there is a RPC coming or is it something else https://deepwiki.com/apache/pulsar-java-contrib/4.1-rpc-system-architecture as far as i understood its a proposal?