[pulsar-broker][replicator] add remote cluster name in producer name (a…

…pache#3553) * [pulsar-broker][replicator] add remote cluster name in producer name * add padding between remote and local cluster-names * broker-producer can parse remote-cluster name in producer-name
JamieCressey · Feb 10, 2019 · 00c71e2 · 00c71e2
1 parent f11e994
commit 00c71e2
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 1 deletion.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
@@ -46,6 +46,7 @@ public abstract class AbstractReplicator {
     protected final PulsarClientImpl client;
 
     protected volatile ProducerImpl producer;
+    public static final String REPL_PRODUCER_NAME_DELIMITER = "-->";
 
     protected final int producerQueueSize;
     protected final ProducerBuilder<byte[]> producerBuilder;

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java
@@ -50,6 +50,7 @@
 import org.apache.pulsar.common.util.DateFormatter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import static org.apache.pulsar.broker.service.AbstractReplicator.REPL_PRODUCER_NAME_DELIMITER;
 
 /**
  * Represents a currently connected producer
@@ -107,7 +108,7 @@ public Producer(Topic topic, ServerCnx cnx, long producerId, String producerName
 
         this.isRemote = producerName
                 .startsWith(cnx.getBrokerService().pulsar().getConfiguration().getReplicatorPrefix());
-        this.remoteCluster = isRemote ? producerName.split("\\.")[2] : null;
+        this.remoteCluster = isRemote ? producerName.split("\\.")[2].split(REPL_PRODUCER_NAME_DELIMITER)[0] : null;
 
         this.isEncrypted = isEncrypted;
         this.schemaVersion = schemaVersion;