[pulsar] branch master updated: In functions log topic appender, don't set producer name (#3544)

[mmerli]

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 c40bb08  In functions log topic appender, don't set producer name 
(#3544)
c40bb08 is described below

commit c40bb08f7e00b6e9dca2ca5f1a0a321113aab9a7
Author: Matteo Merli <mme...@apache.org>
AuthorDate: Fri Feb 8 11:01:22 2019 -0800

    In functions log topic appender, don't set producer name (#3544)
---
 .../java/org/apache/pulsar/functions/instance/LogAppender.java | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

diff --git 
a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/LogAppender.java
 
b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/LogAppender.java
index 3dd71a2..afdc6f9 100644
--- 
a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/LogAppender.java
+++ 
b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/LogAppender.java
@@ -19,6 +19,7 @@
 package org.apache.pulsar.functions.instance;
 
 import org.apache.logging.log4j.core.*;
+import org.apache.pulsar.client.api.CompressionType;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.ProducerBuilder;
 import org.apache.pulsar.client.api.PulsarClient;
@@ -88,13 +89,14 @@ public class LogAppender implements Appender {
     public void start() {
         this.state = State.STARTING;
         try {
-            ProducerBuilder<byte[]> producerBuilder = 
pulsarClient.newProducer()
+            producer = pulsarClient.newProducer()
                     .topic(logTopic)
-                    .producerName(fqn)
                     .blockIfQueueFull(false)
                     .enableBatching(true)
-                    .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS);
-            producer = producerBuilder.create();
+                    .compressionType(CompressionType.LZ4)
+                    .batchingMaxPublishDelay(100, TimeUnit.MILLISECONDS)
+                    .property("function", fqn)
+                    .create();
         } catch (Exception e) {
             throw new RuntimeException("Error starting LogTopic Producer", e);
         }