[pulsar] branch master updated: add instance and fqn into log message properties (#11399)

[sijie]

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

sijie 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 b6d02e4  add instance and fqn into log message properties (#11399)
b6d02e4 is described below

commit b6d02e46b753bc2b321a5da8feca0ab7695c3a56
Author: Neng Lu <n...@streamnative.io>
AuthorDate: Mon Jul 26 15:33:54 2021 -0700

    add instance and fqn into log message properties (#11399)
    
    ### Motivation
    
    Enrich the published log message with `instanceId` and `fqn` set in 
properties.
    
    ### Modifications
    
    add `instanceId` and `fqn` into log message properties
---
 .../org/apache/pulsar/functions/instance/JavaInstanceRunnable.java  | 3 ++-
 .../main/java/org/apache/pulsar/functions/instance/LogAppender.java | 6 +++++-
 2 files changed, 7 insertions(+), 2 deletions(-)

diff --git 
a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
 
b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
index e36dd26..3d79c2b 100644
--- 
a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
+++ 
b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
@@ -574,7 +574,8 @@ public class JavaInstanceRunnable implements AutoCloseable, 
Runnable {
             // to prevent "SSE4.2 CRC32C provider initialized" appearing in 
log topic
             new Crc32cIntChecksum();
             logAppender = new LogAppender(client, 
instanceConfig.getFunctionDetails().getLogTopic(),
-                    
FunctionCommon.getFullyQualifiedName(instanceConfig.getFunctionDetails()));
+                    
FunctionCommon.getFullyQualifiedName(instanceConfig.getFunctionDetails()),
+                    instanceConfig.getInstanceName());
             logAppender.start();
             setupLogTopicAppender(LoggerContext.getContext());
         }
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 e0a4c95..20354d4 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
@@ -38,14 +38,16 @@ public class LogAppender implements Appender {
     private PulsarClient pulsarClient;
     private String logTopic;
     private String fqn;
+    private String instance;
     private State state;
     private ErrorHandler errorHandler;
     private Producer<byte[]> producer;
 
-    public LogAppender(PulsarClient pulsarClient, String logTopic, String fqn) 
{
+    public LogAppender(PulsarClient pulsarClient, String logTopic, String fqn, 
String instance) {
         this.pulsarClient = pulsarClient;
         this.logTopic = logTopic;
         this.fqn = fqn;
+        this.instance = instance;
     }
 
     @Override
@@ -53,6 +55,8 @@ public class LogAppender implements Appender {
         producer.newMessage()
                 
.value(logEvent.getMessage().getFormattedMessage().getBytes(StandardCharsets.UTF_8))
                 .property("loglevel", logEvent.getLevel().name())
+                .property("instance", instance)
+                .property("fqn", fqn)
                 .sendAsync();
     }