From 5397edcb80e74dd490d27a0befa85abcea1a3494 Mon Sep 17 00:00:00 2001
From: Neng Lu <nlu@streamnative.io>
Date: Mon, 26 Jul 2021 15:33:54 -0700
Subject: [PATCH] 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)
---
 .../pulsar/functions/instance/JavaInstanceRunnable.java     | 3 ++-
 .../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 e36dd264c3872..3d79c2ba5c2c1 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 @@ private void setupLogHandler() {
             // 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 ca669be593179..bbca3f9efa1dd 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 void append(LogEvent logEvent) {
         producer.newMessage()
                 .value(logEvent.getMessage().getFormattedMessage().getBytes(StandardCharsets.UTF_8))
                 .property("loglevel", logEvent.getLevel().name())
+                .property("instance", instance)
+                .property("fqn", fqn)
                 .sendAsync();
     }