Avoid setting ReplicatedFrom while init batchMessageMetadata (#107)

pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/ReplicatorTest.java

@@ -551,7 +551,7 @@ public class ReplicatorTest extends ReplicatorTestBase {
 
         for (Future<Void> result : results) {
             try {
-                result.get();
+                result.get(5, TimeUnit.SECONDS);
             } catch (Exception e) {
                 log.error("exception in getting future result ", e);
                 fail(String.format("replication test failed with %s exception", e.getMessage()));

pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/ReplicatorTestBase.java

@@ -252,9 +252,10 @@ public class ReplicatorTestBase {
             ProducerConfiguration producerConfiguration = new ProducerConfiguration();
             if (batch) {
                 producerConfiguration.setBatchingEnabled(true);
+                producerConfiguration.setBatchingMaxPublishDelay(1, TimeUnit.SECONDS);
                 producerConfiguration.setBatchingMaxMessages(5);
             }
-            producer = client.createProducer(topicName);
+            producer = client.createProducer(topicName, producerConfiguration);
 
         }
 

pulsar-common/src/main/java/com/yahoo/pulsar/common/api/Commands.java

@@ -611,7 +611,9 @@ public class Commands {
         messageMetadata.setPublishTime(builder.getPublishTime());
         messageMetadata.setProducerName(builder.getProducerName());
         messageMetadata.setSequenceId(builder.getSequenceId());
-        messageMetadata.setReplicatedFrom(builder.getReplicatedFrom());
+        if (builder.hasReplicatedFrom()) {
+            messageMetadata.setReplicatedFrom(builder.getReplicatedFrom());
+        }
         return builder.getSequenceId();
     }