Add highest sequenceId for CommandSendReceipt.

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

@@ -377,9 +377,8 @@ 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, callBackSequenceId, ledgerId, entryId),
+                    Commands.newSendReceipt(producer.producerId, sequenceId, highestSequenceId, ledgerId, entryId),
                     producer.cnx.ctx().voidPromise());
             producer.cnx.completedSendOperation(producer.isNonPersistentTopic);
             producer.publishOperationCompleted();

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

@@ -1038,8 +1038,9 @@ protected void handleSend(CommandSend send, ByteBuf headersAndPayload) {
             if (nonPersistentPendingMessages > MaxNonPersistentPendingMessages) {
                 final long producerId = send.getProducerId();
                 final long sequenceId = send.getSequenceId();
+                final long highestSequenceId = send.getHighestSequenceId();
                 service.getTopicOrderedExecutor().executeOrdered(producer.getTopic().getName(), SafeRun.safeRun(() -> {
-                    ctx.writeAndFlush(Commands.newSendReceipt(producerId, sequenceId, -1, -1), ctx.voidPromise());
+                    ctx.writeAndFlush(Commands.newSendReceipt(producerId, sequenceId, highestSequenceId, -1, -1), ctx.voidPromise());
                 }));
                 producer.recordMessageDrop(send.getNumMessages());
                 return;

pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java

@@ -267,7 +267,7 @@ private void producerContinuousRetryAfterSendFail(String topic) throws Exception
                 ctx.writeAndFlush(Commands.newSendError(0, 0, ServerError.PersistenceError, "Send Failed"));
                 return;
             }
-            ctx.writeAndFlush(Commands.newSendReceipt(0, 0, 1, 1));
+            ctx.writeAndFlush(Commands.newSendReceipt(0, 0, 0, 1, 1));
         });
 
         try {
@@ -587,7 +587,7 @@ public void testProducerReconnect() throws Exception {
 
         mockBrokerService.setHandleSend((ctx, sendCmd, headersAndPayload) -> {
             msgSent.set(true);
-            ctx.writeAndFlush(Commands.newSendReceipt(0, 0, 1, 1));
+            ctx.writeAndFlush(Commands.newSendReceipt(0, 0, 0, 1, 1));
         });
 
         PulsarClient client = PulsarClient.builder().serviceUrl("http://127.0.0.1:" + WEB_SERVICE_PORT).build();

pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java

@@ -185,7 +185,7 @@ protected void handleSend(CommandSend send, ByteBuf headersAndPayload) {
                 return;
             }
             // default
-            ctx.writeAndFlush(Commands.newSendReceipt(producerId, send.getSequenceId(), 0, 0));
+            ctx.writeAndFlush(Commands.newSendReceipt(producerId, send.getSequenceId(), 0, 0, 0));
         }
 
         @Override

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

@@ -347,6 +347,7 @@ protected void handleSendReceipt(CommandSendReceipt sendReceipt) {
 
         long producerId = sendReceipt.getProducerId();
         long sequenceId = sendReceipt.getSequenceId();
+        long highestSequenceId = sendReceipt.getHighestSequenceId();
         long ledgerId = -1;
         long entryId = -1;
         if (sendReceipt.hasMessageId()) {
@@ -364,7 +365,7 @@ protected void handleSendReceipt(CommandSendReceipt sendReceipt) {
                     ledgerId, entryId);
         }
 
-        producers.get(producerId).ackReceived(this, sequenceId, ledgerId, entryId);
+        producers.get(producerId).ackReceived(this, sequenceId, highestSequenceId, ledgerId, entryId);
     }
 
     @Override

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

@@ -788,38 +788,46 @@ public void terminated(ClientCnx cnx) {
         }
     }
 
