[pulsar] 03/06: add instance and fqn into log message properties (#11399)

[pe...@apache.org]

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

penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git

commit 5397edcb80e74dd490d27a0befa85abcea1a3494
Author: Neng Lu <nl...@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
    
    (cherry picked from commit b6d02e46b753bc2b321a5da8feca0ab7695c3a56)
---
 .../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 ca669be..bbca3f9 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
@@ -35,14 +35,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
@@ -50,6 +52,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();
     }