From 88c70b1d78471a6e9364f42b7daeeb6d5aa2cec3 Mon Sep 17 00:00:00 2001 From: Asaf Mesika Date: Thu, 8 Dec 2022 18:20:03 +0200 Subject: [PATCH 1/3] First draft --- site2/docs/concepts-messaging.md | 114 ++++++++++++++++++++++++++++++- 1 file changed, 113 insertions(+), 1 deletion(-) diff --git a/site2/docs/concepts-messaging.md b/site2/docs/concepts-messaging.md index 357840b8a9452..0ec65a95358e0 100644 --- a/site2/docs/concepts-messaging.md +++ b/site2/docs/concepts-messaging.md @@ -599,10 +599,122 @@ In the diagram below, **Consumer A**, **Consumer B** and **Consumer C** are all #### Key_Shared -In the *Key_Shared* type, multiple consumers can attach to the same subscription. Messages are delivered in distribution across consumers and messages with the same key or same ordering key are delivered to only one consumer. No matter how many times the message is re-delivered, it is delivered to the same consumer. When a consumer connects or disconnects, it causes the served consumer to change some message keys. +In the *Key_Shared* type, multiple consumers can attach to the same subscription. Messages are delivered in distribution across consumers and messages with the same key or same ordering key are delivered to only one consumer. No matter how many times the message is re-delivered, it is delivered to the same consumer. ![Key_Shared subscriptions](/assets/pulsar-key-shared-subscriptions.svg) +There are 3 different types of mapping algorithms dictating how to select a consumer for a given message key (or ordering key): Sticky, Auto-split Hash Range and Auto-split Consistent Hashing. In all of them, the message key (or ordering key) is pass to a hash function (e.g. Murmur3 32bit), which yields a 32bit integer hash. That hash number is then fed to the algorithm to select a consumer from the existing connected consumers. +When a new consumer is connected, thus added to the list of added consumers, the algorithm re-adjusts its mapping such that some keys mapped to existing consumers will be mapped to the newly added consumer. When a consumer is disconnected, thus removed from the list of connected consumers, keys mapped to it will be mapped to other consumers. In the sections below each algorithm will contain an explanation on how a consumer is selected given the message hash, and how the mapping is adjusted given a new consumer is connected or an existing consumer disconnects. + +##### Auto-split Hash Range +The algorithm assumes there is a range of number between 0 to 2^16 (65,536). Each consumer is mapped into a single region in this range, such that together all consumers regions cover all the range. The consumer is selected by running a modulo operation on the message hash by the range size (65,536), the number received ( 0 <= i < 65,536), falls into a single region. The consumer mapped to the region is the one selected. + +Example: + +Suppose we have 4 consumers (C1, C2, C3 and C4), then: +``` + 0 16,384 32,768 49,152 65,536 + |------- C3 ------|------- C2 ------|------- C1 ------|------- C4 ------| +``` + +Given a message key `Order-3459134`, it's hash would be `murmur32("Order-3459134") = 3112179635`, and it's index in the range would be `3112179635 mod 65536 = 6067`. That index is mapped to `[0, 16384)` thus consumer C1 will map to this message key. + +When a consumer is joining, the largest region is chosen and is then split in half - the lower half will be mapped to the newly added consumer and upper half will be mapped to consumer owning that region. Here is how it looks like from 1 to 4 consumers: + +``` +|---------------------------------- C1 ---------------------------------| +|--------------- C2 ----------------|---------------- C1 ---------------| +|------- C3 ------|------- C2 ------|---------------- C1 ---------------| +|------- C3 ------|------- C2 ------|------- C1 ------|------- C4 ------| +``` + +When a consumer is disconnected its region will be merged into the region on its right. Example: +``` +|------- C3 ------|------- C2 ------|---------------- C1 ---------------| +``` + +C1 is disconnected: +``` +|------- C3 ------|-------------------------- C2 -----------------------| +``` + +The advantages of this algorithm is that it affects only a single existing consumer upon add/delete consumer, at the expense of regions not evenly sized. The next algorithm does the other way around. + +##### Auto-split Consistent Hashing +This algorithm uses a Hash Ring. It's a range of number from 0 to MAX_INT (32bit) in which if you traverse the range, when reaching MAX_INT, the next number would be zero. As if you took a line starting from 0 ending at MAX_INT and bent into a circle such that the end glues to the start: + +``` + MAX_INT -----++--------- 0 + || + , - ~ ~ ~ - , + , ' ' , + , , + , , + , , + , , + , , + , , + , , + , , ' + ' - , _ _ _ , ' +``` + +When adding a consumer, we mark 100 points on that circle and associating them to the newly added consumer. For each number between 1 and 100, we concatenate the consumer name to that number and run the hash function on it to get the location of the point on the circle that will be marked. For Example, if the consumer name is "orders-aggregator-pod-2345-consumer" then we would mark 100 points on that circle: +``` + murmur32("orders-aggregator-pod-2345-consumer1") = 1003084738 + murmur32("orders-aggregator-pod-2345-consumer2") = 373317202 + ... + murmur32("orders-aggregator-pod-2345-consumer100") = 320276078 +``` + +Since the hash function has the uniform distribution attribute, those points would be uniformly distributed across the circle. +``` + C1-100 + , - ~ ~ ~ - , C1-1 + , ' ' , + , , + , , C1-2 + , , + , , + , , + , , C1-3 + , , + , , ' + ' - , _ _ _ , ' ... + +``` + +A message is selected by putting its hash on the circle then continue clock-wise on the circle until you reach a marking point. The point might have more than one consumer on it (hash function might have collisions) there for, we run the following operation to get a position within the list of consumers for that point, then we take the consumer in that position: `hash % consumer-list-size = index`. + +When a consumer is added, we add 100 marking points to the circle as explained before. Due to the uniform distribution of the hash function, those 100 points is as if the new consumer takes a small slice of keys out of each existing consumer. It's maintains the even distribution, on the trade-off that it impacts all existing consumers. + +##### Sticky + +The algorithm assumes there is a range of number between 0 to 2^16 (65,536). Each consumer is mapped into a multiple regions in this range. The consumer is selected by running a modulo operation on the message hash by the range size (65,536), the number received ( 0 <= i < 65,536), falls into a single region. The consumer mapped to the region is the one selected. +In this algorithm you have full control. Every newly added consumer lists the ranges it wishes to be mapped to. When the consumer object is built, you can specify the list of ranges. It's your responsibility to make sure there are no collisions and all the range is covered by regions. + +Example: + +Suppose we have 2 consumers (C1 and C2) each specified their ranges, then: +``` +C1 = [0, 16384), (32768, 49152] +C2 = [16384, 32768), (49,152, 65536] + + 0 16,384 32,768 49,152 65,536 + |------- C1 ------|------- C2 ------|------- C1 ------|------- C2 ------| +``` + +Given a message key `Order-3459134`, it's hash would be `murmur32("Order-3459134") = 3112179635`, and it's index in the range would be `3112179635 mod 65536 = 6067`. That index is mapped to `[0, 16384)` thus consumer C1 will map to this message key. + +If the newly connected consumer didn't supply their ranges, or they collide with existing consumer ranges, it's disconnected, removed from the consumers list and reverted as if it never tried to connect. + +##### How to use them? +When building the consumer, you can specify the Key Shared Mode: +* AUTO_SPLIT - Auto-split Hash Range +* STICKY - Sticky + +Auto-split Consistent Hashing will be used instead of Hash Range if the broker configuration `subscriptionKeySharedUseConsistentHashing` is turned on. + :::note When the consumers are using the Key_Shared subscription type, you need to **disable batching** or **use key-based batching** for the producers. From 96cc85e3fe74cc7c8cab057b0b8067c265ed2304 Mon Sep 17 00:00:00 2001 From: Asaf Mesika Date: Sun, 11 Dec 2022 16:56:22 +0200 Subject: [PATCH 2/3] Adding explanation on hashing schemes for key shared subscriptions and explains what happens exactly when a consumer connects and disconnects. --- site2/docs/concepts-messaging.md | 83 +++++++++++++++++++------------- 1 file changed, 49 insertions(+), 34 deletions(-) diff --git a/site2/docs/concepts-messaging.md b/site2/docs/concepts-messaging.md index 0ec65a95358e0..fff3bf272767d 100644 --- a/site2/docs/concepts-messaging.md +++ b/site2/docs/concepts-messaging.md @@ -22,18 +22,18 @@ If the consumption of a message fails and you want this message to be consumed a Messages are the basic "unit" of Pulsar. The following table lists the components of messages. -Component | Description -:---------|:------- -Value / data payload | The data carried by the message. All Pulsar messages contain raw bytes, although message data can also conform to data [schemas](schema-get-started.md). -Key | The key (string type) of the message. It is a short name of message key or partition key. Messages are optionally tagged with keys, which is useful for features like [topic compaction](concepts-topic-compaction.md). -Properties | An optional key/value map of user-defined properties. -Producer name | The name of the producer who produces the message. If you do not specify a producer name, the default name is used. -Topic name | The name of the topic that the message is published to. -Schema version | The version number of the schema that the message is produced with. -Sequence ID | Each Pulsar message belongs to an ordered sequence on its topic. The sequence ID of a message is initially assigned by its producer, indicating its order in that sequence, and can also be customized.
Sequence ID can be used for message deduplication. If `brokerDeduplicationEnabled` is set to `true`, the sequence ID of each message is unique within a producer of a topic (non-partitioned) or a partition. -Message ID | The message ID of a message is assigned by bookies as soon as the message is persistently stored. Message ID indicates a message’s specific position in a ledger and is unique within a Pulsar cluster. -Publish time | The timestamp of when the message is published. The timestamp is automatically applied by the producer. -Event time | An optional timestamp attached to a message by applications. For example, applications attach a timestamp on when the message is processed. If nothing is set to event time, the value is `0`. +| Component | Description | +|:---------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| Value / data payload | The data carried by the message. All Pulsar messages contain raw bytes, although message data can also conform to data [schemas](schema-get-started.md). | +| Key | The key (string type) of the message. It is a short name of message key or partition key. Messages are optionally tagged with keys, which is useful for features like [topic compaction](concepts-topic-compaction.md). | +| Properties | An optional key/value map of user-defined properties. | +| Producer name | The name of the producer who produces the message. If you do not specify a producer name, the default name is used. | +| Topic name | The name of the topic that the message is published to. | +| Schema version | The version number of the schema that the message is produced with. | +| Sequence ID | Each Pulsar message belongs to an ordered sequence on its topic. The sequence ID of a message is initially assigned by its producer, indicating its order in that sequence, and can also be customized.
Sequence ID can be used for message deduplication. If `brokerDeduplicationEnabled` is set to `true`, the sequence ID of each message is unique within a producer of a topic (non-partitioned) or a partition. | +| Message ID | The message ID of a message is assigned by bookies as soon as the message is persistently stored. Message ID indicates a message’s specific position in a ledger and is unique within a Pulsar cluster. | +| Publish time | The timestamp of when the message is published. The timestamp is automatically applied by the producer. | +| Event time | An optional timestamp attached to a message by applications. For example, applications attach a timestamp on when the message is processed. If nothing is set to event time, the value is `0`. | The default size of a message is 5 MB. You can configure the max size of a message with the following configurations. @@ -70,12 +70,12 @@ Producers send messages to brokers synchronously (sync) or asynchronously (async You can have different types of access modes on topics for producers. -Access mode | Description -:-----------|------------ -`Shared` | Multiple producers can publish on a topic.

This is the **default** setting. -`Exclusive` | Only one producer can publish on a topic.

If there is already a producer connected, other producers trying to publish on this topic get errors immediately.

The "old" producer is evicted and a "new" producer is selected to be the next exclusive producer if the "old" producer experiences a network partition with the broker. -`ExclusiveWithFencing`|Only one producer can publish on a topic.

If there is already a producer connected, it will be removed and invalidated immediately. -`WaitForExclusive` | If there is already a producer connected, the producer creation is pending (rather than timing out) until the producer gets the `Exclusive` access.

The producer that succeeds in becoming the exclusive one is treated as the leader. Consequently, if you want to implement a leader election scheme for your application, you can use this access mode. Note that the leader pattern scheme mentioned refers to using Pulsar as a Write-Ahead Log (WAL) which means the leader writes its "decisions" to the topic. On error cases, the leader will get notified it is no longer the leader *only* when it tries to write a message and fails on appropriate error, by the broker. +| Access mode | Description | +|:-----------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `Shared` | Multiple producers can publish on a topic.

This is the **default** setting. | +| `Exclusive` | Only one producer can publish on a topic.

If there is already a producer connected, other producers trying to publish on this topic get errors immediately.

The "old" producer is evicted and a "new" producer is selected to be the next exclusive producer if the "old" producer experiences a network partition with the broker. | +| `ExclusiveWithFencing` | Only one producer can publish on a topic.

If there is already a producer connected, it will be removed and invalidated immediately. | +| `WaitForExclusive` | If there is already a producer connected, the producer creation is pending (rather than timing out) until the producer gets the `Exclusive` access.

The producer that succeeds in becoming the exclusive one is treated as the leader. Consequently, if you want to implement a leader election scheme for your application, you can use this access mode. Note that the leader pattern scheme mentioned refers to using Pulsar as a Write-Ahead Log (WAL) which means the leader writes its "decisions" to the topic. On error cases, the leader will get notified it is no longer the leader *only* when it tries to write a message and fails on appropriate error, by the broker. | :::note @@ -603,11 +603,11 @@ In the *Key_Shared* type, multiple consumers can attach to the same subscription ![Key_Shared subscriptions](/assets/pulsar-key-shared-subscriptions.svg) -There are 3 different types of mapping algorithms dictating how to select a consumer for a given message key (or ordering key): Sticky, Auto-split Hash Range and Auto-split Consistent Hashing. In all of them, the message key (or ordering key) is pass to a hash function (e.g. Murmur3 32bit), which yields a 32bit integer hash. That hash number is then fed to the algorithm to select a consumer from the existing connected consumers. -When a new consumer is connected, thus added to the list of added consumers, the algorithm re-adjusts its mapping such that some keys mapped to existing consumers will be mapped to the newly added consumer. When a consumer is disconnected, thus removed from the list of connected consumers, keys mapped to it will be mapped to other consumers. In the sections below each algorithm will contain an explanation on how a consumer is selected given the message hash, and how the mapping is adjusted given a new consumer is connected or an existing consumer disconnects. +There are three types of mapping algorithms dictating how to select a consumer for a given message key (or ordering key): Sticky, Auto-split Hash Range, and Auto-split Consistent Hashing. Before using the algorithm, the message key (or ordering key) is first passed to a hash function (e.g., Murmur3 32-bit), yielding a 32-bit integer hash. That hash number is then fed to the algorithm to select a consumer from the existing connected consumers. +When a new consumer is connected and thus added to the list of connected consumers, the algorithm re-adjusts the mapping such that some keys currently mapped to existing consumers will be mapped to the newly added consumer. When a consumer is disconnected, thus removed from the list of connected consumers, keys mapped to it will be mapped to other consumers. The sections below will explain how a consumer is selected given the message hash and how the mapping is adjusted given a new consumer is connected or an existing consumer disconnects for each algorithm. ##### Auto-split Hash Range -The algorithm assumes there is a range of number between 0 to 2^16 (65,536). Each consumer is mapped into a single region in this range, such that together all consumers regions cover all the range. The consumer is selected by running a modulo operation on the message hash by the range size (65,536), the number received ( 0 <= i < 65,536), falls into a single region. The consumer mapped to the region is the one selected. +The algorithm assumes there is a range of numbers between 0 to 2^16 (65,536). Each consumer is mapped into a single region in this range, so all mapped regions cover the entire range, and no regions overlap. A consumer is selected for a given key by running a modulo operation on the message hash by the range size (65,536). The number received ( 0 <= i < 65,536) is contained within a single region. The consumer mapped to that region is the one selected. Example: @@ -617,18 +617,27 @@ Suppose we have 4 consumers (C1, C2, C3 and C4), then: |------- C3 ------|------- C2 ------|------- C1 ------|------- C4 ------| ``` -Given a message key `Order-3459134`, it's hash would be `murmur32("Order-3459134") = 3112179635`, and it's index in the range would be `3112179635 mod 65536 = 6067`. That index is mapped to `[0, 16384)` thus consumer C1 will map to this message key. +Given a message key `Order-3459134`, its hash would be `murmur32("Order-3459134") = 3112179635`, and its index in the range would be `3112179635 mod 65536 = 6067`. That index is contained within region `[0, 16384)` thus consumer C1 will be mapped to this message key. -When a consumer is joining, the largest region is chosen and is then split in half - the lower half will be mapped to the newly added consumer and upper half will be mapped to consumer owning that region. Here is how it looks like from 1 to 4 consumers: +When a new consumer is connected, the largest region is chosen and is then split in half - the lower half will be mapped to the newly added consumer and upper half will be mapped to the consumer owning that region. Here is how it looks like from 1 to 4 consumers: ``` +C1 connected: |---------------------------------- C1 ---------------------------------| + +C2 connected: |--------------- C2 ----------------|---------------- C1 ---------------| + +C3 connected: |------- C3 ------|------- C2 ------|---------------- C1 ---------------| + +C4 connected: |------- C3 ------|------- C2 ------|------- C1 ------|------- C4 ------| ``` -When a consumer is disconnected its region will be merged into the region on its right. Example: +When a consumer is disconnected its region will be merged into the region on its right. Examples: + +C4 is disconnected: ``` |------- C3 ------|------- C2 ------|---------------- C1 ---------------| ``` @@ -638,10 +647,10 @@ C1 is disconnected: |------- C3 ------|-------------------------- C2 -----------------------| ``` -The advantages of this algorithm is that it affects only a single existing consumer upon add/delete consumer, at the expense of regions not evenly sized. The next algorithm does the other way around. +The advantages of this algorithm is that it affects only a single existing consumer upon add/delete consumer, at the expense of regions not evenly sized. Thi means some consumers gets more keys that others. The next algorithm does the other way around. ##### Auto-split Consistent Hashing -This algorithm uses a Hash Ring. It's a range of number from 0 to MAX_INT (32bit) in which if you traverse the range, when reaching MAX_INT, the next number would be zero. As if you took a line starting from 0 ending at MAX_INT and bent into a circle such that the end glues to the start: +This algorithm uses a Hash Ring. It's a range of number from 0 to MAX_INT (32-bit) in which if you traverse the range, when reaching MAX_INT, the next number would be zero. It is as if you took a line starting from 0 ending at MAX_INT and bent into a circle such that the end glues to the start: ``` MAX_INT -----++--------- 0 @@ -659,7 +668,7 @@ This algorithm uses a Hash Ring. It's a range of number from 0 to MAX_INT (32bit ' - , _ _ _ , ' ``` -When adding a consumer, we mark 100 points on that circle and associating them to the newly added consumer. For each number between 1 and 100, we concatenate the consumer name to that number and run the hash function on it to get the location of the point on the circle that will be marked. For Example, if the consumer name is "orders-aggregator-pod-2345-consumer" then we would mark 100 points on that circle: +When adding a consumer, we mark 100 points on that circle and associate them to the newly added consumer. For each number between 1 and 100, we concatenate the consumer name to that number and run the hash function on it to get the location of the point on the circle that will be marked. For Example, if the consumer name is "orders-aggregator-pod-2345-consumer" then we would mark 100 points on that circle: ``` murmur32("orders-aggregator-pod-2345-consumer1") = 1003084738 murmur32("orders-aggregator-pod-2345-consumer2") = 373317202 @@ -684,14 +693,14 @@ Since the hash function has the uniform distribution attribute, those points wou ``` -A message is selected by putting its hash on the circle then continue clock-wise on the circle until you reach a marking point. The point might have more than one consumer on it (hash function might have collisions) there for, we run the following operation to get a position within the list of consumers for that point, then we take the consumer in that position: `hash % consumer-list-size = index`. +A consumer is selected for a given message key by putting its hash on the circle then continue clock-wise on the circle until you reach a marking point. The point might have more than one consumer on it (hash function might have collisions) there for, we run the following operation to get a position within the list of consumers for that point, then we take the consumer in that position: `hash % consumer_list_size = index`. -When a consumer is added, we add 100 marking points to the circle as explained before. Due to the uniform distribution of the hash function, those 100 points is as if the new consumer takes a small slice of keys out of each existing consumer. It's maintains the even distribution, on the trade-off that it impacts all existing consumers. +When a consumer is added, we add 100 marking points to the circle as explained before. Due to the uniform distribution of the hash function, those 100 points act as if the new consumer takes a small slice of keys out of each existing consumer. It maintains the even distribution, on the trade-off that it impacts all existing consumers. [This video](https://www.youtube.com/watch?v=zaRkONvyGr8) explains the concept of Consistent Hashing quite well (the only difference is that in Pulsar's case we used K points instead of K hash functions as noted in the comments) ##### Sticky -The algorithm assumes there is a range of number between 0 to 2^16 (65,536). Each consumer is mapped into a multiple regions in this range. The consumer is selected by running a modulo operation on the message hash by the range size (65,536), the number received ( 0 <= i < 65,536), falls into a single region. The consumer mapped to the region is the one selected. -In this algorithm you have full control. Every newly added consumer lists the ranges it wishes to be mapped to. When the consumer object is built, you can specify the list of ranges. It's your responsibility to make sure there are no collisions and all the range is covered by regions. +The algorithm assumes there is a range of numbers between 0 to 2^16 (65,536). Each consumer is mapped to a multiple regions in this range and there is no overlap between regions. The consumer is selected by running a modulo operation on the message hash by the range size (65,536), the number received (0 <= i < 65,536), is contained within a single region. The consumer mapped to the region is the one selected. +In this algorithm you have full control. Every newly added consumer specifies the ranges it wishes to be mapped to by using Consumer API. When the consumer object is constructed, you can specify the list of ranges. It's your responsibility to make sure there are no overlaps and all the range is covered by regions. Example: @@ -704,17 +713,23 @@ C2 = [16384, 32768), (49,152, 65536] |------- C1 ------|------- C2 ------|------- C1 ------|------- C2 ------| ``` -Given a message key `Order-3459134`, it's hash would be `murmur32("Order-3459134") = 3112179635`, and it's index in the range would be `3112179635 mod 65536 = 6067`. That index is mapped to `[0, 16384)` thus consumer C1 will map to this message key. +Given a message key `Order-3459134`, it's hash would be `murmur32("Order-3459134") = 3112179635`, and it's index in the range would be `3112179635 mod 65536 = 6067`. That index is contained within `[0, 16384)` thus consumer C1 will map to this message key. -If the newly connected consumer didn't supply their ranges, or they collide with existing consumer ranges, it's disconnected, removed from the consumers list and reverted as if it never tried to connect. +If the newly connected consumer didn't supply their ranges, or they overlap with existing consumer ranges, it's disconnected, removed from the consumers list and reverted as if it never tried to connect. ##### How to use them? When building the consumer, you can specify the Key Shared Mode: * AUTO_SPLIT - Auto-split Hash Range * STICKY - Sticky -Auto-split Consistent Hashing will be used instead of Hash Range if the broker configuration `subscriptionKeySharedUseConsistentHashing` is turned on. +Consistent Hashing will be used instead of Hash Range for Auto-split if the broker configuration `subscriptionKeySharedUseConsistentHashing` is enabled. + +##### Preserving order of processing + +Key Shared Subscription type guarantees a key will be processed by a *single* consumer at any given time. When a new consumer a connected, some key will be mapped to it from existing consumers. The broker will not deliver messages to the new consumer until all messages delivered up until connection time will be acknowledged. This will guarantee a certain key is processed by a single consumer at any given time. The trade-off is that if one of the existing consumers is stuck and no time-out was defined (acknowledging for you), the new consumer won't receive any messages until the stuck consumer will resume or gets disconnected. +That requirement can be relaxed by enabling `allowOutOfOrderDelivery` via the Consumer API. If set on the new consumer, then when it is connected, the broker will allow it to receive messages knowing some messages of that key may be still be processing in other consumers at the time, thus order may be affected for that short period of adding a new consumer. +##### Batching for Key Shared Subscriptions :::note When the consumers are using the Key_Shared subscription type, you need to **disable batching** or **use key-based batching** for the producers. From 3ad593ff7b72199e8878a64dee54f46c1d246a73 Mon Sep 17 00:00:00 2001 From: Asaf Mesika Date: Mon, 12 Dec 2022 12:26:28 +0200 Subject: [PATCH 3/3] PR Fixes #1 --- site2/docs/concepts-messaging.md | 26 +++++++++++++++++++++++--- 1 file changed, 23 insertions(+), 3 deletions(-) diff --git a/site2/docs/concepts-messaging.md b/site2/docs/concepts-messaging.md index fff3bf272767d..e8cd375629860 100644 --- a/site2/docs/concepts-messaging.md +++ b/site2/docs/concepts-messaging.md @@ -603,15 +603,27 @@ In the *Key_Shared* type, multiple consumers can attach to the same subscription ![Key_Shared subscriptions](/assets/pulsar-key-shared-subscriptions.svg) -There are three types of mapping algorithms dictating how to select a consumer for a given message key (or ordering key): Sticky, Auto-split Hash Range, and Auto-split Consistent Hashing. Before using the algorithm, the message key (or ordering key) is first passed to a hash function (e.g., Murmur3 32-bit), yielding a 32-bit integer hash. That hash number is then fed to the algorithm to select a consumer from the existing connected consumers. +There are three types of mapping algorithms dictating how to select a consumer for a given message key (or ordering key): Sticky, Auto-split Hash Range, and Auto-split Consistent Hashing. The steps for all algorithms are: +1. The message key (or ordering key) is passed to a hash function (e.g., Murmur3 32-bit), yielding a 32-bit integer hash. +2. That hash number is fed to the algorithm to select a consumer from the existing connected consumers. + +``` + +--------------+ +-----------+ +Message Key -----> / Hash Function / ----- hash (32-bit) -------> / Algorithm / ----> Consumer + +---------------+ +----------+ +``` + + When a new consumer is connected and thus added to the list of connected consumers, the algorithm re-adjusts the mapping such that some keys currently mapped to existing consumers will be mapped to the newly added consumer. When a consumer is disconnected, thus removed from the list of connected consumers, keys mapped to it will be mapped to other consumers. The sections below will explain how a consumer is selected given the message hash and how the mapping is adjusted given a new consumer is connected or an existing consumer disconnects for each algorithm. ##### Auto-split Hash Range + The algorithm assumes there is a range of numbers between 0 to 2^16 (65,536). Each consumer is mapped into a single region in this range, so all mapped regions cover the entire range, and no regions overlap. A consumer is selected for a given key by running a modulo operation on the message hash by the range size (65,536). The number received ( 0 <= i < 65,536) is contained within a single region. The consumer mapped to that region is the one selected. Example: Suppose we have 4 consumers (C1, C2, C3 and C4), then: + ``` 0 16,384 32,768 49,152 65,536 |------- C3 ------|------- C2 ------|------- C1 ------|------- C4 ------| @@ -638,11 +650,13 @@ C4 connected: When a consumer is disconnected its region will be merged into the region on its right. Examples: C4 is disconnected: + ``` |------- C3 ------|------- C2 ------|---------------- C1 ---------------| ``` C1 is disconnected: + ``` |------- C3 ------|-------------------------- C2 -----------------------| ``` @@ -650,6 +664,7 @@ C1 is disconnected: The advantages of this algorithm is that it affects only a single existing consumer upon add/delete consumer, at the expense of regions not evenly sized. Thi means some consumers gets more keys that others. The next algorithm does the other way around. ##### Auto-split Consistent Hashing + This algorithm uses a Hash Ring. It's a range of number from 0 to MAX_INT (32-bit) in which if you traverse the range, when reaching MAX_INT, the next number would be zero. It is as if you took a line starting from 0 ending at MAX_INT and bent into a circle such that the end glues to the start: ``` @@ -676,7 +691,8 @@ When adding a consumer, we mark 100 points on that circle and associate them to murmur32("orders-aggregator-pod-2345-consumer100") = 320276078 ``` -Since the hash function has the uniform distribution attribute, those points would be uniformly distributed across the circle. +Since the hash function has the uniform distribution attribute, those points would be uniformly distributed across the circle. + ``` C1-100 , - ~ ~ ~ - , C1-1 @@ -705,6 +721,7 @@ In this algorithm you have full control. Every newly added consumer specifies th Example: Suppose we have 2 consumers (C1 and C2) each specified their ranges, then: + ``` C1 = [0, 16384), (32768, 49152] C2 = [16384, 32768), (49,152, 65536] @@ -718,6 +735,7 @@ Given a message key `Order-3459134`, it's hash would be `murmur32("Order-3459134 If the newly connected consumer didn't supply their ranges, or they overlap with existing consumer ranges, it's disconnected, removed from the consumers list and reverted as if it never tried to connect. ##### How to use them? + When building the consumer, you can specify the Key Shared Mode: * AUTO_SPLIT - Auto-split Hash Range * STICKY - Sticky @@ -726,10 +744,12 @@ Consistent Hashing will be used instead of Hash Range for Auto-split if the brok ##### Preserving order of processing -Key Shared Subscription type guarantees a key will be processed by a *single* consumer at any given time. When a new consumer a connected, some key will be mapped to it from existing consumers. The broker will not deliver messages to the new consumer until all messages delivered up until connection time will be acknowledged. This will guarantee a certain key is processed by a single consumer at any given time. The trade-off is that if one of the existing consumers is stuck and no time-out was defined (acknowledging for you), the new consumer won't receive any messages until the stuck consumer will resume or gets disconnected. +Key Shared Subscription type guarantees a key will be processed by a *single* consumer at any given time. When a new consumer a connected, some key will be mapped to it from existing consumers. The broker will not deliver messages to the new consumer until all messages delivered up until connection time will be acknowledged. This will guarantee a certain key is processed by a single consumer at any given time. The trade-off is that if one of the existing consumers is stuck and no time-out was defined (acknowledging for you), the new consumer won't receive any messages until the stuck consumer will resume or gets disconnected. + That requirement can be relaxed by enabling `allowOutOfOrderDelivery` via the Consumer API. If set on the new consumer, then when it is connected, the broker will allow it to receive messages knowing some messages of that key may be still be processing in other consumers at the time, thus order may be affected for that short period of adding a new consumer. ##### Batching for Key Shared Subscriptions + :::note When the consumers are using the Key_Shared subscription type, you need to **disable batching** or **use key-based batching** for the producers.