[Issue 5476]Fix message deduplicate issue while using external sequence id with batch produce

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java

@@ -132,6 +132,25 @@ public boolean equals(Object obj) {
     }
 
     public void publishMessage(long producerId, long sequenceId, ByteBuf headersAndPayload, long batchSize) {
+        beforePublish(producerId, sequenceId, headersAndPayload, batchSize);
+        publishMessageToTopic(headersAndPayload, sequenceId, batchSize);
+    }
+
+    public void publishMessage(long producerId, long lowestSequenceId, long highestSequenceId,
+           ByteBuf headersAndPayload, long batchSize) {
+        if (lowestSequenceId > highestSequenceId) {
+            cnx.ctx().channel().eventLoop().execute(() -> {
+                cnx.ctx().writeAndFlush(Commands.newSendError(producerId, highestSequenceId, ServerError.MetadataError,
+                        "Invalid lowest or highest sequence id"));
+                cnx.completedSendOperation(isNonPersistentTopic);
+            });
+            return;
+        }
+        beforePublish(producerId, highestSequenceId, headersAndPayload, batchSize);
+        publishMessageToTopic(headersAndPayload, lowestSequenceId, highestSequenceId, batchSize);
+    }
+
+    public void beforePublish(long producerId, long sequenceId, ByteBuf headersAndPayload, long batchSize) {
         if (isClosed) {
             cnx.ctx().channel().eventLoop().execute(() -> {
                 cnx.ctx().writeAndFlush(Commands.newSendError(producerId, sequenceId, ServerError.PersistenceError,
@@ -170,11 +189,20 @@ public void publishMessage(long producerId, long sequenceId, ByteBuf headersAndP
         }
 
         startPublishOperation((int) batchSize, headersAndPayload.readableBytes());
+    }
+
+    private void publishMessageToTopic(ByteBuf headersAndPayload, long sequenceId, long batchSize) {
         topic.publishMessage(headersAndPayload,
                 MessagePublishContext.get(this, sequenceId, msgIn, headersAndPayload.readableBytes(), batchSize,
                         System.nanoTime()));
     }
 
+    private void publishMessageToTopic(ByteBuf headersAndPayload, long lowestSequenceId, long highestSequenceId, long batchSize) {
+        topic.publishMessage(headersAndPayload,
+                MessagePublishContext.get(this, lowestSequenceId, highestSequenceId, msgIn, headersAndPayload.readableBytes(), batchSize,
+                        System.nanoTime()));
+    }
+
     private boolean verifyChecksum(ByteBuf headersAndPayload) {
         if (hasChecksum(headersAndPayload)) {
             int readerIndex = headersAndPayload.readerIndex();
@@ -257,6 +285,9 @@ private static final class MessagePublishContext implements PublishContext, Runn
         private String originalProducerName;
         private long originalSequenceId;
 
+        private long highestSequenceId;
+        private long originalHighestSequenceId;
+
         public String getProducerName() {
             return producer.getProducerName();
         }
@@ -265,6 +296,11 @@ public long getSequenceId() {
             return sequenceId;
         }
 
+        @Override
+        public long getHighestSequenceId() {
+            return highestSequenceId;
+        }
+
         @Override
         public void setOriginalProducerName(String originalProducerName) {
             this.originalProducerName = originalProducerName;
@@ -285,6 +321,16 @@ public long getOriginalSequenceId() {
             return originalSequenceId;
         }
 
+        @Override
+        public void setOriginalHighestSequenceId(long originalHighestSequenceId) {
+            this.originalHighestSequenceId = originalHighestSequenceId;
+        }
+
+        @Override
+        public long getOriginalHighestSequenceId() {
+            return originalHighestSequenceId;
+        }
+
         /**
          * Executed from managed ledger thread when the message is persisted
          */
@@ -298,7 +344,8 @@ public void completed(Exception exception, long ledgerId, long entryId) {
                     if (!(exception instanceof TopicClosedException)) {
                         // For TopicClosed exception there's no need to send explicit error, since the client was
                         // already notified
-                        producer.cnx.ctx().writeAndFlush(Commands.newSendError(producer.producerId, sequenceId,
+                        long callBackSequenceId = Math.max(highestSequenceId, sequenceId);
+                        producer.cnx.ctx().writeAndFlush(Commands.newSendError(producer.producerId, callBackSequenceId,
                                 serverError, exception.getMessage()));
                     }
                     producer.cnx.completedSendOperation(producer.isNonPersistentTopic);
@@ -330,8 +377,9 @@ public void run() {
             // stats
             rateIn.recordMultipleEvents(batchSize, msgSize);
             producer.topic.recordAddLatency(System.nanoTime() - startTimeNs, TimeUnit.NANOSECONDS);
+            long callBackSequenceId = Math.max(highestSequenceId, sequenceId);
             producer.cnx.ctx().writeAndFlush(
-                    Commands.newSendReceipt(producer.producerId, sequenceId, ledgerId, entryId),
+                    Commands.newSendReceipt(producer.producerId, callBackSequenceId, ledgerId, entryId),
                     producer.cnx.ctx().voidPromise());
             producer.cnx.completedSendOperation(producer.isNonPersistentTopic);
             producer.publishOperationCompleted();
@@ -347,7 +395,22 @@ static MessagePublishContext get(Producer producer, long sequenceId, Rate rateIn
             callback.msgSize = msgSize;
             callback.batchSize = batchSize;
             callback.originalProducerName = null;
-            callback.originalSequenceId = -1;
+            callback.originalSequenceId = -1L;
+            callback.startTimeNs = startTimeNs;
+            return callback;
+        }
+
+        static MessagePublishContext get(Producer producer, long lowestSequenceId, long highestSequenceId, Rate rateIn,
+                 int msgSize, long batchSize, long startTimeNs) {
+            MessagePublishContext callback = RECYCLER.get();
+            callback.producer = producer;
+            callback.sequenceId = lowestSequenceId;
+            callback.highestSequenceId = highestSequenceId;
+            callback.rateIn = rateIn;
+            callback.msgSize = msgSize;
+            callback.batchSize = batchSize;
+            callback.originalProducerName = null;
+            callback.originalSequenceId = -1L;
             callback.startTimeNs = startTimeNs;
             return callback;
         }
@@ -366,13 +429,16 @@ protected MessagePublishContext newObject(Recycler.Handle<MessagePublishContext>
 
         public void recycle() {
             producer = null;
-            sequenceId = -1;
+            sequenceId = -1L;
+            highestSequenceId = -1L;
+            originalSequenceId = -1L;
+            originalHighestSequenceId = -1L;
             rateIn = null;
             msgSize = 0;
-            ledgerId = -1;
-            entryId = -1;
-            batchSize = 0;
-            startTimeNs = -1;
+            ledgerId = -1L;
+            entryId = -1L;
+            batchSize = 0L;
+            startTimeNs = -1L;
             recyclerHandle.recycle(this);
         }
     }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -1051,7 +1051,12 @@ protected void handleSend(CommandSend send, ByteBuf headersAndPayload) {
         startSendOperation(producer);
 
         // Persist the message
-        producer.publishMessage(send.getProducerId(), send.getSequenceId(), headersAndPayload, send.getNumMessages());
+        if (send.hasHighestSequenceId() && send.getSequenceId() <= send.getHighestSequenceId()) {
+            producer.publishMessage(send.getProducerId(), send.getSequenceId(), send.getHighestSequenceId(),
+                    headersAndPayload, send.getNumMessages());
+        } else {
+            producer.publishMessage(send.getProducerId(), send.getSequenceId(), headersAndPayload, send.getNumMessages());
+        }
     }
 
     private void printSendCommandDebug(CommandSend send, ByteBuf headersAndPayload) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java

@@ -53,7 +53,7 @@ default String getProducerName() {
         }
 
         default long getSequenceId() {
-            return -1;
+            return -1L;
         }
 
         default void setOriginalProducerName(String originalProducerName) {
@@ -73,10 +73,22 @@ default String getOriginalProducerName() {
         }
 
         default long getOriginalSequenceId() {
-            return -1;
+            return -1L;
         }
 
         void completed(Exception e, long ledgerId, long entryId);
+
+        default long getHighestSequenceId() {
+            return  -1L;
+        }
+
+        default void setOriginalHighestSequenceId(long originalHighestSequenceId) {
+
+        }
+
+        default long getOriginalHighestSequenceId() {
+            return  -1L;
+        }
     }
 
     void publishMessage(ByteBuf headersAndPayload, PublishContext callback);

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java

@@ -160,7 +160,7 @@ public void readEntriesComplete(List<Entry> entries, Object ctx) {
                     MessageMetadata md = Commands.parseMessageMetadata(messageMetadataAndPayload);
 
                     String producerName = md.getProducerName();
-                    long sequenceId = md.getSequenceId();
+                    long sequenceId = Math.max(md.getHighestSequenceId(), md.getSequenceId());
                     highestSequencedPushed.put(producerName, sequenceId);
                     highestSequencedPersisted.put(producerName, sequenceId);
 
@@ -283,15 +283,18 @@ public MessageDupStatus isDuplicate(PublishContext publishContext, ByteBuf heade
 
         String producerName = publishContext.getProducerName();
         long sequenceId = publishContext.getSequenceId();
+        long highestSequenceId = Math.max(publishContext.getHighestSequenceId(), sequenceId);
         if (producerName.startsWith(replicatorPrefix)) {
             // Message is coming from replication, we need to use the original producer name and sequence id
             // for the purpose of deduplication and not rely on the "replicator" name.
             int readerIndex = headersAndPayload.readerIndex();
             MessageMetadata md = Commands.parseMessageMetadata(headersAndPayload);
             producerName = md.getProducerName();
             sequenceId = md.getSequenceId();
+            highestSequenceId = Math.max(md.getHighestSequenceId(), sequenceId);
             publishContext.setOriginalProducerName(producerName);
             publishContext.setOriginalSequenceId(sequenceId);
+            publishContext.setOriginalHighestSequenceId(highestSequenceId);
             headersAndPayload.readerIndex(readerIndex);
             md.recycle();
         }
@@ -317,8 +320,7 @@ public MessageDupStatus isDuplicate(PublishContext publishContext, ByteBuf heade
                     return MessageDupStatus.Unknown;
                 }
             }
-
-            highestSequencedPushed.put(producerName, sequenceId);
+            highestSequencedPushed.put(producerName, highestSequenceId);
         }
         return MessageDupStatus.NotDup;
     }
@@ -333,13 +335,15 @@ public void recordMessagePersisted(PublishContext publishContext, PositionImpl p
 
         String producerName = publishContext.getProducerName();
         long sequenceId = publishContext.getSequenceId();
+        long highestSequenceId = publishContext.getHighestSequenceId();
         if (publishContext.getOriginalProducerName() != null) {
             // In case of replicated messages, this will be different from the current replicator producer name
             producerName = publishContext.getOriginalProducerName();
             sequenceId = publishContext.getOriginalSequenceId();
+            highestSequenceId = publishContext.getOriginalHighestSequenceId();
         }
 
-        highestSequencedPersisted.put(producerName, sequenceId);
+        highestSequencedPersisted.put(producerName, Math.max(highestSequenceId, sequenceId));
         if (++snapshotCounter >= snapshotInterval) {
             snapshotCounter = 0;
             takeSnapshot(position);

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java

@@ -50,6 +50,7 @@
 import org.apache.pulsar.broker.service.Replicator;
 import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.Type;
 import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.impl.Backoff;
 import org.apache.pulsar.client.impl.MessageImpl;
 import org.apache.pulsar.client.impl.ProducerImpl;
@@ -377,7 +378,7 @@ private static final class ProducerSendCallback implements SendCallback {
 
         @Override
         public void sendComplete(Exception exception) {
-            if (exception != null) {
+            if (exception != null && !(exception instanceof PulsarClientException.InvalidMessageException)) {
                 log.error("[{}][{} -> {}] Error producing on remote broker", replicator.topicName,
                         replicator.localCluster, replicator.remoteCluster, exception);
                 // cursor shoud be rewinded since it was incremented when readMoreEntries

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/MessageDuplicationTest.java

@@ -45,6 +45,7 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertTrue;
 
 @Slf4j
@@ -125,6 +126,24 @@ public void testIsDuplicate() {
         lastSequenceIdPushed = messageDeduplication.highestSequencedPushed.get(producerName1);
         assertTrue(lastSequenceIdPushed != null);
         assertEquals(lastSequenceIdPushed.longValue(), 5);
+
+        // update highest sequence persisted
+        messageDeduplication.highestSequencedPushed.put(producerName1, 0L);
+        messageDeduplication.highestSequencedPersisted.put(producerName1, 0L);
+        byteBuf1 = getMessage(producerName1, 0);
+        publishContext1 = getPublishContext(producerName1, 1, 5);
+        status = messageDeduplication.isDuplicate(publishContext1, byteBuf1);
+        assertEquals(status, MessageDeduplication.MessageDupStatus.NotDup);
+        lastSequenceIdPushed = messageDeduplication.highestSequencedPushed.get(producerName1);
+        assertNotNull(lastSequenceIdPushed);
+        assertEquals(lastSequenceIdPushed.longValue(), 5);
+
+        publishContext1 = getPublishContext(producerName1, 4, 8);
+        status = messageDeduplication.isDuplicate(publishContext1, byteBuf1);
+        assertEquals(status, MessageDeduplication.MessageDupStatus.Unknown);
+        lastSequenceIdPushed = messageDeduplication.highestSequencedPushed.get(producerName1);
+        assertNotNull(lastSequenceIdPushed);
+        assertEquals(lastSequenceIdPushed.longValue(), 5);
     }
 
     @Test
@@ -319,4 +338,28 @@ public void completed(Exception e, long ledgerId, long entryId) {
             }
         });
     }
+
+    public Topic.PublishContext getPublishContext(String producerName, long seqId, long lastSequenceId) {
+        return spy(new Topic.PublishContext() {
+            @Override
+            public String getProducerName() {
+                return producerName;
+            }
+
+            @Override
+            public long getSequenceId() {
+                return seqId;
+            }
+
+            @Override
+            public long getHighestSequenceId() {
+                return lastSequenceId;
+            }
+
+            @Override
+            public void completed(Exception e, long ledgerId, long entryId) {
+
+            }
+        });
+    }
 }