> I think a simple better approach is to only check the deduplication for the last chunk of the large message. The consumer only gets the whole message after receiving the last chunk. We don't need to check the deduplication for all previous chunks. Also by doing this we only need bug fixes, we don't need to introduce a new PIP.
I believe in this PIP "similar to the existing "sequence ID map", to facilitate effective filtering" actually means tracking the last chunkId(not all chunk ids) on the broker side. This is still a behavior change(deduping chunk messages on the broker), and I believe we need to discuss this addition as a PIP. > If duplication is enabled, the chunk messages 2, 3, 4, and 5 will be filtered. The message also will be discarded. Can you explain this part? It appears that only 3 and 4 will be filtered out, and 1,2, and 5 will complete the chunking. > There is no guarantee that the producer will split the message in the same way twice before and after. There is no guarantee that the producer will split the message in the same way twice before and after. For example, the producer's maxMessageSize may be different. This may cause the consumer to receive a corrupt message. Good point. For this case, I think brokers can track the last chunkMaxMessageSize for each producer. - if the current chunkId is 0 <= last chunkId, - and if any chunk(excluding the last one)'s size is different from this last chunkMaxMessageSize, then the brokers can assume that the producer is resending the chunks from the beginning with a different scheme(restarted with a different chunkMaxMessageSize) and accept those new chunks from the beginning. Thanks, Heesung On Wed, Aug 23, 2023 at 2:16 AM Xiangying Meng <xiangy...@apache.org> wrote: > Hi Zike > Thank you for your attention. > >For the existing behavior, the consumer assembles messages 3,4,5 into the > original large message. But the changes brought about by this PIP will > cause the consumer to use messages 1,2,5 for assembly. There is no > guarantee that the The producer will split the message in the same way > twice before and after. For example, the producer's maxMessageSize may be > different. This may cause the consumer to receive a corrupt message. > > For the previous behavior, if duplication is not enabled, messages 1, > 2, 3, 4, and 5 will all be persisted. When the consumer receives > message 3 (sequenceID: 0, ChunkID: 0), it will find that the message > is out of order and rewind the cursor. Loop this operation, and > discard this message after it expires instead of assembling 3, 4, 5 > into a message. > If duplication is enabled, the chunk messages 2, 3, 4, and 5 will be > filtered. The message also will be discarded. > > >I think a simple better approach is to only check the deduplication for > the last chunk of the large message. > > This solution also cannot solve the out-of-order messages inside the > chunks. For example, the above five messages will still be persisted. > > On Wed, Aug 23, 2023 at 12:34 PM Zike Yang <z...@apache.org> wrote: > > > > Hi, xiangying, > > > > Thanks for your PIP. > > > > IIUC, this may change the existing behavior and may introduce > inconsistencies. > > Suppose that we have a large message with 3 chunks. But the producer > > crashes and resends the message after sending the chunk-1. It will > > send a total of 5 messages to the Pulsar topic: > > > > 1. SequenceID: 0, ChunkID: 0 > > 2. SequenceID: 0, ChunkID: 1 > > 3. SequenceID: 0, ChunkID: 0 -> This message will be dropped > > 4. SequenceID: 0, ChunkID: 1 -> Will also be dropped > > 5. SequenceID: 0, ChunkID: 2 -> The last chunk of the message > > > > For the existing behavior, the consumer assembles messages 3,4,5 into > > the original large message. But the changes brought about by this PIP > > will cause the consumer to use messages 1,2,5 for assembly. There is > > no guarantee that the producer will split the message in the same way > > twice before and after. For example, the producer's maxMessageSize may > > be different. This may cause the consumer to receive a corrupt > > message. > > > > Also, this PIP increases the complexity of handling chunks on the > > broker side. Brokers should, in general, treat the chunk as a normal > > message. > > > > I think a simple better approach is to only check the deduplication > > for the last chunk of the large message. The consumer only gets the > > whole message after receiving the last chunk. We don't need to check > > the deduplication for all previous chunks. Also by doing this we only > > need bug fixes, we don't need to introduce a new PIP. > > > > BR, > > Zike Yang > > > > On Fri, Aug 18, 2023 at 7:54 PM Xiangying Meng <xiangy...@apache.org> > wrote: > > > > > > Dear Community, > > > > > > I hope this email finds you well. I'd like to address an important > > > issue related to Apache Pulsar and discuss a solution I've proposed on > > > GitHub. The problem pertains to the handling of Chunk Messages after > > > enabling deduplication. > > > > > > In the current version of Apache Pulsar, all chunks of a Chunk Message > > > share the same sequence ID. However, enabling the depublication > > > feature results in an inability to send Chunk Messages. To tackle this > > > problem, I've proposed a solution [1] that ensures messages are not > > > duplicated throughout end-to-end delivery. While this fix addresses > > > the duplication issue for end-to-end messages, there remains a > > > possibility of duplicate chunks within topics. > > > > > > To address this concern, I believe we should introduce a "Chunk ID > > > map" at the Broker level, similar to the existing "sequence ID map", > > > to facilitate effective filtering. However, implementing this has led > > > to a challenge: a producer requires storage for two Long values > > > simultaneously (sequence ID and chunk ID). Because the snapshot of the > > > sequence ID map is stored through the properties of the cursor > > > (Map<String, Long>), so in order to satisfy the storage of two Longs > > > (sequence ID, chunk ID) corresponding to one producer, we hope to add > > > a mark DeleteProperties (Map<String, Long>) String, String>) to > > > replace the properties (Map<String, Long>) field. To resolve this, > > > I've proposed an alternative proposal [2] involving the introduction > > > of a "mark DeleteProperties" (Map<String, String>) to replace the > > > current properties (Map<String, Long>) field. > > > > > > I'd appreciate it if you carefully review both PRs and share your > > > valuable feedback and insights. Thank you immensely for your time and > > > attention. I eagerly anticipate your valuable opinions and > > > recommendations. > > > > > > Warm regards, > > > Xiangying > > > > > > [1] https://github.com/apache/pulsar/pull/20948 > > > [2] https://github.com/apache/pulsar/pull/21027 >
