From 47414e0d537272487c44ded16f2996b27078ffa5 Mon Sep 17 00:00:00 2001
From: "Zaiming (Stone) Shi" <zmstone@gmail.com>
Date: Fri, 13 Jan 2023 08:31:26 +0100
Subject: [PATCH] docs: improve kafka key and value field description

---
 .../emqx_plugin_libs/src/emqx_placeholder.erl |  2 +-
 .../i18n/emqx_ee_bridge_kafka.conf            | 15 ++++++-----
 .../src/emqx_ee_bridge_kafka.erl              |  6 ++---
 .../kafka/emqx_bridge_impl_kafka_producer.erl | 25 +++++++++++++------
 .../emqx_bridge_impl_kafka_producer_SUITE.erl |  7 +++++-
 5 files changed, 37 insertions(+), 18 deletions(-)

diff --git a/apps/emqx_plugin_libs/src/emqx_placeholder.erl b/apps/emqx_plugin_libs/src/emqx_placeholder.erl
index b7dea5256..3e98fa149 100644
--- a/apps/emqx_plugin_libs/src/emqx_placeholder.erl
+++ b/apps/emqx_plugin_libs/src/emqx_placeholder.erl
@@ -298,7 +298,7 @@ replace_with(Tmpl, RE, '$n') ->
         ),
     Res.
 parse_nested(<<".", R/binary>>) ->
-    %% ignroe the root .
+    %% ignore the root .
     parse_nested(R);
 parse_nested(Attr) ->
     case string:split(Attr, <<".">>, all) of
