Streams: consider allowing non-numerical values for publishingId

[zenios]

Is your feature request related to a problem? Please describe.

Hi

I was in the process of implementing embedded debezium using postgres connector and rabbitmq streams utilizing publishingId based on the provider wal ids from debezium for deduplication

Unfortunately with postgresql wal files there is no single long value that is increasing and ordered. THey do provide the so called LSN number but this is not guarantteed to be ordered since you might have concurrent transactions.

If you want something which is increasing and ordered then you need to combine both last commit lsn and current processing lsn.

Former provides an increasing and ordered number between transactions and later provides an increasing number between multiple statemenets of a transaction.

So in general what needs to be done is to combine both longs in a uint128_t in order to have an increasing and ordered deduplicationId.

More context can be provided on the following links
https://issues.redhat.com/browse/DBZ-2338
https://debezium.zulipchat.com/#narrow/stream/348249-community-postgresql/topic/LSN.20.20order

Describe the solution you'd like

I was wondering if there are discussions in place to lift the long as restriction of publishinId and let the consumer pass anything. If my undertanding is correct publishingId as of now is just saved in the stream and its up to the consumer to retrieve it and compare it in order to decided if they want to deduplicate or not.

If this is correct then might as well allow it to be anything the consumer requires.

Describe alternatives you've considered

No response

Additional context

No response

[michaelklishin]

@zenios thank you for the context. We move the majority of issues to discussions, at least at first, because they lack context. But this is a fairly specific change request and you did explain what you are doing and why. Thank you.

My primary concern is that this would be a breaking protocol change, so we can likely only ship it in 4.0, if publishingId even can accept non-numerical values.

[zenios]

Protocol wise indeed it would be. The clients though shipped as of now i am sure can be converted in a compatible way to support old and new way of working.

For informational purposes is there a specific reason publishingId was designed as a uint64 instad of byte[]?

[michaelklishin]

@zenios according to other members of the team, this was a conscious protocol design decision. Incrementing positive integers are naturally unique, easy to reason about for humans, and most importantly, are very efficient to compare.

I don't have an alternative to offer for Debezium but we would like to keep the protocol this way.

[zenios]

Ok, what about allowing the consumer to set something on the stream on every message published and be able to get it back just like consumerid and more or less what the consumer protoc already has with x-stream-offset

The consumer then can do the filtering by itself.

Also, if you want to stay with number have, you considered converting publishingid from uint64 to uint128? That will cover debezium as well

[zenios]

If we can get the last message published on the consumer, then we will be able to filter on the applications side

[michaelklishin]

By "set something" you mean set a piece of metadata? That sounds reasonable and most of the messaging protocols we support have a way of doing that.

We will discuss that next week. If you have a specific example in mind, of course, please do share it. I'd look at AMQP 1.0 for inspiration as more and more core decisions around RabbitMQ are based around that protocol and its features/ideas.

[zenios]

The same such as publishingId.This is saved somewhere inside the streams and its available for retrieve.If we could have the option to save more things as well i will save the debezum id and do the deduplication my-self before sending to the producer.

[kjnilsson]

I was wondering if there are discussions in place to lift the long as restriction of publishinId and let the consumer pass anything. If my undertanding is correct publishingId as of now is just saved in the stream and its up to the consumer to retrieve it and compare it in order to decided if they want to deduplicate or not.

No the stream maintains the publishing-id and will not write any messages for a given publisher if they send an id that is lower than or equal to the last publishing id seen for the given publisher name. Consumers do not need to do anything additional to de-duplicate. In fact this publishing-id is not exposed to consumers (unless you add it to the message meta data yourself).

[kjnilsson]

I am not sure what "next to each other" means. Can you provide a realistic, achievable example where combining two lsns results in a uint64 overflow?

[zenios]

        long firstLong = 1234567890123456789L;
        long secondLong = 9876543210987654321L;
        String combinedString = String.valueOf(firstLong) + String.valueOf(secondLong);
        BigInteger bigInteger = new BigInteger(combinedString);

Thats what i mean combining two longs into a BigInteger

[kjnilsson]

1234567890123456789 would at a rate of 1M increments a second take 40000 years to get to.

[kjnilsson]

I think we can save ourselves some time here. We are not going to add 8 bytes of overhead to every de-duplication Id (publishing-id) we write to the stream as we write a lot of them. A uint64 is plenty large enough to handle all reasonable use cases. I am not even sure that your approach of combining is even guaranteed to be incremental given what I could glean about the behaviour of the two lsns in the attached links.

And if you really want to handle the case where you overflow a uint64 you can detect this in your code and let the values wrap but use a new publisher-id (as I already suggested above).

[zenios]

An LSN is two unit32. First one is the log file (0 based) and the second is byte offset of location in log file.