Questions regarding pulsar active-active geo-replication

[Apurva007]

1. In active-active replication explanation on the pulsar website, it says "consumers can consume all messages from all data centers". The diagram shows the consumer instances of subscription S1 are connecting to both Clusters A & B respectively. In this cases, how are the offsets managed across clusters? How is this pattern not causing 100% data duplication in consumption due to same data being available on both clusters?
2. In the same pattern, if the subscription state is replicated, and the consumers of S1 subscription are connecting to both clusters, is there an internal protection in the replicators to make sure that replicated state does not override the current offsets in the cluster due to active consumers already using these offsets?

[lhotari]

Thank you @Apurva007, good questions.

In this cases, how are the offsets managed across clusters?

The messages in different clusters don't share the same message ids. The message ids of the originating cluster are independent of the message ids in the remote cluster.

There are 2 parts to what you could call "offset management" across clusters.

For replication itself, messages originating from one cluster to a remote cluster are handled by a replicator instance for each topic in the originating cluster which will publish (push) messages to the remote cluster and keep the state in a special subscription about this. To prevent replication loops, the message that is published in the remote cluster will contain metadata about the originating cluster and the original message id. The replicator is special in this sense that it's like a consumer but it's directly implemented in the Pulsar broker on top of the "managed ledger" layer without a consumer.
This can be seen in the Pulsar architecture diagram as "global replicators".

For replicated subscriptions, reading "PIP 33: Replicated subscriptions" and especially the "Construction a cursor snapshot" is helpful in understanding how "offset management" works under the covers and what the limitations are. There's also a blog post that contains a useful summary of the limitations. The subscription snapshotting seems to be an application of Vector clocks although this isn't explicitly mentioned in the PIP-33 design document. There's another discussion #21612 which contains useful observations and details about replicated subscriptions.

In the same pattern, if the subscription state is replicated, and the consumers of S1 subscription are connecting to both clusters, is there an internal protection in the replicators to make sure that replicated state does not override the current offsets in the cluster due to active consumers already using these offsets?

Shared subscriptions using the same replicated subscription across geo-replication clusters don't have consistent behavior. It "works", but the same offsets would get consumed in both clusters in non deterministic ways. I haven't validated this what I'm saying, but I have the understanding that in many cases the messages would get processed by the concurrent consumers sharing the same replicated subscription name in both clusters, but not at all times. For use cases where there's a requirement to have at-least-once processing in any of the clusters with the replicated cluster, this is fine when a lot of duplicates aren't a problem. My understanding is that replicated subscriptions are designed to be used for active-passive configurations where some overlap isn't a problem and where there's an external solution for handling the solution for choosing which consumer should be active for a particular replicated subscription. It seems that the documentation supports this:

In case of failover, a consumer can restart consuming from the failure point in a different cluster.

[Apurva007]

@lhotari Thanks for the great explanation. That helps clear most of my questions.
A follow-up question to the "offset management" was "How is this pattern not causing 100% data duplication in consumption due to same data being available on both clusters?"

Please can you help explain how this diagram works:

Eg. A client application in its service url added the URLs of both cluster A and cluster B as comma separated values. Geo replication of data is enabled in both clusters. Subscription replication is disabled.

Messages published to Cluster A: M1, M2, M3
Messages published to Cluster B: M4, M5
Data availability on cluster A & B after replication: M1, M4, M2, M3, M5

As per above diagram, lets say subscription S1 having C1 and C2 consumers connecting to both cluster A and cluster B in the same instance. What would be the expected consumption behavior?

1. S1 receives M1, M4, M2, M3, M5 only once
2. S1 receives M1, M4, M2, M3 and M5 twice.

If only once, then how is the subscription being tracked across clusters without subscription replication?

[lhotari]

I'm sorry I missed the reference to the active-active replication docs in your question. Thanks for the follow up.

If only once, then how is the subscription being tracked across clusters without subscription replication?

It seems that the example in the documentation is missing that detail. If there wouldn't be subscription replication, the subscriptions would be completely independent.

Eg. A client application in its service url added the URLs of both cluster A and cluster B as comma separated values.

This detail makes the scenario active-passive from the application (consumers/producers) point of view. The Pulsar client and its consumer would connect to only one cluster at a time. This is needed for consistent usage of replicated subscriptions. As I mentioned in my previous message, the behaviour isn't consistent when the replicated subscription is actively used in more than one cluster at a time.

Even with replicated subscriptions, the diagram doesn't make full sense to me since there are two separate consumers C1 and C2 in the diagram. When there are 2 service URLs for the client, it would connect to the first cluster that is available and this would be the correct way to use replicated subscriptions.

There are important limitations for replicated subscriptions. For at-least-once messaging with a consumer for a replicated subscription consuming only on one cluster at a time, this is usually fine when delayed messages aren't used.

The main limitation of replicated subscription is that only the "mark delete" position is replicated. Any individually "deleted" (acknowledged) messages will be ignored. This is explained in Penghui's presentation at 1:12:26. Naturally, batch index acknowledgements aren't supported either.

Delayed messages prevent the mark delete position from moving forward until the delayed message has been delivered and acknowledged. This is why delayed messages together with replicated subscriptions isn't a good solution if the large amount of duplicates are a problem when the consumer switches to consume from the other cluster.

The current documentation for geo replication needs improvements so that it wouldn't cause surprises and unrealistic expectations. Contributions to improve the docs are more than welcome to clarify the points that you have brought up in your questions.

[Apurva007]

@lhotari No problem! This makes alot of sense. Thanks alot of the explanation. I am currently working on learning more about the geo-replication piece of Pulsar. I will contribute a summary of the above explanation to the docs soon. Thanks again!