diff --git a/lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_kafka.conf b/lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_kafka.conf
index 7ccf20d0b..201eeba3b 100644
--- a/lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_kafka.conf
+++ b/lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_kafka.conf
@@ -250,9 +250,10 @@ emqx_ee_bridge_kafka {
     kafka_message_key {
         desc {
             en: "Template to render Kafka message key. "
-                "If the desired variable for this template is not found in the input data "
-                "<code>NULL</code> is used."
-            zh: "生成 Kafka 消息 Key 的模版。当所需要的输入没有时，会使用 <code>NULL</code>。"
+                "If the template is rendered into a NULL value (i.e. there is no such data field in Rule Engine context) "
+                "then Kafka's <code>NULL</code> (but not empty string) is used."
+            zh: "生成 Kafka 消息 Key 的模版。如果模版生成后为空值，"
+                "则会使用 Kafka 的 <code>NULL</code> ，而非空字符串。"
         }
         label {
             en: "Message Key"
@@ -262,9 +263,11 @@ emqx_ee_bridge_kafka {
     kafka_message_value {
         desc {
             en: "Template to render Kafka message value. "
-                "If the desired variable for this template is not found in the input data "
-                "<code>NULL</code> is used."
-            zh: "生成 Kafka 消息 Value 的模版。当所需要的输入没有时，会使用 <code>NULL</code>。"
+                "If the template is rendered "
+                "into a NULL value (i.e. there is no such data field in Rule Engine context) "
+                "then Kafka's <code>NULL</code> (but not empty string) is used."
+            zh: "生成 Kafka 消息 Value 的模版。如果模版生成后为空值，"
+                "则会使用 Kafka 的 <code>NULL</code>，而非空字符串。"
         }
         label {
             en: "Message Value"
diff --git a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_kafka.erl b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_kafka.erl
index 9fae4f30a..4a6c1411c 100644
--- a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_kafka.erl
+++ b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_kafka.erl
@@ -207,11 +207,11 @@ fields(producer_kafka_opts) ->
     ];
 fields(kafka_message) ->
     [
-        {key, mk(string(), #{default => "${clientid}", desc => ?DESC(kafka_message_key)})},
-        {value, mk(string(), #{default => "${payload}", desc => ?DESC(kafka_message_value)})},
+        {key, mk(string(), #{default => "${.clientid}", desc => ?DESC(kafka_message_key)})},
+        {value, mk(string(), #{default => "${.}", desc => ?DESC(kafka_message_value)})},
         {timestamp,
             mk(string(), #{
-                default => "${timestamp}", desc => ?DESC(kafka_message_timestamp)
+                default => "${.timestamp}", desc => ?DESC(kafka_message_timestamp)
             })}
     ];
 fields(producer_buffer) ->
diff --git a/lib-ee/emqx_ee_bridge/src/kafka/emqx_bridge_impl_kafka_producer.erl b/lib-ee/emqx_ee_bridge/src/kafka/emqx_bridge_impl_kafka_producer.erl
index a7de100d6..d7b4de0c0 100644
--- a/lib-ee/emqx_ee_bridge/src/kafka/emqx_bridge_impl_kafka_producer.erl
+++ b/lib-ee/emqx_ee_bridge/src/kafka/emqx_bridge_impl_kafka_producer.erl
@@ -145,15 +145,19 @@ on_query(_InstId, {send_message, Message}, #{message_template := Template, produ
     {_Partition, _Pid} = wolff:send(Producers, [KafkaMessage], {fun ?MODULE:on_kafka_ack/3, [#{}]}),
     {async_return, ok}.
 
-compile_message_template(#{
-    key := KeyTemplate, value := ValueTemplate, timestamp := TimestampTemplate
-}) ->
+compile_message_template(T) ->
+    KeyTemplate = maps:get(key, T, <<"${.clientid}">>),
+    ValueTemplate = maps:get(value, T, <<"${.}">>),
+    TimestampTemplate = maps:get(value, T, <<"${.timestamp}">>),
     #{
-        key => emqx_plugin_libs_rule:preproc_tmpl(KeyTemplate),
-        value => emqx_plugin_libs_rule:preproc_tmpl(ValueTemplate),
-        timestamp => emqx_plugin_libs_rule:preproc_tmpl(TimestampTemplate)
+        key => preproc_tmpl(KeyTemplate),
+        value => preproc_tmpl(ValueTemplate),
+        timestamp => preproc_tmpl(TimestampTemplate)
     }.
 
+preproc_tmpl(Tmpl) ->
+    emqx_plugin_libs_rule:preproc_tmpl(Tmpl).
+
 render_message(
     #{key := KeyTemplate, value := ValueTemplate, timestamp := TimestampTemplate}, Message
 ) ->
@@ -164,7 +168,14 @@ render_message(
     }.
 
 render(Template, Message) ->
-    emqx_plugin_libs_rule:proc_tmpl(Template, Message).
+    Opts = #{
+        var_trans => fun
+            (undefined) -> <<"">>;
+            (X) -> emqx_plugin_libs_rule:bin(X)
+        end,
+        return => full_binary
+    },
+    emqx_plugin_libs_rule:proc_tmpl(Template, Message, Opts).
 
 render_timestamp(Template, Message) ->
     try
diff --git a/lib-ee/emqx_ee_bridge/test/emqx_bridge_impl_kafka_producer_SUITE.erl b/lib-ee/emqx_ee_bridge/test/emqx_bridge_impl_kafka_producer_SUITE.erl
index 8e5b1fa95..2d67a9941 100644
--- a/lib-ee/emqx_ee_bridge/test/emqx_bridge_impl_kafka_producer_SUITE.erl
+++ b/lib-ee/emqx_ee_bridge/test/emqx_bridge_impl_kafka_producer_SUITE.erl
@@ -260,7 +260,11 @@ kafka_bridge_rest_api_helper(Config) ->
                 topic => <<"t/#">>
             },
             <<"kafka">> => #{
-                <<"topic">> => erlang:list_to_binary(KafkaTopic)
+                <<"topic">> => erlang:list_to_binary(KafkaTopic),
+                <<"message">> => #{
+                    <<"key">> => <<"${clientid}">>,
+                    <<"value">> => <<"${.payload}">>
+                }
             }
         }
     },
@@ -501,6 +505,7 @@ producer = {
     }
     kafka = {
         topic = \"{{ kafka_topic }}\"
+        message = {key = \"${clientid}\", value = \"${.payload}\"}
     }
 }
 """.