[fix][broker] Geo Replication lost messages or frequently fails due to Deduplication is not appropriate for Geo-Replication (apache#23697)

(cherry picked from commit 4ac4f3c)
(cherry picked from commit 307b5c9)

Author: poorbarcode

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

@@ -177,6 +177,7 @@ public void startProducer() {
         prepareCreateProducer().thenCompose(ignore -> {
             ProducerBuilderImpl builderImpl = (ProducerBuilderImpl) producerBuilder;
             builderImpl.getConf().setNonPartitionedTopicExpected(true);
+            builderImpl.getConf().setReplProducer(true);
             return producerBuilder.createAsync().thenAccept(producer -> {
                 setProducerAndTriggerReadEntries(producer);
             });

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

@@ -20,6 +20,8 @@
 
 import static com.scurrilous.circe.checksum.Crc32cIntChecksum.computeChecksum;
 import static org.apache.pulsar.broker.service.AbstractReplicator.REPL_PRODUCER_NAME_DELIMITER;
+import static org.apache.pulsar.client.impl.GeoReplicationProducerImpl.MSG_PROP_IS_REPL_MARKER;
+import static org.apache.pulsar.client.impl.GeoReplicationProducerImpl.MSG_PROP_REPL_SOURCE_POSITION;
 import static org.apache.pulsar.common.protocol.Commands.hasChecksum;
 import static org.apache.pulsar.common.protocol.Commands.readChecksum;
 import com.google.common.annotations.VisibleForTesting;
@@ -87,6 +89,7 @@ public class Producer {
 
     private final PublisherStatsImpl stats;
     private final boolean isRemote;
+    private final boolean isRemoteOrShadow;
     private final String remoteCluster;
     private final boolean isNonPersistentTopic;
     private final boolean isShadowTopic;
@@ -148,6 +151,7 @@ public Producer(Topic topic, TransportCnx cnx, long producerId, String producerN
 
         String replicatorPrefix = serviceConf.getReplicatorPrefix() + ".";
         this.isRemote = producerName.startsWith(replicatorPrefix);
+        this.isRemoteOrShadow = isRemoteOrShadow(producerName, serviceConf.getReplicatorPrefix());
         this.remoteCluster = parseRemoteClusterName(producerName, isRemote, replicatorPrefix);
 
         this.isEncrypted = isEncrypted;
@@ -160,6 +164,13 @@ public Producer(Topic topic, TransportCnx cnx, long producerId, String producerN
         this.brokerInterceptor = cnx.getBrokerService().getInterceptor();
     }
 
+    /**
+     * Difference with "isRemote" is whether the prefix string is end with a dot.
+     */
+    public static boolean isRemoteOrShadow(String producerName, String replicatorPrefix) {
+        return producerName != null && producerName.startsWith(replicatorPrefix);
+    }
+
     /**
      * Producer name for replicator is in format.
      * "replicatorPrefix.localCluster" (old)
@@ -281,11 +292,16 @@ private boolean checkCanProduceTxnOnTopic(long sequenceId, ByteBuf headersAndPay
         return true;
     }
 
+    private boolean isSupportsReplDedupByLidAndEid() {
+        // Non-Persistent topic does not have ledger id or entry id, so it does not support.
+        return cnx.isClientSupportsReplDedupByLidAndEid() && topic.isPersistent();
+    }
+
     private void publishMessageToTopic(ByteBuf headersAndPayload, long sequenceId, long batchSize, boolean isChunked,
                                        boolean isMarker, Position position) {
         MessagePublishContext messagePublishContext =
                 MessagePublishContext.get(this, sequenceId, msgIn, headersAndPayload.readableBytes(),
-                        batchSize, isChunked, System.nanoTime(), isMarker, position);
+                        batchSize, isChunked, System.nanoTime(), isMarker, position, isSupportsReplDedupByLidAndEid());
         if (brokerInterceptor != null) {
             brokerInterceptor
                     .onMessagePublish(this, headersAndPayload, messagePublishContext);
@@ -297,7 +313,7 @@ private void publishMessageToTopic(ByteBuf headersAndPayload, long lowestSequenc
                                        long batchSize, boolean isChunked, boolean isMarker, Position position) {
         MessagePublishContext messagePublishContext = MessagePublishContext.get(this, lowestSequenceId,
                 highestSequenceId, msgIn, headersAndPayload.readableBytes(), batchSize,
-                isChunked, System.nanoTime(), isMarker, position);
+                isChunked, System.nanoTime(), isMarker, position, isSupportsReplDedupByLidAndEid());
         if (brokerInterceptor != null) {
             brokerInterceptor
                     .onMessagePublish(this, headersAndPayload, messagePublishContext);
@@ -393,6 +409,7 @@ private static final class MessagePublishContext implements PublishContext, Runn
         private long batchSize;
         private boolean chunked;
         private boolean isMarker;
+        private boolean supportsReplDedupByLidAndEid;
 
         private long startTimeNs;
 
@@ -483,6 +500,11 @@ public long getOriginalSequenceId() {
             return originalSequenceId;
         }
 
+        @Override
+        public boolean supportsReplDedupByLidAndEid() {
+            return supportsReplDedupByLidAndEid;
+        }
+
         @Override
         public void setOriginalHighestSequenceId(long originalHighestSequenceId) {
             this.originalHighestSequenceId = originalHighestSequenceId;
@@ -550,8 +572,12 @@ public void run() {
             // stats
             rateIn.recordMultipleEvents(batchSize, msgSize);
             producer.topic.recordAddLatency(System.nanoTime() - startTimeNs, TimeUnit.NANOSECONDS);
-            producer.cnx.getCommandSender().sendSendReceiptResponse(producer.producerId, sequenceId, highestSequenceId,
-                    ledgerId, entryId);
+            if (producer.isRemoteOrShadow && producer.isSupportsReplDedupByLidAndEid()) {
+                sendSendReceiptResponseRepl();
+            } else {
+                // Repl V1 is the same as normal for this handling.
+                sendSendReceiptResponseNormal();
+            }
             producer.cnx.completedSendOperation(producer.isNonPersistentTopic, msgSize);
             if (this.chunked) {
                 producer.chunkedMessageRate.recordEvent();
@@ -564,8 +590,46 @@ public void run() {
             recycle();
         }
 
+        private void sendSendReceiptResponseRepl() {
+            // Case-1: is a repl marker.
+            boolean isReplMarker = getProperty(MSG_PROP_IS_REPL_MARKER) != null;
+            if (isReplMarker) {
+                producer.cnx.getCommandSender().sendSendReceiptResponse(producer.producerId, sequenceId, Long.MIN_VALUE,
+                        ledgerId, entryId);
+
+                return;
+            }
+            // Case-2: is a repl message.
+            Object positionPairObj = getProperty(MSG_PROP_REPL_SOURCE_POSITION);
+            if (positionPairObj == null || !(positionPairObj instanceof long[])
+                    || ((long[]) positionPairObj).length < 2) {
+                log.error("[{}] Message can not determine whether the message is duplicated due to the acquired"
+                                + " messages props were are invalid. producer={}. supportsReplDedupByLidAndEid: {},"
+                                + " sequence-id {}, prop-{}: not in expected format",
+                        producer.topic.getName(), producer.producerName,
+                        supportsReplDedupByLidAndEid(), getSequenceId(),
+                        MSG_PROP_REPL_SOURCE_POSITION);
+                producer.cnx.getCommandSender().sendSendError(producer.producerId,
+                        Math.max(highestSequenceId, sequenceId),
+                        ServerError.PersistenceError, "Message can not determine whether the message is"
+                                + " duplicated due to the acquired messages props were are invalid");
+                return;
+            }
+            long[] positionPair = (long[]) positionPairObj;
+            long replSequenceLId = positionPair[0];
+            long replSequenceEId = positionPair[1];
+            producer.cnx.getCommandSender().sendSendReceiptResponse(producer.producerId, replSequenceLId,
+                    replSequenceEId, ledgerId, entryId);
+        }
+
+        private void sendSendReceiptResponseNormal() {
+            producer.cnx.getCommandSender().sendSendReceiptResponse(producer.producerId, sequenceId, highestSequenceId,
+                    ledgerId, entryId);
+        }
+
         static MessagePublishContext get(Producer producer, long sequenceId, Rate rateIn, int msgSize,
-                long batchSize, boolean chunked, long startTimeNs, boolean isMarker, Position position) {
+                             long batchSize, boolean chunked, long startTimeNs, boolean isMarker, Position position,
+                             boolean supportsReplDedupByLidAndEid) {
             MessagePublishContext callback = RECYCLER.get();
             callback.producer = producer;
             callback.sequenceId = sequenceId;
@@ -577,6 +641,7 @@ static MessagePublishContext get(Producer producer, long sequenceId, Rate rateIn
             callback.originalSequenceId = -1L;
             callback.startTimeNs = startTimeNs;
             callback.isMarker = isMarker;
+            callback.supportsReplDedupByLidAndEid = supportsReplDedupByLidAndEid;
             callback.ledgerId = position == null ? -1 : position.getLedgerId();
             callback.entryId = position == null ? -1 : position.getEntryId();
             if (callback.propertyMap != null) {
@@ -586,7 +651,8 @@ static MessagePublishContext get(Producer producer, long sequenceId, Rate rateIn
         }
 
         static MessagePublishContext get(Producer producer, long lowestSequenceId, long highestSequenceId, Rate rateIn,
-                int msgSize, long batchSize, boolean chunked, long startTimeNs, boolean isMarker, Position position) {
+                int msgSize, long batchSize, boolean chunked, long startTimeNs, boolean isMarker, Position position,
+                boolean supportsReplDedupByLidAndEid) {
             MessagePublishContext callback = RECYCLER.get();
             callback.producer = producer;
             callback.sequenceId = lowestSequenceId;
@@ -599,6 +665,7 @@ static MessagePublishContext get(Producer producer, long lowestSequenceId, long
             callback.startTimeNs = startTimeNs;
             callback.chunked = chunked;
             callback.isMarker = isMarker;
+            callback.supportsReplDedupByLidAndEid = supportsReplDedupByLidAndEid;
             callback.ledgerId = position == null ? -1 : position.getLedgerId();
             callback.entryId = position == null ? -1 : position.getEntryId();
             if (callback.propertyMap != null) {
@@ -829,7 +896,8 @@ public void publishTxnMessage(TxnID txnID, long producerId, long sequenceId, lon
         }
         MessagePublishContext messagePublishContext =
                 MessagePublishContext.get(this, sequenceId, highSequenceId, msgIn,
-                        headersAndPayload.readableBytes(), batchSize, isChunked, System.nanoTime(), isMarker, null);
+                        headersAndPayload.readableBytes(), batchSize, isChunked, System.nanoTime(), isMarker, null,
+                        cnx.isClientSupportsReplDedupByLidAndEid());
         if (brokerInterceptor != null) {
             brokerInterceptor
                     .onMessagePublish(this, headersAndPayload, messagePublishContext);

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

@@ -64,6 +64,7 @@
 import javax.net.ssl.SSLSession;
 import javax.ws.rs.WebApplicationException;
 import javax.ws.rs.core.Response;
+import lombok.Getter;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.bookkeeper.mledger.ManagedLedgerException;
@@ -235,6 +236,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
 
     // Flag to manage throttling-rate by atomically enable/disable read-channel.
     private volatile boolean autoReadDisabledRateLimiting = false;
+    @Getter
     private FeatureFlags features;
 
     private PulsarCommandSender commandSender;

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

@@ -127,6 +127,10 @@ default long getEntryTimestamp() {
         default void setEntryTimestamp(long entryTimestamp) {
 
         }
+
+        default boolean supportsReplDedupByLidAndEid() {
+            return false;
+        }
     }
 
     CompletableFuture<Void> initialize();

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

@@ -23,6 +23,7 @@
 import java.net.SocketAddress;
 import java.util.concurrent.CompletableFuture;
 import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.common.api.proto.FeatureFlags;
 
 public interface TransportCnx {
 
@@ -90,4 +91,11 @@ public interface TransportCnx {
      * is null if the connection liveness check is disabled.
      */
     CompletableFuture<Boolean> checkConnectionLiveness();
+
+    FeatureFlags getFeatures();
+
+    default boolean isClientSupportsReplDedupByLidAndEid() {
+        return getFeatures() != null && getFeatures().hasSupportsReplDedupByLidAndEid()
+                && getFeatures().isSupportsReplDedupByLidAndEid();
+    }
 }

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

@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.broker.service.persistent;
 
+import static org.apache.pulsar.client.impl.GeoReplicationProducerImpl.MSG_PROP_REPL_SOURCE_POSITION;
 import io.netty.buffer.ByteBuf;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
@@ -195,9 +196,15 @@ protected boolean replicateEntries(List<Entry> entries) {
                     msg.setSchemaInfoForReplicator(schemaFuture.get());
                     msg.getMessageBuilder().clearTxnidMostBits();
                     msg.getMessageBuilder().clearTxnidLeastBits();
+                    // Add props for sequence checking.
+                    msg.getMessageBuilder().addProperty().setKey(MSG_PROP_REPL_SOURCE_POSITION)
+                            .setValue(String.format("%s:%s", entry.getLedgerId(), entry.getEntryId()));
                     msgOut.recordEvent(headersAndPayload.readableBytes());
                     // Increment pending messages for messages produced locally
                     PENDING_MESSAGES_UPDATER.incrementAndGet(this);
+                    if (log.isDebugEnabled()) {
+                        log.debug("[{}] Publishing {}:{}", replicatorId, entry.getLedgerId(), entry.getEntryId());
+                    }
                     producer.sendAsync(msg, ProducerSendCallback.create(this, entry, msg));
                     atLeastOneMessageSentForReplication = true;
                 }