From e6bfcabe3bf403f1fa240ceb1c3a2d1879e3bb11 Mon Sep 17 00:00:00 2001 From: xiangying <1984997880@qq.com> Date: Fri, 18 Aug 2023 19:16:24 +0800 Subject: [PATCH] [improve][pip] Change cursor`s properties to store chunk ID map. --- pip/pip-295.md | 128 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 128 insertions(+) create mode 100644 pip/pip-295.md diff --git a/pip/pip-295.md b/pip/pip-295.md new file mode 100644 index 0000000000000..6c26be0602f1a --- /dev/null +++ b/pip/pip-295.md @@ -0,0 +1,128 @@ + +# Background knowledge + +In [PIP 37](https://github.com/apache/pulsar/wiki/PIP-37:-Large-message-size-handling-in-Pulsar), Pulsar introduced chunk messages to handle the large message. It will separate a large message into some chunks when the producer sends the significant message to the broker. On the consumer side, a consumer will wait to receive all the chunks of a message and then assemble them into a single chunk message before returning it. +In [PIP 6](https://github.com/apache/pulsar/wiki/PIP-6:-Guaranteed-Message-Deduplication), Pulsar introduced deduplication to make sure the messages sent by the producer are non-repeating. +In PIP 6, each producer will have a sequence ID that starts at 0 and increase for each message. The message with a lower sequence ID will be dropped in the broker. + +# Motivation + +In the earliest design, all the chunks in a single chunk message have the same sequence ID which causes the chunk message can not work when enabling deduplication. For example, we have a chunk message consisting of chunk-1 and chunk-2. When Broker receives chunk-1, it will update the last sequence ID to the sequence ID of chunk-1. And then, when the broker gets chunk-2, the chunk-2 will be dropped by depublication. +I opened a [PR](https://github.com/apache/pulsar/pull/20948) to resolve this case. It allowed the chunks of a single chunk message to use the same sequence ID and filter duplicated chunks in a single-chunk message on the consumer side. +It can resolve message duplication end to end, but the message duplication still exists in the topic. + +# Goals + +## In Scope +Chunk messages can be effectively filtered on the broker side. Ensure that chunk messages work normally after enabling deduplication and the topic has no duplicate chunks. + +## Out of Scope + + + +# High Level Design +Introduce a mechanism similar to [PIP 37](https://github.com/apache/pulsar/wiki/PIP-37:-Large-message-size-handling-in-Pulsar) to check the chunk ID. +For normal messages, we still only check sequence ID, but we will check both sequence ID and chunk ID for chunk messages. + +# Detailed Design + +## Design & Implementation Details + +Add `chunkIDPushed` and `chunkIDPersisted` to store the chunk of each producer`s ongoing chunk messages. It will be used to check whether the chunks in a single message are duplicated. + +``` + @VisibleForTesting + final ConcurrentOpenHashMap chunkIDPushed = + ConcurrentOpenHashMap.newBuilder() + .expectedItems(16) + .concurrencyLevel(1) + .build(); + + @VisibleForTesting + final ConcurrentOpenHashMap chunkIDPersisted = + ConcurrentOpenHashMap.newBuilder() + .expectedItems(16) + .concurrencyLevel(1) + .build(); +``` + +Optimize the `properties` of the `MarkDeleteEntry` from `Map` to `Map`. In the depublication design, the ' MarkDeleteEntry' properties are used as a snapshot to store the sequence ID map. After introducing the chunk ID map, it cannot hold two long for each producer. So we hope to change the `MarkDeleteEntry' properties from `Map` to `Map` to make it more flexible. + + + +## Public-facing Changes +None + +### Public API +None + +### Binary protocol +Add `repeated StringProperty markDeleteProperties = 9;` to replace the original `repeated LongProperty properties = 5;`. + +### Configuration + +### CLI + +### Metrics + + +# Monitoring + + +# Security Considerations + + +# Backward & Forward Compatibility + +## Revert +When reverting to the old version of Pulsar, the `ManagedCursorInfo` will not contain the properties(`repeated LongProperty properties = 5;`). Because the new version of pulsar use markDeleteProperties (`repeated StringProperty markDeleteProperties = 9;`) to record mark delete properties. +So It can only be reverted if losing many last persistent sequence ID data. + +## Upgrade + +Add an upgrade logic in `recover(final VoidCallback callback)`. +The original logic: +```java + Map recoveredProperties = Collections.emptyMap(); + if (info.getPropertiesCount() > 0) { + // Recover properties map + recoveredProperties = new HashMap<>(); + for (int i = 0; i < info.getPropertiesCount(); i++) { + LongProperty property = info.getProperties(i); + recoveredProperties.put(property.getName(), property.getValue()); + } + } + + recoveredCursor(recoveredPosition, recoveredProperties, recoveredCursorProperties, null); +``` +Change to: +```java + // Recover properties map + Map recoveredProperties; + if (info.getPropertiesCount() == 0 && info.getmarkDeletePropertiesCount() == 0) { + recoveredProperties = Collections.emptyMap(); + } else if (info.getPropertiesCount() > 0) { + recoveredProperties = new HashMap<>(); + for (int i = 0; i < info.getPropertiesCount(); i++) { + LongProperty property = info.getProperties(i); + //At this time, the correction of deduplication for chunk messages is not promised. + recoveredProperties.put(property.getName(), String.valueOf(property.getValue())); + } + } else if (info.getmarkDeletePropertiesCount() > 0) { + recoveredProperties = new HashMap<>(); + for (int i = 0; i < info.getmarkDeletePropertiesCount(); i++) { + StringProperty property = info.getmarkDeleteProperties(i); + recoveredProperties.put(property.getName(), property.getValue()); + } + } +``` + + +# Alternatives + +# General Notes + +# Links + +* Mailing List discussion thread: +* Mailing List voting thread: