[pulsar] branch master updated: Expose `Replicated From` info in Message (#4524)

[mm...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new f33be02   Expose `Replicated From` info in Message (#4524)
f33be02 is described below

commit f33be02aef139af5e03ebe955288d9e445575935
Author: Jia Zhai <zh...@apache.org>
AuthorDate: Fri Jun 14 08:10:27 2019 +0800

     Expose `Replicated From` info in Message (#4524)
    
    * expose replicated info in message
    
    * add negative test case
---
 .../java/org/apache/pulsar/client/api/Message.java | 17 +++++
 .../org/apache/pulsar/client/impl/MessageImpl.java |  2 +
 .../pulsar/client/impl/TopicMessageImpl.java       | 10 +++
 .../org/apache/pulsar/client/impl/MessageTest.java | 83 ++++++++++++++++++++++
 4 files changed, 112 insertions(+)

diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Message.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Message.java
index e192916..6041632 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Message.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Message.java
@@ -202,4 +202,21 @@ public interface Message<T> {
      * @return Schema version of the message if the message is produced with schema otherwise null.
      */
     byte[] getSchemaVersion();
+
+    /**
+     * Check whether the message is replicated from other cluster.
+     *
+     * @since 2.4.0
+     * @return true if the message is replicated from other cluster.
+     *         false otherwise.
+     */
+    boolean isReplicated();
+
+    /**
+     * Get name of cluster, from which the message is replicated.
+     *
+     * @since 2.4.0
+     * @return the name of cluster, from which the message is replicated.
+     */
+    String getReplicatedFrom();
 }
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java
index 7fc37cc..fab8683 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java
@@ -189,11 +189,13 @@ public class MessageImpl<T> implements Message<T> {
         msgMetadataBuilder.setReplicatedFrom(cluster);
     }
 
+    @Override
     public boolean isReplicated() {
         checkNotNull(msgMetadataBuilder);
         return msgMetadataBuilder.hasReplicatedFrom();
     }
 
+    @Override
     public String getReplicatedFrom() {
         checkNotNull(msgMetadataBuilder);
         return msgMetadataBuilder.getReplicatedFrom();
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicMessageImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicMessageImpl.java
index 723875c..fb62ba3 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicMessageImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicMessageImpl.java
@@ -158,6 +158,16 @@ public class TopicMessageImpl<T> implements Message<T> {
         return msg.getSchemaVersion();
     }
 
+    @Override
+    public boolean isReplicated() {
+        return msg.isReplicated();
+    }
+
+    @Override
+    public String getReplicatedFrom() {
+        return msg.getReplicatedFrom();
+    }
+
     public Message<T> getMessage() {
         return msg;
     }
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MessageTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MessageTest.java
new file mode 100644
index 0000000..4b96c0a
--- /dev/null
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MessageTest.java
@@ -0,0 +1,83 @@
+/**
+ * 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.client.impl;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+import java.nio.ByteBuffer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
+import org.testng.annotations.Test;
+
+/**
+ * Unit test of {@link Message} methods.
+ */
+public class MessageTest {
+
+    @Test
+    public void testMessageImplReplicatedInfo() {
+        String from = "ClusterNameOfReplicatedFrom";
+        MessageMetadata.Builder builder = MessageMetadata.newBuilder().setReplicatedFrom(from);
+        ByteBuffer payload = ByteBuffer.wrap(new byte[0]);
+        Message<byte[]> msg = MessageImpl.create(builder, payload, Schema.BYTES);
+
+        assertTrue(msg.isReplicated());
+        assertEquals(msg.getReplicatedFrom(), from);
+    }
+
+    @Test
+    public void testMessageImplNoReplicatedInfo() {
+        MessageMetadata.Builder builder = MessageMetadata.newBuilder();
+        ByteBuffer payload = ByteBuffer.wrap(new byte[0]);
+        Message<byte[]> msg = MessageImpl.create(builder, payload, Schema.BYTES);
+
+        assertFalse(msg.isReplicated());
+        assertTrue(msg.getReplicatedFrom().isEmpty());
+    }
+
+    @Test
+    public void testTopicMessageImplReplicatedInfo() {
+        String from = "ClusterNameOfReplicatedFromForTopicMessage";
+        String topicName = "myTopic";
+        MessageMetadata.Builder builder = MessageMetadata.newBuilder().setReplicatedFrom(from);
+        ByteBuffer payload = ByteBuffer.wrap(new byte[0]);
+        MessageImpl<byte[]> msg = MessageImpl.create(builder, payload, Schema.BYTES);
+        msg.setMessageId(new MessageIdImpl(-1, -1, -1));
+        TopicMessageImpl<byte[]> topicMessage = new TopicMessageImpl<>(topicName, topicName, msg);
+
+        assertTrue(topicMessage.isReplicated());
+        assertEquals(msg.getReplicatedFrom(), from);
+    }
+
+    @Test
+    public void testTopicMessageImplNoReplicatedInfo() {
+        String topicName = "myTopic";
+        MessageMetadata.Builder builder = MessageMetadata.newBuilder();
+        ByteBuffer payload = ByteBuffer.wrap(new byte[0]);
+        MessageImpl<byte[]> msg = MessageImpl.create(builder, payload, Schema.BYTES);
+        msg.setMessageId(new MessageIdImpl(-1, -1, -1));
+        TopicMessageImpl<byte[]> topicMessage = new TopicMessageImpl<>(topicName, topicName, msg);
+
+        assertFalse(topicMessage.isReplicated());
+        assertTrue(topicMessage.getReplicatedFrom().isEmpty());
+    }
+}