-
Notifications
You must be signed in to change notification settings - Fork 0
Home
Subscriptions cursors are local today. This means that consumers cannot move their state across clusters and start where they left off. This is a proposal to provide some sort of ability within limits, to migrate subscriptions.
This proposal is different from PIP-33
- in how the cursor state is represented and calculated
- is resilient to cluster member failure
- does not require storing predetermined snapshots
and similar in the mechanism to update subscriptions in remote regions
Full-fledged global ordering is not considered or attempted even roughly, by this proposal. Instead, the approach here is to make cursor state approximately transferable globally. In practical terms this means that when migrating subscriptions, there will be duplicates delivered to the subscription.
The rest of this document is in terms of three physical clusters A, B, C, but it is generally applicable to any number of clusters.
This proposal applies to a single physical topic; for partitioned topics it will apply on a per partition basis. If this document mentions “ from/in cluster A”, it is to be construed as a shortened version of on the given topic from/in cluster A
We term the set of cluster on which a topic is replicated as the replication set. (A, B, C).
Each cluster has local producers, publishing a stream of msgs, denoted by a letter,number. The letter is the publishing cluster and the number is the sequence-id/msg-id. This number is assigned at the publish time is termed source sequence-id or the SSID
Eg: a1, a2, a3..etc in A, and b1, b2, b3....etc in B and c1, c2, c3 in C and so on.
Here a1 is a message published at A with a local (or source) SSID of 1, ...and so on.
No ordering is implied or assumed across a, b or c streams.
Pulsar replicates messages in order, from cluster to cluster. Given messages published in a cluster A, (a1, a2, a3…) those messages are replicated from A to other clusters in the same order as seen by a subscriber at A Given messages published in cluster A, messages are replicated from A to B and from A to C in the same order as at A. i.e (a1, a2, a3… )
Incoming replication message streams are merged with the locally published message stream in a manner that preserves per source ordering. This merged stream is dispatched to subscribers. The merge preserves the following properties A subscriber at A is guaranteed to receive b1 before b2, b2 before b3... Similarly, c1 before c2 , c2 before c3, etc No order is preserved across replication streams b and c (for eg: b1 can appear after c3 or before c3, depending on timing and/or replication delay) Whatever ordering across streams is established locally during merge at a cluster is preserved durably at that cluster For eg: if the merge at A establishes the order a1, b1, b2, b3, c1, a2, a3, b4 c2 … every subscriber at A will see the same order. Every cluster creates its own, and possibly different merge ordering. For eg: at Cluster A it could be a1, b1, b2, b3, c1, a2, a3, b4 c2, … while at Cluster B it could be b1, b2, c1, a1, b3, a2, a3, c2, b4 …
If_ local SSIDs are transmitted and preserved across replication, then, any position in the merged stream can be identified by the union of the positions of the substream.
A set of sub-stream cursor positions (cursor set) completely preserves the cursor state globally. A cursor set( a-x, b-y c-z) establishes a global state. A read can resume at any cluster in the replication set by starting at any position (a-p, b-q, c-r) in that stream at that cluster which satisfies (p < x && q < y && r < z) and filtering. Since (a-0, b-0, c-0) always satisfies this condition for any (x,y,z), the set (a-p, b-q, c-r) is guaranteed to exist.
For eg: Given Cluster A has order a1, b1, b2, b3, c1, a2, a3, b4 c2, and Cluster B has order b1, b2, c1, a1, b3, a2, a3, c2, b4 …
A cursor set (a2, b3, c1) can be resumed(*) at A (or B), by starting at b3 in cluster A (or c1 in cluster B). The reader at B can filter the messages preceding the substream positions in the set
Cluster A a1, b1, b2, b3*, c1, a2, a3, b4 c2, ... Cluster B b1, b2, c1*, a1, b3, a2, a3, c2, b4 …
Pulsar cannot transfer cursor state as it is
Pulsar does not send source sequence ids (SSID) across replication. As implemented, the remote messages persisted in a cluster has no data to reconstruct the SSIDs. The post-merge sequence-ids/msg-ids are generated in the receiving cluster. The source name can be reconstructed from a msg as it is part of the msg.
Persisted message stream. What’s discussed above, vs what’s implemented in Pulsar: Cluster A a1, b1, b2, b3, c1, a2, a3, b4, c2, Cluster A a1(a), a2(b), a3(b), a4(b), a5(c), a6(a), a7(a), a8(b), a9(c),
Cluster B b1, b2, c1, a1, b3, a2, a3, c2, b4 … Cluster B b1(b), b2(b), b3(c), b4(a), b5(b), b6(a), b7(a), b8(c), b9(b) …
There is no means to transfer cursor state across clusters, because the merged streams have sequence-ids that cannot be compared.
Solution 1 (trivial)
Send source sequence id (SSID) as a message key. Message keys are preserved across replication. Extract the SSID from the msgs, and use it to construct the possible cursor set. On receiving an ack, update the cursor state
Cons: Every message need to carry its SSID Subscriber acks are on post-merge local msgids. Broker has to maintain a cursor state map which will map (local msgid->source-seqid) to maintain the position of the cursor as a cursor set
For eg: On Cluster A, with SSIDs in the msgs, the merge stream will be persisted as below
Cluster A a1(a,1), a2(b,1), a3(b,2), a4(b,3), a5(c,1), a6(a,2), a7(a,3), a8(b,4), a9(c,2),
where a2(b,3), denotes a msg generated at B with an SSID of 3, persisted at A as the 3rd message
Solution 2
Similar to Soln. 1, but not every message carries an SSID key with it . Keys are set only on msgs at a specific interval (every n msgs, or every t secs)
For eg: if only every alternate msg had the source sequence-id key, then the persisted stream is
Cluster A a1(a,1), a2(b,1), a3(b), a4(b,3), a5(c,1), a6(a), a7(a,3), a8(b), a9(c), ..
Every time a message with a key is read for dispatch, there is a new possible cursor state created for that subscriber
.. -(0,0,0) --> S0 a1 (1,0,0) -->S1 a2 (1,1,0) -->S2 a4 (1,3, 0) -->S3 a5 (1,3,1) -->S4 a7 (3,1,1) -->S5
With this, it’s possible to reason about the where the cursor would be.
Till Consumer acks all msgs to Cursor state is
a1 S0 a2 S1 a4 S2
… and so on.
Note that the cursor set changes only when a msg with a key is seen. The larger the gap between msgs with keys, the less the number of possible cursor states and vice versa.
At any time the broker will need to only maintain the possible cursor states in the (read cursor --- ack cursor) window.
Solution 3
Modify Solution2. In this, SSIDs are not set as keys, but inserted by each cluster as separate ticker msgs into the msg stream, and gets replicated.
If we impose the following simple constraint on tickers
A cluster can publish an SSID in a ticker msg only after the message with that SSID is published. A cluster cannot publish a ticker with an SSID less than in the previous published ticker.
then the same algorithm as in Soln 2 for keeping cursor state applies, except that the cursor states are constructed from ticker messages instead of msg keys.
For eg: a ticker is published from A as [a-3]; If a reader processing the merged stream in B or C, reads that ticker, then the reader can assert that (1) a data message published in A with SSID 3 would have preceded this ticker, and (2) any following ticker message from A will NOT have an SSID less than 3
This algorithm is uncoordinated. Each cluster can decide on its own about whether/when to publish a ticker on its own. The freshness/staleness of the snapshot will depend on the integrity/fickleness of clusters to publish tickers. Optimizations of not publishing/slowing a ticker depending on what rate msgs get published at a cluster are possible
Implementation
Producer side Every broker inserts a ticker into the topic stream at some interval.
Consumer side. Dispatcher maintains cursor set maps, as it processes ticker msgs. Ack processing updates cursor state when the cursor set changes. Cursor updates publish the snapshot into the replication stream Local tickers Strictly speaking, the dispatcher can ignore local tickers, and use the msg-id of the local ticker itself as the SSID for the local cluster, to be more accurate. Every local message, in theory, is a new ticker. But for efficiency, the map should not add a new cursor state for every local message. We should only consider a new state on local tickers, and not on every local message.
Cursor persisted state in the cursor ledger is a tuple of (msg-id, state), where state has all the substream cursors. This can be ReplicatedSubscriptionsUpdate and can be done exactly as in PIP-33 . Because there is no predetermined snapshots, these updates could be deltas on the substream cursors, reducing the size of the updates.