Playing with the producer name as a way of adjusting message deduplication?

[aradalvand]

Hey there; it seems like message dedupliation doesn't work across the partitions of a given topic (meaning that topic-partition has its own <producer name, sequence ID> tuple); this is undesirable in most cases, I would say.

If the messages that your publisher publishes have partition keys, would it be a sound approach to use the partition key as part of the producer name, so as to effectively give each "logical" partition its own sequence ID?

[lhotari]

Hey there; it seems like message dedupliation doesn't work across the partitions of a given topic (meaning that topic-partition has its own <producer name, sequence ID> tuple); this is undesirable in most cases, I would say.

I don't think that global deduplication across partitions would make much sense with Pulsar. Perhaps there's another way to solve your use case. Please describe your use case.

If the messages that your publisher publishes have partition keys, would it be a sound approach to use the partition key as part of the producer name, so as to effectively give each "logical" partition its own sequence ID?

Are you routing each message on the producer side to a single partition by the message key (this is enabled by default when messages contain keys)? routing to a single partition by message key would most likely solve your deduplication problem in the first place, as long as you have other conditions of deduplication fulfilled. Please check the docs about message deduplication.

[aradalvand]

@lhotari

Please describe your use case.

I have streams of messages that belong to the same topic but are categorized into groups by what I call a "partition key", meaning that messages within each "logical partition" should have their own set of sequence IDs. I need to somehow express this in Pulsar, I figured using the producer name for that purpose would do the trick.

Are you routing each message on the producer side to a single partition by the message key (this is enabled by default when messages contain keys)? routing to a single partition by message key would most likely solve your deduplication problem in the first place

Yes, but that's not sufficient, because the number of physical partitions in a given Pulsar topic might not (and in fact often does
not) match the number of logical messages partitions — or "groups", if that helps disambiguate what I'm talking about from physical Pulsar/Kafka partitions.

My problem would've been solved if Pulsar tracked sequence IDs per message key; but it doesn't. But I can simulate that functionality by using these partition/group keys in the producer name; what would be the problem with that?

Thanks.

[aradalvand]

@lhotari

Looks reasonable if you have a reasonable amount of your "logical partition keys".

What if there are a lot of those though? :)

You must close producers after use. If your code is for a one-off use it looks fine, but if you are repetitively creating new clients or producers, that could cause a lot of unnecessary overhead.

Is there any other approach you'd suggest for my kind of use case though? What would you do if it were you?

[lhotari]

@lhotari

Looks reasonable if you have a reasonable amount of your "logical partition keys".

What if there are a lot of those though? :)

Well, it adds overhead if there are a lot of producers for each partition. You'd have to test it yourself what impact it has to your Pulsar usage and operations.
How many different "logical partition keys" do you have?

You must close producers after use. If your code is for a one-off use it looks fine, but if you are repetitively creating new clients or producers, that could cause a lot of unnecessary overhead.

Is there any other approach you'd suggest for my kind of use case though? What would you do if it were you?

Adding caching for producers. Since producers are exclusive, you will have to have a way how to deal with that when you are producing the messages.

[aradalvand]

@lhotari Thanks for the responsiveness.

How many different "logical partition keys" do you have?

It's very dynamic. In the case of say a topic like "UserEvents", the partition key of each message would be the user ID, so we'll
potentially have as many "logical partition keys" as users.

Adding caching for producers. Since producers are exclusive, you will have to have a way how to deal with that when you are producing the messages.

So you mean you would still use the same "producer name as partition key" approach then?

[lhotari]

It's very dynamic. In the case of say a topic like "UserEvents", the partition key of each message would be the user ID, so we'll potentially have as many "logical partition keys" as users.

That could be a problem since you'd have a lot of active producers in that case and also a lot of state related to the deduplication. The main problem is that it hasn't been designed to be used this way in my opinion. You could run into scaling challenges later. You can use Pulsar in any way you choose to do, but the recommendation is to at least do scaling tests to ensure that the solution can scale to whatever scale your requirements are.

Adding caching for producers. Since producers are exclusive, you will have to have a way how to deal with that when you are producing the messages.

So you mean you would still use the same "producer name as partition key" approach then?

Well in any case producers should be cached. If there's not a finite and known number of partition keys, it feels that the Pulsar producer deduplication is misused and perhaps also misunderstood. The main reason for Pulsar deduplication is to prevent duplicates that occur due to Pulsar at-least-once messaging model. For that purpose you shouldn't need to have logical partition keys are part of the producer names etc..
One of the problems is that the Pulsar documentation for message deduplication doesn't really explain what it's about.
The purpose is described in PIP-6: "To ensure data is written only once in the topic storage, we need to do preventive de-deduplication to identify and reject messages that are being resent after failures."
It's possible to use it for somewhat general purpose deduplication in cases where there's a application level sequence id. However I don't see how this would match your use case. I don't have a ready made solution to your problem, but it's definitely a generic problem to be able to deduplicate messages based on a key. It's a common problem about how to achieve idempotency in message processing pipelines.

One clarifying question about your use case:
What is the source of your duplicate messages? Is it caused by Pulsar itself? Do duplicate messages already arrive to the entry point where you produce the messages and you'd like to deduplicate them?

[aradalvand]

...also a lot of state related to the deduplication

@lhotari True; this is what prompted me to post this question, I wasn't sure if the potentially large amount of "producerName, sequenced" state would be problematic in Pulsar, or not.

Well in any case producers should be cached. If there's not a finite and known number of partition keys, it feels that the Pulsar producer deduplication is misused and perhaps also misunderstood.

Hmm, I see. Bummer.

What is the source of your duplicate messages? Is it caused by Pulsar itself? Do duplicate messages already arrive to the entry point where you produce the messages and you'd like to deduplicate them?

I'm using the transactional outbox pattern, which inherently has at-least-once semantics, introducing the possibility of duplicate publishing of messages — e.g. if the app crashes after publishing a message to the broker but before marking it as published in the outbox data store.