-    void ackReceived(ClientCnx cnx, long sequenceId, long ledgerId, long entryId) {
+    void ackReceived(ClientCnx cnx, long sequenceId, long highestSequenceId, long ledgerId, long entryId) {
         OpSendMsg op = null;
         boolean callback = false;
         synchronized (this) {
             op = pendingMessages.peek();
             if (op == null) {
                 if (log.isDebugEnabled()) {
-                    log.debug("[{}] [{}] Got ack for timed out msg {}", topic, producerName, sequenceId);
+                    log.debug("[{}] [{}] Got ack for timed out msg {} - {}", topic, producerName, sequenceId, highestSequenceId);
                 }
                 return;
             }
-            long expectedSequenceId = getHighestSequenceId(op);
-            if (sequenceId > expectedSequenceId) {
-                log.warn("[{}] [{}] Got ack for msg. expecting: {} - got: {} - queue-size: {}", topic, producerName,
-                        expectedSequenceId, sequenceId, pendingMessages.size());
+
+            if (sequenceId > op.sequenceId) {
+                log.warn("[{}] [{}] Got ack for msg. expecting: {} - {} - got: {} - {} - queue-size: {}", topic, producerName,
+                        op.sequenceId, op.highestSequenceId, sequenceId, highestSequenceId, pendingMessages.size());
                 // Force connection closing so that messages can be re-transmitted in a new connection
                 cnx.channel().close();
-            } else if (sequenceId < expectedSequenceId) {
+            } else if (sequenceId < op.sequenceId) {
                 // Ignoring the ack since it's referring to a message that has already timed out.
                 if (log.isDebugEnabled()) {
-                    log.debug("[{}] [{}] Got ack for timed out msg {} last-seq: {}", topic, producerName, sequenceId,
-                            expectedSequenceId);
+                    log.debug("[{}] [{}] Got ack for timed out msg. expecting: {} - {} - got: {} - {}", topic, producerName,
+                            op.sequenceId, op.highestSequenceId, sequenceId, highestSequenceId);
                 }
             } else {
-                // Message was persisted correctly
-                if (log.isDebugEnabled()) {
-                    log.debug("[{}] [{}] Received ack for msg {} ", topic, producerName, sequenceId);
+                // Add check `sequenceId >= highestSequenceId` for backward compatibility.
+                if (sequenceId >= highestSequenceId || highestSequenceId == op.highestSequenceId) {
+                    // Message was persisted correctly
+                    if (log.isDebugEnabled()) {
+                        log.debug("[{}] [{}] Received ack for msg {} ", topic, producerName, sequenceId);
+                    }
+                    pendingMessages.remove();
+                    releaseSemaphoreForSendOp(op);
+                    callback = true;
+                    pendingCallbacks.add(op);
+                } else {
+                    log.warn("[{}] [{}] Got ack for batch msg error. expecting: {} - {} - got: {} - {} - queue-size: {}", topic, producerName,
+                            op.sequenceId, op.highestSequenceId, sequenceId, highestSequenceId, pendingMessages.size());
+                    // Force connection closing so that messages can be re-transmitted in a new connection
+                    cnx.channel().close();
                 }
-                pendingMessages.remove();
-                releaseSemaphoreForSendOp(op);
-                callback = true;
-                pendingCallbacks.add(op);
             }
         }
         if (callback) {
@@ -844,6 +852,7 @@ void ackReceived(ClientCnx cnx, long sequenceId, long ledgerId, long entryId) {
     private long getHighestSequenceId(OpSendMsg op) {
         return Math.max(op.highestSequenceId, op.sequenceId);
     }
+
     private void releaseSemaphoreForSendOp(OpSendMsg op) {
         semaphore.release(isBatchMessagingEnabled() ? op.numMessagesInBatch : 1);
     }
@@ -952,7 +961,6 @@ protected static final class OpSendMsg {
         long createdAt;
         long batchSizeByte = 0;
         int numMessagesInBatch = 1;
-        long lowestSequenceId;
         long highestSequenceId;
 
         static OpSendMsg create(MessageImpl<?> msg, ByteBufPair cmd, long sequenceId, SendCallback callback) {
@@ -981,9 +989,8 @@ static OpSendMsg create(List<MessageImpl<?>> msgs, ByteBufPair cmd, long lowestS
             op.msgs = msgs;
             op.cmd = cmd;
             op.callback = callback;
-            op.lowestSequenceId = lowestSequenceId;
-            op.highestSequenceId = highestSequenceId;
             op.sequenceId = lowestSequenceId;
+            op.highestSequenceId = highestSequenceId;
             op.createdAt = System.currentTimeMillis();
             return op;
         }
@@ -996,7 +1003,6 @@ void recycle() {
             rePopulate = null;
             sequenceId = -1L;
             createdAt = -1L;
-            lowestSequenceId = -1L;
             highestSequenceId = -1L;
             recyclerHandle.recycle(this);
         }

pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java

@@ -336,10 +336,12 @@ public static ByteBuf newError(long requestId, ServerError error, String message
 
     }
 
-    public static ByteBuf newSendReceipt(long producerId, long sequenceId, long ledgerId, long entryId) {
+    public static ByteBuf newSendReceipt(long producerId, long sequenceId, long highestId, long ledgerId,
+             long entryId) {
         CommandSendReceipt.Builder sendReceiptBuilder = CommandSendReceipt.newBuilder();
         sendReceiptBuilder.setProducerId(producerId);
         sendReceiptBuilder.setSequenceId(sequenceId);
+        sendReceiptBuilder.setHighestSequenceId(highestId);
         MessageIdData.Builder messageIdBuilder = MessageIdData.newBuilder();
         messageIdBuilder.setLedgerId(ledgerId);
         messageIdBuilder.setEntryId(entryId);

pulsar-common/src/main/proto/PulsarApi.proto

@@ -423,6 +423,7 @@ message CommandSendReceipt {
     required uint64 producer_id = 1;
     required uint64 sequence_id = 2;
     optional MessageIdData message_id = 3;
+    optional uint64 highest_sequence_id = 4 [default = 0];
 }
 
 message CommandSendError {

tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/PulsarStandaloneTestSuite2_3.java

@@ -0,0 +1,42 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.integration.backwardscompatibility;
+
+import org.apache.pulsar.tests.integration.containers.PulsarContainer;
+import org.apache.pulsar.tests.integration.topologies.PulsarStandaloneTestBase;
+import org.testng.ITest;
+import org.testng.annotations.AfterSuite;
+import org.testng.annotations.BeforeSuite;
+
+public class PulsarStandaloneTestSuite2_3 extends PulsarStandaloneTestBase implements ITest {
+
+    @BeforeSuite
+    public void setUpCluster() throws Exception {
+        super.startCluster(PulsarContainer.PULSAR_2_3_IMAGE_NAME);
+    }
+
+    @AfterSuite
+    public void tearDownCluster() throws Exception {
+        super.stopCluster();
+    }
+    @Override
+    public String getTestName() {
+        return "pulsar-standalone-suite";
+    }
+}

tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/PulsarStandaloneTestSuite2_4.java

@@ -0,0 +1,42 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.integration.backwardscompatibility;
+
+import org.apache.pulsar.tests.integration.containers.PulsarContainer;
+import org.apache.pulsar.tests.integration.topologies.PulsarStandaloneTestBase;
+import org.testng.ITest;
+import org.testng.annotations.AfterSuite;
+import org.testng.annotations.BeforeSuite;
+
+public class PulsarStandaloneTestSuite2_4 extends PulsarStandaloneTestBase implements ITest {
+
+    @BeforeSuite
+    public void setUpCluster() throws Exception {
+        super.startCluster(PulsarContainer.PULSAR_2_4_IMAGE_NAME);
+    }
+
+    @AfterSuite
+    public void tearDownCluster() throws Exception {
+        super.stopCluster();
+    }
+    @Override
+    public String getTestName() {
+        return "pulsar-standalone-suite";
+    }
+}