>> Consumer receive:
>1. SequenceID: 0, ChunkID: 0
>2. SequenceID: 0, ChunkID: 1
>3. SequenceID: 0, ChunkID: 0 // chunk ID out of order. Release this
>chunk and recycle its `chunkedMsgCtx`.
>4. SequenceID: 0, ChunkID: 1 // chunkedMsgCtx == null Release it.
>5. SequenceID: 0, ChunkID: 2 // chunkedMsgCtx == null Release it.
>
>I think this case is wrong. For the current implementation, the
>message 3,4,5 will be assembled as a original large message.
Hi Zike
You can see the processMessageChunk method of the ConsumerImpl.
```
ChunkedMessageCtx chunkedMsgCtx = chunkedMessagesMap.get(msgMetadata.getUuid());
if (msgMetadata.getChunkId() == 0 && chunkedMsgCtx == null) {
//assemble a chunkedMsgCtx and put into
pendingChunkedMessageUuidQueue and chunkedMessagesMap.
}
if (chunkedMsgCtx == null || chunkedMsgCtx.chunkedMsgBuffer == null
|| msgMetadata.getChunkId() !=
(chunkedMsgCtx.lastChunkedMessageId + 1)) {
if (chunkedMsgCtx != null) {
if (chunkedMsgCtx.chunkedMsgBuffer != null) {
ReferenceCountUtil.safeRelease(chunkedMsgCtx.chunkedMsgBuffer);
}
chunkedMsgCtx.recycle();
}
chunkedMessagesMap.remove(msgMetadata.getUuid());
compressedPayload.release();
increaseAvailablePermits(cnx);
}
```
On Sat, Aug 26, 2023 at 3:48 PM Zike Yang <z...@apache.org> wrote:
>
> > Consumer receive:
> 1. SequenceID: 0, ChunkID: 0
> 2. SequenceID: 0, ChunkID: 1
> 3. SequenceID: 0, ChunkID: 0 // chunk ID out of order. Release this
> chunk and recycle its `chunkedMsgCtx`.
> 4. SequenceID: 0, ChunkID: 1 // chunkedMsgCtx == null Release it.
> 5. SequenceID: 0, ChunkID: 2 // chunkedMsgCtx == null Release it.
>
> I think this case is wrong. For the current implementation, the
> message 3,4,5 will be assembled as a original large message.
>
> HI, Heesung
>
>
> > I think brokers probably need to track map<uuid, last_chunk_id> to dedup
>
> I propose a simpler solution in this mail thread earlier, which
> doesn't need to introduce map<uuid, last_chunk_id> :
>
> > 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.
>
> Could you explain or show a case in what cases would lead to this
> simpler solution not working?
>
> Thanks,
> Zike Yang
>
> On Sat, Aug 26, 2023 at 1:38 PM Heesung Sohn
> <heesung.s...@streamnative.io.invalid> wrote:
> >
> > > In this case, the consumer only can receive m1.
> >
> > Regarding this comment, can you explain how the consumer only receives m1?
> > Here, m1's and m2's uuid and msgId will be different(if we suffix with a
> > chunkingSessionId), although the sequence id is the same.
> >
> > > If we throw an exception when users use the same sequence to send the
> > message.
> > Do You mean `If "producers" throw an exception when users use the same
> > sequence to send the message.`.
> > Again, when the producers restart, they lose the last sequence id sent.
> >
> >
> > > If we do not throw an exception when users use the same sequence to
> > send the message.
> >
> > For this logic, how do we handle if the producer suddenly resends the
> > chunked message with a different chunking scheme(e.g. maxMessageSize) ?
> > uuid=1, sid=0, cid=0
> > uuid=1, sid=0, cid=1
> > uuid=2, sid=0, cid=0
> > uuid=2, sid=0, cid=1
> >
> > We could refine what to track and algo logic on the broker side more, but
> > do we agree that the broker chunk dedup logic is needed?
> >
> > I will continue to think more next week. Have a nice weekend.
> >
> >
> >
> >
> > On Fri, Aug 25, 2023 at 9:14 PM Xiangying Meng <xiangy...@apache.org> wrote:
> >
> > > Hi Heesung,
> > >
> > > Maybe we only need to maintain the last chunk ID in a map.
> > > Map<producername, chunkID> map1.
> > > And we already have a map maintaining the last sequence ID.
> > > Map<producername, sequence ID> map2.
> > >
> > > If we do not throw an exception when users use the same sequence to
> > > send the message.
> > >
> > > For any incoming msg, m :
> > > chunk ID = -1;
> > > If m is a chunk message:
> > > chunk ID = m.chunkID.
> > > If m.currentSeqid < LastSeqId, dedup.
> > > If m.currentSeqid > LastSeqId && m.chunk ID = 0, nodedup
> > > if chunk ID exists in the map.
> > > Update it and log an error. This means there is an
> > > incomplete chunk message.
> > > If chunk ID does not exist in the map.
> > > Put it on the map.
> > > If m.currentSeqid == LastSeqId,
> > > 1. if m.chunk ID == -1 || m.chunk ID == 0. dedup.
> > > 2. If 1 <= m.chunkID <= total chunk.
> > > 1. If chunk ID does not exist in the map. dedup.
> > > 2. If chunk ID exists in the map. dedup. Check the order
> > > of the chunkID to determine whether dedup;
> > > 3. If m.chunkID == total chunk, persistent the chunk and
> > > remove the chunkID in the map.
> > >
> > >
> > > If we throw an exception when users use the same sequence to send the
> > > message.
> > >
> > > For any incoming msg, m :
> > > chunk ID = 0;
> > > If m is a chunk message:
> > > chunk ID = m.chunkID.
> > > If m.currentSeqid < LastSeqId, dedup.
> > > If m.currentSeqid == LastSeqId.
> > > If chunkID > 0, Check the last chunkID to determine whether to
> > > dedup.
> > > If chunkID == 1, put chunkID into the map if absent.
> > > IF chunkID = 0, dedup.
> > >
> > > BR,
> > > xiangying
> > >
> > > On Sat, Aug 26, 2023 at 11:53 AM Heesung Sohn
> > > <heesung.s...@streamnative.io.invalid> wrote:
> > > >
> > > > However, what If the producer jvm gets restarted after the broker
> > > persists
> > > > the m1 (but before updating their sequence id in their persistence
> > > > storage), and the producer is trying to resend the same msg(so m2) with
> > > the
> > > > same sequence id after restarting?
> > > >
> > > >
> > > >
> > > >
> > > >
> > > > On Fri, Aug 25, 2023 at 8:22 PM Xiangying Meng <xiangy...@apache.org>
> > > wrote:
> > > >
> > > > > Hi Heesung,
> > > > >
> > > > > In this case, the consumer only can receive m1.
> > > > >
> > > > > But it has the same content as the previous case: What should we do if
> > > > > the user sends messages with the sequence ID that was used previously?
> > > > >
> > > > > I am afraid to introduce the incompatibility in this case, so I only
> > > > > added a warning log in the PR[0] instead of throwing an exception.
> > > > > Regarding this matter, what do you think? Should we throw an exception
> > > > > or add error logs?
> > > > >
> > > > > I'm looking forward to hearing your viewpoint.
> > > > >
> > > > > Thanks,
> > > > > Xiangying
> > > > >
> > > > > [0] https://github.com/apache/pulsar/pull/21047
> > > > >
> > > > > On Sat, Aug 26, 2023 at 10:58 AM Heesung Sohn
> > > > > <heesung.s...@streamnative.io.invalid> wrote:
> > > > > >
> > > > > > Actually, can we think about this case too?
> > > > > >
> > > > > > What happens if the cx sends the same chunked msg with the same seq
> > > id
> > > > > when
> > > > > > dedup is enabled?
> > > > > >
> > > > > > // user send a chunked msg, m1
> > > > > > s1, c0
> > > > > > s1, c1
> > > > > > s1, c2 // complete
> > > > > >
> > > > > > // user resend the duplicate msg, m2
> > > > > > s1, c0
> > > > > > s1, c1
> > > > > > s1, c2 //complete
> > > > > >
> > > > > > Do consumers receive m1 and m2(no dedup)?
> > > > > >
> > > > > >
> > > > > >
> > > > > > On Fri, Aug 25, 2023 at 6:55 PM Xiangying Meng <xiangy...@apache.org
> > > >
> > > > > wrote:
> > > > > >
> > > > > > > Hi Heesung,
> > > > > > >
> > > > > > > >I think this means, for the PIP, the broker side's chunk
> > > > > deduplication.
> > > > > > > >I think brokers probably need to track map<uuid, last_chunk_id>
> > > > > > > >to
> > > > > dedup
> > > > > > >
> > > > > > > What is the significance of doing this?
> > > > > > > My understanding is that if the client crashes and restarts after
> > > > > > > sending half of a chunk message and then it resends the previous
> > > chunk
> > > > > > > message, the resent chunk message should be treated as a new
> > > message
> > > > > > > since it calls the producer's API again.
> > > > > > > If deduplication is enabled, users should ensure that their
> > > customized
> > > > > > > sequence ID is not lower than the previous sequence ID.
> > > > > > > I have considered this scenario and added a warning log in PR[0].
> > > (I'm
> > > > > > > not sure whether an error log should be added or an exception
> > > thrown.)
> > > > > > > If deduplication is not enabled, on the consumer side, there
> > > should be
> > > > > > > an incomplete chunk message received alongside another complete
> > > chunk
> > > > > > > message, each with a different UUID, and they will not interfere
> > > with
> > > > > > > each other.
> > > > > > >
> > > > > > > My main point is that every message sent using
> > > > > > > `producer.newMessage().send()` should be treated as a new message.
> > > > > > > UUID is solely used for the consumer side to identify different
> > > chunk
> > > > > > > messages.
> > > > > > >
> > > > > > > BR
> > > > > > > Xiangying
> > > > > > >
> > > > > > > [0] https://github.com/apache/pulsar/pull/21047
> > > > > > >
> > > > > > > On Sat, Aug 26, 2023 at 9:34 AM Heesung Sohn
> > > > > > > <heesung.s...@streamnative.io.invalid> wrote:
> > > > > > > >
> > > > > > > > I think this means, for the PIP, the broker side's chunk
> > > > > deduplication.
> > > > > > > > I think brokers probably need to track map<uuid, last_chunk_id>
> > > to
> > > > > dedup
> > > > > > > > chunks on the broker side.
> > > > > > > >
> > > > > > > >
> > > > > > > >
> > > > > > > >
> > > > > > > > On Fri, Aug 25, 2023 at 6:16 PM Xiangying Meng <
> > > xiangy...@apache.org
> > > > > >
> > > > > > > wrote:
> > > > > > > >
> > > > > > > > > Hi Heesung
> > > > > > > > >
> > > > > > > > > It is a good point.
> > > > > > > > > Assume the producer application jvm restarts in the middle of
> > > > > chunking
> > > > > > > and
> > > > > > > > > resends the message chunks from the beginning with the
> > > > > > > > > previous
> > > > > > > sequence
> > > > > > > > > id.
> > > > > > > > >
> > > > > > > > > For the previous version, it should be:
> > > > > > > > >
> > > > > > > > > Producer send:
> > > > > > > > > 1. SequenceID: 0, ChunkID: 0
> > > > > > > > > 2. SequenceID: 0, ChunkID: 1
> > > > > > > > > 3. SequenceID: 0, ChunkID: 0
> > > > > > > > > 4. SequenceID: 0, ChunkID: 1
> > > > > > > > > 5. SequenceID: 0, ChunkID: 2
> > > > > > > > >
> > > > > > > > > Consumer receive:
> > > > > > > > > 1. SequenceID: 0, ChunkID: 0
> > > > > > > > > 2. SequenceID: 0, ChunkID: 1
> > > > > > > > > 3. SequenceID: 0, ChunkID: 0 // chunk ID out of order. Release
> > > this
> > > > > > > > > chunk and recycle its `chunkedMsgCtx`.
> > > > > > > > > 4. SequenceID: 0, ChunkID: 1 // chunkedMsgCtx == null Release
> > > it.
> > > > > > > > > 5. SequenceID: 0, ChunkID: 2 // chunkedMsgCtx == null Release
> > > it.
> > > > > > > > >
> > > > > > > > > Therefore, for the previous version, this chunk message can
> > > not be
> > > > > > > > > received by the consumer. It is not an incompatibility issue.
> > > > > > > > >
> > > > > > > > > However, the solution of optimizing the `uuid` is valuable to
> > > the
> > > > > new
> > > > > > > > > implementation.
> > > > > > > > > I will modify this in the PR[0]. Thank you very much for your
> > > > > reminder
> > > > > > > > > and the provided UUID optimization solution.
> > > > > > > > >
> > > > > > > > > BR,
> > > > > > > > > Xiangying
> > > > > > > > >
> > > > > > > > > [0] https://github.com/apache/pulsar/pull/20948
> > > > > > > > >
> > > > > > > > > On Sat, Aug 26, 2023 at 8:48 AM Heesung Sohn
> > > > > > > > > <heesung.s...@streamnative.io.invalid> wrote:
> > > > > > > > > >
> > > > > > > > > > Hi, I meant
> > > > > > > > > >
> > > > > > > > > > What if the producer application jvm restarts in the middle
> > > of
> > > > > > > chunking
> > > > > > > > > and
> > > > > > > > > > resends the message chunks from the beginning with the
> > > previous
> > > > > > > sequence
> > > > > > > > > id?
> > > > > > > > > >
> > > > > > > > > >
> > > > > > > > > >
> > > > > > > > > > On Fri, Aug 25, 2023 at 5:15 PM Xiangying Meng <
> > > > > xiangy...@apache.org
> > > > > > > >
> > > > > > > > > wrote:
> > > > > > > > > >
> > > > > > > > > > > Hi Heesung
> > > > > > > > > > >
> > > > > > > > > > > It is a good idea to cover this incompatibility case if
> > > > > > > > > > > the
> > > > > > > producer
> > > > > > > > > > > splits the chunk message again when retrying.
> > > > > > > > > > >
> > > > > > > > > > > But in fact, the producer only resents the chunks that are
> > > > > > > assembled
> > > > > > > > > > > to `OpSendMsg` instead of splitting the chunk message
> > > again.
> > > > > > > > > > > So, there is no incompatibility issue of resenting the
> > > chunk
> > > > > > > message
> > > > > > > > > > > by splitting the chunk message again.
> > > > > > > > > > >
> > > > > > > > > > > The logic of sending chunk messages can be found here:
> > > > > > > > > > >
> > > > > > > > > > >
> > > > > > > > >
> > > > > > >
> > > > >
> > > https://github.com/apache/pulsar/blob/e0c481e5f8d7fa5534d3327785928a234376789e/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L533
> > > > > > > > > > >
> > > > > > > > > > > The logic of resending the message can be found here:
> > > > > > > > > > >
> > > > > > > > > > >
> > > > > > > > >
> > > > > > >
> > > > >
> > > https://github.com/apache/pulsar/blob/e0c481e5f8d7fa5534d3327785928a234376789e/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L1892
> > > > > > > > > > >
> > > > > > > > > > > BR,
> > > > > > > > > > > Xiangying
> > > > > > > > > > >
> > > > > > > > > > >
> > > > > > > > > > >
> > > > > > > > > > >
> > > > > > > > > > >
> > > > > > > > > > >
> > > > > > > > > > >
> > > > > > > > > > > On Sat, Aug 26, 2023 at 2:24 AM Heesung Sohn
> > > > > > > > > > > <heesung.s...@streamnative.io.invalid> wrote:
> > > > > > > > > > > >
> > > > > > > > > > > > >> I think brokers can track the last
> > > chunkMaxMessageSize for
> > > > > > > each
> > > > > > > > > > > producer.
> > > > > > > > > > > >
> > > > > > > > > > > > > Using different chunkMaxMessageSize is just one of the
> > > > > > > aspects. In
> > > > > > > > > > > > PIP-132 [0], we have included the message metadata size
> > > when
> > > > > > > checking
> > > > > > > > > > > > maxMessageSize.The message metadata can be changed after
> > > > > > > splitting
> > > > > > > > > the
> > > > > > > > > > > > chunks. We are still uncertain about the way the chunked
> > > > > message
> > > > > > > is
> > > > > > > > > > > split,
> > > > > > > > > > > > even using the same ss chunkMaxMessageSize.
> > > > > > > > > > > >
> > > > > > > > > > > > This sounds like we need to revisit chunking uuid logic.
> > > > > > > > > > > > Like I commented here,
> > > > > > > > > > > >
> > > > > https://github.com/apache/pulsar/pull/20948/files#r1305997883
> > > > > > > > > > > > Why don't we add a chunk session id suffix to identify
> > > the
> > > > > > > ongoing
> > > > > > > > > > > chunking
> > > > > > > > > > > > uniquely?
> > > > > > > > > > > >
> > > > > > > > > > > > Currently,
> > > > > > > > > > > >
> > > > > > > > > > > > chunking uuid = producer + sequence_id
> > > > > > > > > > > >
> > > > > > > > > > > > Proposal
> > > > > > > > > > > > chunking uuid = producer + sequence_id +
> > > chunkingSessionId
> > > > > > > > > > > >
> > > > > > > > > > > > * chunkingSessionId could be a timestamp when the
> > > chunking
> > > > > > > started.
> > > > > > > > > > > >
> > > > > > > > > > > >
> > > > > > > > > > > >
> > > > > > > > > > > > On Fri, Aug 25, 2023 at 6:02 AM Xiangying Meng <
> > > > > > > xiangy...@apache.org
> > > > > > > > > >
> > > > > > > > > > > wrote:
> > > > > > > > > > > >
> > > > > > > > > > > > > Hi Zike,
> > > > > > > > > > > > >
> > > > > > > > > > > > > >How would this happen to get two duplicated and
> > > > > consecutive
> > > > > > > > > ChunkID-1
> > > > > > > > > > > > > >messages? The producer should guarantee to retry the
> > > whole
> > > > > > > chunked
> > > > > > > > > > > > > >messages instead of some parts of the chunks.
> > > > > > > > > > > > >
> > > > > > > > > > > > > If the producer guarantees to retry the whole chunked
> > > > > messages
> > > > > > > > > instead
> > > > > > > > > > > > > of some parts of the chunks,
> > > > > > > > > > > > > Why doesn't the bug of the producer retry chunk
> > > messages in
> > > > > > > the PR
> > > > > > > > > [0]
> > > > > > > > > > > > > appear?
> > > > > > > > > > > > > And why do you need to set `chunkId` in
> > > `op.rePopulate`?
> > > > > > > > > > > > > It will be rested when split chunk messages again if
> > > the
> > > > > > > producer
> > > > > > > > > > > > > guarantees to retry the whole chunked messages.
> > > > > > > > > > > > > ```
> > > > > > > > > > > > > final MessageMetadata finalMsgMetadata = msgMetadata;
> > > > > > > > > > > > > op.rePopulate = () -> {
> > > > > > > > > > > > > if (msgMetadata.hasChunkId()) {
> > > > > > > > > > > > > // The message metadata is shared between all chunks
> > > in a
> > > > > large
> > > > > > > > > message
> > > > > > > > > > > > > // We need to reset the chunk id for each call of this
> > > > > method
> > > > > > > > > > > > > // It's safe to do that because there is only 1 thread
> > > to
> > > > > > > > > manipulate
> > > > > > > > > > > > > this message metadata
> > > > > > > > > > > > > finalMsgMetadata.setChunkId(chunkId);
> > > > > > > > > > > > > }
> > > > > > > > > > > > > op.cmd = sendMessage(producerId, sequenceId,
> > > numMessages,
> > > > > > > > > messageId,
> > > > > > > > > > > > > finalMsgMetadata,
> > > > > > > > > > > > > encryptedPayload);
> > > > > > > > > > > > > };
> > > > > > > > > > > > >
> > > > > > > > > > > > > ```
> > > > > > > > > > > > >
> > > > > > > > > > > > > >> But chunks 1, 2, 3, and 4 are still persisted in
> > > > > > > > > > > > > >> the
> > > > > topic.
> > > > > > > > > > > > > >
> > > > > > > > > > > > > >I think it's OK to persist them all on the topic. Is
> > > > > there any
> > > > > > > > > issue
> > > > > > > > > > > > > >with doing that?
> > > > > > > > > > > > >
> > > > > > > > > > > > > This is just one scenario. Whether only check the
> > > sequence
> > > > > ID
> > > > > > > of
> > > > > > > > > the
> > > > > > > > > > > > > first chunk (as I used in PR[1]) or check the sequence
> > > ID
> > > > > of
> > > > > > > the
> > > > > > > > > last
> > > > > > > > > > > > > chunk (as you suggested), in reality, neither of these
> > > > > methods
> > > > > > > can
> > > > > > > > > > > > > deduplicate chunks on the broker side because the
> > > broker
> > > > > cannot
> > > > > > > > > know
> > > > > > > > > > > > > the chunk ID of the previous message.
> > > > > > > > > > > > >
> > > > > > > > > > > > > However, if combined with the optimization of
> > > consumer-side
> > > > > > > logic,
> > > > > > > > > > > > > end-to-end deduplication can be completed.
> > > > > > > > > > > > > This is also a less-than-perfect solution I mentioned
> > > in my
> > > > > > > first
> > > > > > > > > > > > > email and implemented in PR[1].
> > > > > > > > > > > > >
> > > > > > > > > > > > > The reason I propose this proposal is not to solve the
> > > > > > > end-to-end
> > > > > > > > > > > > > deduplication of chunk messages between producers and
> > > > > > > consumers.
> > > > > > > > > That
> > > > > > > > > > > > > aspect has essentially been addressed in PR[1] and is
> > > still
> > > > > > > > > undergoing
> > > > > > > > > > > > > review.
> > > > > > > > > > > > >
> > > > > > > > > > > > > This proposal aims to ensure that no corrupt data
> > > exists
> > > > > > > within the
> > > > > > > > > > > > > topic, as our data might be offloaded and used
> > > elsewhere in
> > > > > > > > > scenarios
> > > > > > > > > > > > > where consumer logic is not optimized.
> > > > > > > > > > > > >
> > > > > > > > > > > > > BR,
> > > > > > > > > > > > > Xiangying
> > > > > > > > > > > > >
> > > > > > > > > > > > > [0] https://github.com/apache/pulsar/pull/21048
> > > > > > > > > > > > > [1] https://github.com/apache/pulsar/pull/20948
> > > > > > > > > > > > >
> > > > > > > > > > > > > On Fri, Aug 25, 2023 at 5:18 PM Zike Yang <
> > > z...@apache.org
> > > > > >
> > > > > > > wrote:
> > > > > > > > > > > > > >
> > > > > > > > > > > > > > HI xiangying
> > > > > > > > > > > > > >
> > > > > > > > > > > > > > > The rewind operation is seen in the test log.
> > > > > > > > > > > > > >
> > > > > > > > > > > > > > That seems weird. Not sure if this rewind is related
> > > to
> > > > > the
> > > > > > > chunk
> > > > > > > > > > > > > consuming.
> > > > > > > > > > > > > >
> > > > > > > > > > > > > > > 1. SequenceID: 0, ChunkID: 0
> > > > > > > > > > > > > > 2. SequenceID: 0, ChunkID: 1
> > > > > > > > > > > > > > 3. SequenceID: 0, ChunkID: 1
> > > > > > > > > > > > > > 4. SequenceID: 0, ChunkID: 2
> > > > > > > > > > > > > > Such four chunks cannot be processed correctly by
> > > > > > > > > > > > > > the
> > > > > > > consumer.
> > > > > > > > > > > > > >
> > > > > > > > > > > > > > How would this happen to get two duplicated and
> > > > > consecutive
> > > > > > > > > ChunkID-1
> > > > > > > > > > > > > > messages? The producer should guarantee to retry the
> > > > > whole
> > > > > > > > > chunked
> > > > > > > > > > > > > > messages instead of some parts of the chunks.
> > > > > > > > > > > > > >
> > > > > > > > > > > > > > > But chunks 1, 2, 3, and 4 are still persisted in
> > > the
> > > > > topic.
> > > > > > > > > > > > > >
> > > > > > > > > > > > > > I think it's OK to persist them all in the topic. Is
> > > > > there
> > > > > > > any
> > > > > > > > > issue
> > > > > > > > > > > > > > with doing that?
> > > > > > > > > > > > > >
> > > > > > > > > > > > > > > There is another point. The resend of the chunk
> > > message
> > > > > > > has a
> > > > > > > > > bug
> > > > > > > > > > > that
> > > > > > > > > > > > > > I shared with you, and you fixed in PR [0]. It will
> > > make
> > > > > this
> > > > > > > > > case
> > > > > > > > > > > > > > happen in another way.
> > > > > > > > > > > > > >
> > > > > > > > > > > > > > If the user sets the sequence ID manually, the case
> > > > > could be
> > > > > > > > > > > reproduced.
> > > > > > > > > > > > > >
> > > > > > > > > > > > > > BR,
> > > > > > > > > > > > > > Zike Yang
> > > > > > > > > > > > > >
> > > > > > > > > > > > > > On Thu, Aug 24, 2023 at 8:48 PM Xiangying Meng <
> > > > > > > > > xiangy...@apache.org
> > > > > > > > > > > >
> > > > > > > > > > > > > wrote:
> > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > >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
> > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > Hi Zike
> > > > > > > > > > > > > > > There is another point. The resend of the chunk
> > > message
> > > > > > > has a
> > > > > > > > > bug
> > > > > > > > > > > that
> > > > > > > > > > > > > > > I shared with you, and you fixed in PR [0]. It
> > > > > > > > > > > > > > > will
> > > > > make
> > > > > > > this
> > > > > > > > > case
> > > > > > > > > > > > > > > happen in another way.
> > > > > > > > > > > > > > > Sample description for the bug:
> > > > > > > > > > > > > > > Because the chunk message uses the same message
> > > > > metadata,
> > > > > > > if
> > > > > > > > > the
> > > > > > > > > > > chunk
> > > > > > > > > > > > > > > is not sent out immediately. Then, when resending,
> > > all
> > > > > > > chunks
> > > > > > > > > of
> > > > > > > > > > > the
> > > > > > > > > > > > > > > same chunk message use the chunk ID of the last
> > > chunk.
> > > > > > > > > > > > > > > In this case, It should happen as:
> > > > > > > > > > > > > > > 1. SequenceID: 0, ChunkID: 0 (Put op1 into
> > > > > > > `pendingMessages`
> > > > > > > > > and
> > > > > > > > > > > send)
> > > > > > > > > > > > > > > 2. SequenceID: 0, ChunkID: 1 (Put op2 into
> > > > > > > `pendingMessages`
> > > > > > > > > and
> > > > > > > > > > > send)
> > > > > > > > > > > > > > > 3. SequenceID: 0, ChunkID: 2 -> (Put op3 into
> > > > > > > > > `pendingMessages`)
> > > > > > > > > > > > > > > 4. SequenceID: 0, ChunkID: 2 -> (Resend op1)
> > > > > > > > > > > > > > > 5. SequenceID: 0, ChunkID: 2 -> (Resend op2)
> > > > > > > > > > > > > > > 6. SequenceID: 0, ChunkID: 2 -> (Send op3)
> > > > > > > > > > > > > > >
> > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > BR,
> > > > > > > > > > > > > > > Xiangying
> > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > [0] - https://github.com/apache/pulsar/pull/21048
> > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > On Thu, Aug 24, 2023 at 8:09 PM Xiangying Meng <
> > > > > > > > > > > xiangy...@apache.org>
> > > > > > > > > > > > > wrote:
> > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > >> This solution also cannot solve the
> > > out-of-order
> > > > > > > messages
> > > > > > > > > > > inside
> > > > > > > > > > > > > the
> > > > > > > > > > > > > > > > >>chunks. For example, the above five messages
> > > will
> > > > > > > still be
> > > > > > > > > > > > > persisted.
> > > > > > > > > > > > > > > > >The consumer already handles this case. The
> > > above 5
> > > > > > > messages
> > > > > > > > > > > will
> > > > > > > > > > > > > all
> > > > > > > > > > > > > > > > >be persisted but the consumer will skip message
> > > 1
> > > > > and 2.
> > > > > > > > > > > > > > > > >For messages 3, 4, and 5. The producer can
> > > guarantee
> > > > > > > these
> > > > > > > > > > > chunks
> > > > > > > > > > > > > are in order.
> > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > The rewind operation is seen in the test log.
> > > Every
> > > > > time
> > > > > > > an
> > > > > > > > > > > incorrect
> > > > > > > > > > > > > > > > chunk message is received, it will rewind, and
> > > the
> > > > > code
> > > > > > > has
> > > > > > > > > yet
> > > > > > > > > > > to be
> > > > > > > > > > > > > > > > studied in depth.
> > > > > > > > > > > > > > > > If it does not call rewind, then this case is
> > > > > considered
> > > > > > > a
> > > > > > > > > > > workable
> > > > > > > > > > > > > > > > case. Let's look at another case.
> > > > > > > > > > > > > > > > 1. SequenceID: 0, ChunkID: 0
> > > > > > > > > > > > > > > > 2. SequenceID: 0, ChunkID: 1
> > > > > > > > > > > > > > > > 3. SequenceID: 0, ChunkID: 1
> > > > > > > > > > > > > > > > 4. SequenceID: 0, ChunkID: 2
> > > > > > > > > > > > > > > > Such four chunks cannot be processed correctly
> > > by the
> > > > > > > > > consumer.
> > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > In fact, this solution is my original idea. The
> > > PR I
> > > > > > > > > mentioned
> > > > > > > > > > > in the
> > > > > > > > > > > > > > > > first email above uses a similar solution and
> > > > > modifies
> > > > > > > the
> > > > > > > > > logic
> > > > > > > > > > > on
> > > > > > > > > > > > > > > > the consumer side.
> > > > > > > > > > > > > > > > Also, as I mentioned in the first email, this
> > > > > solution
> > > > > > > can
> > > > > > > > > only
> > > > > > > > > > > solve
> > > > > > > > > > > > > > > > the problem of end-to-end duplication. But
> > > chunks 1,
> > > > > 2,
> > > > > > > 3,
> > > > > > > > > and 4
> > > > > > > > > > > are
> > > > > > > > > > > > > > > > still persisted in the topic.
> > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > On Thu, Aug 24, 2023 at 3:00 PM Zike Yang <
> > > > > > > z...@apache.org>
> > > > > > > > > > > wrote:
> > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > Hi Heesung,
> > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > > 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.
> > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > With this simple solution, I think we don't
> > > need to
> > > > > > > track
> > > > > > > > > the
> > > > > > > > > > > > > > > > > (sequenceID, chunkID) on the broker side at
> > > all.
> > > > > The
> > > > > > > broker
> > > > > > > > > > > just
> > > > > > > > > > > > > needs
> > > > > > > > > > > > > > > > > to apply the deduplication logic to the last
> > > chunk
> > > > > > > instead
> > > > > > > > > of
> > > > > > > > > > > all
> > > > > > > > > > > > > > > > > previous chunks. This PIP actually could do
> > > that,
> > > > > but
> > > > > > > it
> > > > > > > > > will
> > > > > > > > > > > > > > > > > introduce a new data format and compatibility
> > > > > issue.
> > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > > This is still a behavior change(deduping
> > > chunk
> > > > > > > messages
> > > > > > > > > on
> > > > > > > > > > > the
> > > > > > > > > > > > > broker),
> > > > > > > > > > > > > > > > > and I believe we need to discuss this addition
> > > as a
> > > > > > > PIP.
> > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > Actually, we didn't specifically state the
> > > deduping
> > > > > > > chunk
> > > > > > > > > > > message
> > > > > > > > > > > > > > > > > behavior before. The chunked message should be
> > > > > equally
> > > > > > > > > > > applicable
> > > > > > > > > > > > > to
> > > > > > > > > > > > > > > > > the de-duplication logic as a regular message.
> > > > > > > Therefore, I
> > > > > > > > > > > think
> > > > > > > > > > > > > it
> > > > > > > > > > > > > > > > > should be considered as a bug fix. But if this
> > > FIX
> > > > > is
> > > > > > > worth
> > > > > > > > > > > > > discussing
> > > > > > > > > > > > > > > > > in depth. I have no objection to it being a
> > > > > > > > > > > > > > > > > new
> > > > > PIP.
> > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > > I think brokers can track the last
> > > > > > > chunkMaxMessageSize
> > > > > > > > > for
> > > > > > > > > > > > > > > > > each producer.
> > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > Using different chunkMaxMessageSize is just
> > > one of
> > > > > the
> > > > > > > > > > > aspects. In
> > > > > > > > > > > > > > > > > PIP-132 [0], we have included the message
> > > metadata
> > > > > size
> > > > > > > > > when
> > > > > > > > > > > > > checking
> > > > > > > > > > > > > > > > > maxMessageSize.
> > > > > > > > > > > > > > > > > The message metadata can be changed after
> > > > > splitting the
> > > > > > > > > > > chunks. We
> > > > > > > > > > > > > are
> > > > > > > > > > > > > > > > > still uncertain about the way the chunked
> > > message
> > > > > is
> > > > > > > split,
> > > > > > > > > > > even
> > > > > > > > > > > > > using
> > > > > > > > > > > > > > > > > the same ss 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.
> > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > Regarding this, it seems like we are
> > > implementing
> > > > > > > dynamic
> > > > > > > > > > > > > > > > > configuration for the chunkMaxMessageSize. I'm
> > > > > afraid
> > > > > > > that
> > > > > > > > > this
> > > > > > > > > > > > > would
> > > > > > > > > > > > > > > > > change the expected behavior and introduce
> > > > > > > > > > > > > > > > > more
> > > > > > > complexity
> > > > > > > > > to
> > > > > > > > > > > this
> > > > > > > > > > > > > > > > > configuration.
> > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > [0]
> > > https://github.com/apache/pulsar/pull/14007
> > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > BR,
> > > > > > > > > > > > > > > > > Zike Yang
> > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > On Thu, Aug 24, 2023 at 2:21 PM Zike Yang <
> > > > > > > z...@apache.org
> > > > > > > > > >
> > > > > > > > > > > wrote:
> > > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > > Hi, xiangying
> > > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > > > 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.
> > > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > > Could you point out where the implementation
> > > for
> > > > > > > this?
> > > > > > > > > From
> > > > > > > > > > > my
> > > > > > > > > > > > > > > > > > understanding, there should not be any
> > > > > > > > > > > > > > > > > > rewind
> > > > > > > operation
> > > > > > > > > for
> > > > > > > > > > > the
> > > > > > > > > > > > > > > > > > chunking feature. You can check more detail
> > > here:
> > > > > > > > > > > > > > > > > >
> > > > > > > > > > > > >
> > > > > > > > > > >
> > > > > > > > >
> > > > > > >
> > > > >
> > > https://streamnative.io/blog/deep-dive-into-message-chunking-in-pulsar#how-message-chunking-is-implemented
> > > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > > > This solution also cannot solve the
> > > > > out-of-order
> > > > > > > > > messages
> > > > > > > > > > > > > inside the
> > > > > > > > > > > > > > > > > > chunks. For example, the above five messages
> > > will
> > > > > > > still
> > > > > > > > > be
> > > > > > > > > > > > > persisted.
> > > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > > The consumer already handles this case. The
> > > > > above 5
> > > > > > > > > messages
> > > > > > > > > > > > > will all
> > > > > > > > > > > > > > > > > > be persisted but the consumer will skip
> > > message 1
> > > > > > > and 2.
> > > > > > > > > > > > > > > > > > For messages 3, 4, and 5. The producer can
> > > > > guarantee
> > > > > > > > > these
> > > > > > > > > > > > > chunks are in order.
> > > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > > BR,
> > > > > > > > > > > > > > > > > > Zike Yang
> > > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > > On Thu, Aug 24, 2023 at 11:48 AM Yubiao Feng
> > > > > > > > > > > > > > > > > > <yubiao.f...@streamnative.io.invalid> wrote:
> > > > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > > > > 1. SequenceID: 0, ChunkID: 0
> > > > > > > > > > > > > > > > > > > > 2. SequenceID: 0, ChunkID: 1
> > > > > > > > > > > > > > > > > > > > 3. SequenceID: 0, ChunkID: 0
> > > > > > > > > > > > > > > > > > > > 4. SequenceID: 0, ChunkID: 1
> > > > > > > > > > > > > > > > > > > > 5. SequenceID: 0, ChunkID: 2
> > > > > > > > > > > > > > > > > > > > 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.
> > > > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > > > Good point.
> > > > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > > > Thanks
> > > > > > > > > > > > > > > > > > > Yubiao Feng
> > > > > > > > > > > > > > > > > > >
> > > > > > > > > > > > > > > > > > > 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
> > > > > > > > > > > > > > > > > > > >
> > > > > > > > > > > > >
> > > > > > > > > > >
> > > > > > > > >
> > > > > > >
> > > > >
> > >
