rdhabalia commented on a change in pull request #12171:
URL: https://github.com/apache/pulsar/pull/12171#discussion_r717152592
##########
File path:
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
##########
@@ -1000,6 +1004,19 @@ void ackReceived(ClientCnx cnx, long sequenceId, long
highestSequenceId, long le
OpSendMsg finalOp = op;
LAST_SEQ_ID_PUBLISHED_UPDATER.getAndUpdate(this, last ->
Math.max(last, getHighestSequenceId(finalOp)));
op.setMessageId(ledgerId, entryId, partitionIndex);
+ if (op.totalChunks > 1) {
+ if (op.chunkId == 0) {
+ chunkMessageIds.put(op.msg.getMessageBuilder().getUuid(),
+ new MessageIdImpl(ledgerId, entryId, partitionIndex));
+ } else if (op.chunkId == op.totalChunks - 1) {
+ MessageIdImpl firstChunkMsgId =
chunkMessageIds.get(op.msg.getMessageBuilder().getUuid());
+ if (firstChunkMsgId != null) {
+ op.setMessageId(firstChunkMsgId.ledgerId,
firstChunkMsgId.entryId, firstChunkMsgId.partitionIndex);
Review comment:
this is the challenge here. last chunk is already published and persist
at server side. this is at `ackReceived` method, means you have received the
ack for already published message and updating messageId will not reflect any
change to message which is already persist at server side.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
