feat(kafka-upstream): add some how-to document pages for the new feature of kong gateway's kafka-upstream plugin (#8546)

* feat(kafka-upstream): add some how-to document pages for the new feature of kong gateway's kafka-upstream plugin

* lint

* style

* fix

* Apply suggestions from code review

Co-authored-by: lena-larionova <[email protected]>

* break apart codeblocks and add prereqs

---------

Co-authored-by: lena-larionova <[email protected]>
Co-authored-by: lena-larionova <[email protected]>

Author: windmgc

app/_hub/kong-inc/kafka-upstream/how-to/_custom-message-format.md

@@ -0,0 +1,244 @@
+---
+nav_title: Customize Kafka Messages
+title: Customize Kafka Messages with the Kafka Upstream Plugin
+minimum_version: 3.10.x
+---
+
+
+Starting in {{site.base_gateway}} 3.10, the Kafka Upstream plugin supports customizing the Kafka message format.
+Using the configuration field `config.message_by_lua_functions`, you can define a function chain with multiple functions that can manipulate both the Kafka message format and the content of the message.
+
+## Prerequisites
+* You have created a Kafka topic
+* The [`kcat` utility](https://github.com/edenhill/kcat) is installed
+
+## Using the plugin
+
+The configuration field `message_by_lua_functions` is an array of strings, where each string is a customized Lua function. The default message generated by the Kafka Upstream plugin gets reduced through the following functions:
+
+```
+Message(default) -> f1(Message) -> f2(Message) -> ... -> Producer(New_Message)
+```
+
+The Lua function must be in the following format, where the function receives one message argument, and returns the new message as a single return value:
+
+```lua
+return function(message)
+    ... -- Some manipulation code
+    return new_message
+end
+```
+
+### Example: A function that does nothing
+
+You can just return the message argument as the return value, so that the function makes no modifications to the default message:
+
+```lua
+return function(message)
+    return message
+end
+```
+
+Let's configure this function with an example declarative config:
+
+```yaml
+_format_version: "3.0"
+services:
+  - name: kafka-service
+    url: http://mock-upstream
+    routes:
+      - name: kafka-route
+        paths:
+          - /kafka
+    plugins:
+      - name: kafka-upstream
+        config:
+          bootstrap_servers:
+            - host: localhost
+              port: 9092
+          topic: "my-topic"
+          forward_method: true
+          forward_uri: true
+          forward_headers: true
+          forward_body: true
+          message_by_lua_functions:
+            - 'return function(message) return message end'
+```
+
+Make sure the Kafka topic is created, then send a request to the route path and check the message:
+
+```sh
+curl http://localhost:8000/kafka
+```
+
+Response:
+```
+{"message":"message sent"}
+```
+
+You can use the `kcat` tool to check the message:
+```sh
+kcat -b localhost:9092 -G mygroup1 -o beginning -t my-topic -p 0 -C
+```
+
+You should see that the default message was sent to the topic:
+```json
+{"body_base64":true,"headers":{"accept":"*\/*","host":"localhost:8000","user-agent":"curl\/8.7.1"},"uri_args":{},"uri":"\/kafka","body_args":{},"body":"","method":"GET"}
+```
+
+### Example: A function that returns a fixed table
+
+In the custom function, users can return whatever data they want as the return value. 
+The returned value must be JSON-serializable, otherwise it can't be sent to Kafka.
+
+```lua
+return function(message)
+    return {a="1", b="2"}
+end
+```
+
+Configure this function with an example declarative config:
+
+```yaml
+_format_version: "3.0"
+services:
+  - name: kafka-service
+    url: http://mock-upstream
+    routes:
+      - name: kafka-route
+        paths:
+          - /kafka
+    plugins:
+      - name: kafka-upstream
+        config:
+          bootstrap_servers:
+            - host: localhost
+              port: 9092
+          topic: "my-topic"
+          message_by_lua_functions:
+            - 'return function(message) return {a="1", b="2"} end'
+```
+
+Make sure the Kafka topic is created, then send a request to the route path and check the message:
+
+```sh
+curl http://localhost:8000/kafka
+```
+
+Response:
+```json
+{"message":"message sent"}
+```
+
+You can use the `kcat` tool to check the message:
+```sh
+kcat -b localhost:9092 -G mygroup1 -o beginning -t my-topic -p 0 -C
+```
+
+You should see that the new message was sent to the topic:
+
+```json
+.....OLD MESSAGES.....
+% Reached end of topic my-topic [0] at offset 8
+{"b":"2","a":"1"}
+% Reached end of topic my-topic [0] at offset 9
+```
+
+You can also return the encoded message string as a return value:
+
+```lua
+return function(message)
+    return [[{"a":"1","b":"2"}]]
+end
+```
+
+The result should be the same as the one returned by the `{a="1", b="2"}` table.
+
+{:.important}
+> **Note**: The Kafka Upstream plugin only supports JSON encoding on the message. Other serialization methods are not supported.
+
+### Example: Functions that insert consumer info and redact a sensitive request header
+
+Let's use the following functions, which call the `get_consumer` PDK function and manipulate the request header information:
+
+```lua
+--- FUNCTION 1 that inject the consumer information to the message
+return function(message)
+    message.consumer = kong.client.get_consumer()
+
+    return message
+end
+```
+
+```lua
+--- FUNCTION 2 that clean an apikey header
+return function(message)
+    local headers = message and message.headers
+    local apikey = headers and headers.apikey
+    if apikey then
+        headers.apikey = "[REDACTED]"
+    end
+
+    return message
+end
+```
+
+Create an example declarative config for these two functions:
+
+```yaml
+_format_version: "3.0"
+
+services:
+  - name: kafka-service
+    url: http://mock-upstream
+    routes:
+      - name: kafka-route
+        paths:
+          - /kafka
+    plugins:
+      - name: key-auth
+      - name: kafka-upstream
+        config:
+          bootstrap_servers:
+            - host: localhost
+              port: 9092
+          topic: "my-topic"
+          forward_method: true
+          forward_uri: true
+          forward_headers: true
+          forward_body: true
+          message_by_lua_functions:
+            - "return function(message) message.consumer = kong.client.get_consumer() return message end"
+            - "return function(message) local headers = message and message.headers local apikey = headers and headers.apikey if apikey then headers.apikey = '[REDACTED]' end return message end"
+
+consumers:
+  - username: alice
+    keyauth_credentials:
+      - key: "example-api-key"
+```
+
+Trigger the API call and check the message:
+
+```sh
+curl http://localhost:8000/kafka -H 'apikey: example-api-key'
+```
+
+Response:
+```
+{"message":"message sent"}
+```
+
+You can use the `kcat` tool to check the message:
+```
+kcat -b localhost:9092 -G mygroup1 -o beginning -t my-topic -p 0 -C
+```
+
+You should see the `REDACTED` API key in the message:
+
+```json
+.....OLD MESSAGES.....
+% Reached end of topic my-topic [0] at offset 20
+{"body_base64":true,"method":"GET","consumer":{"id":"25f66f8c-b5e1-4bbd-9b17-2cbe733d1895","created_at":1741768764,"username":"alice","username_lower":"alice","type":0,"updated_at":1741768764},"body_args":{},"body":"","headers":{"apikey":"[REDACTED]","accept":"*\/*","host"
+:"localhost:8000","x-consumer-id":"25f66f8c-b5e1-4bbd-9b17-2cbe733d1895","x-credential-identifier":"795fb073-f698-403d-9302-209ba3abdd55","user-agent":"curl\/8.7.1","x-consumer-username":"alice"},"uri_args":{},"uri":"\/kafka"}
+% Reached end of topic my-topic [0] at offset 21
+```

app/_hub/kong-inc/kafka-upstream/how-to/_multiple-topics-and-acl.md

@@ -0,0 +1,100 @@
+---
+nav_title: Configure Multiple Allowed Producer Topics
+title: Configure Multiple Allowed Producer Topics with the Kafka Upstream Plugin
+minimum_version: 3.10.x
+---
+
+Starting in {{site.base_gateway}} 3.10, the Kafka Upstream plugin supports sending messages to multiple client-defined Kafka topics by using a query parameter that contains a list of target topic names. 
+To prevent client requests from sending messages to arbitrary topics, you can also define a topic allowlist.
+
+You can use the following fields in combination:
+* `config.topics_query_arg`:  Define a request parameter name that contains a list of the target topics that Kafka Upstream will produce messages to. 
+* `config.allowed_topics`: Define the list of allowed topic names to which messages can be sent.
+
+{:.note}
+> The default topic configured in the `topic` field is always allowed, regardless of its inclusion in `allowed_topics`. When `allowed_topics` is not defined, only the default topic configured in the `topic` field is allowed.
+
+## Prerequisites
+* You have created a Kafka topic
+* The [`kcat` utility](https://github.com/edenhill/kcat) is installed
+
+## Using the plugin
+
+Here is an example config that defines an allowlist and a topic list:
+
+```yaml
+_format_version: "3.0"
+
+services:
+  - name: kafka-service
+    url: http://mock-upstream
+    routes:
+      - name: kafka-route
+        paths:
+          - /kafka
+    plugins:
+      - name: key-auth
+      - name: kafka-upstream
+        config:
+          bootstrap_servers:
+            - host: localhost
+              port: 9092
+          topic: "my-topic"
+          topics_query_arg: "topic-list"
+          allowed_topics:
+            - "my-topic"
+            - "topic1"
+            - "topic2"
+
+consumers:
+  - username: alice
+    keyauth_credentials:
+      - key: "example-api-key"
+
+```
+
+A client can use the `topic-list` query argument to send the message to multiple topics, including any topics in the  `allowed_topic` list, as well as the default topic defined in the `topic` field:
+
+```sh
+curl 'http://localhost:8000/kafka?topic-list=my-topic,topic1'
+```
+
+Response:
+```
+{"message":"message sent"}
+```
+
+Check the message in the topics using the `kcat` tool:
+```sh
+kcat -b 127.0.0.1:9092 -G mygroup1 -t my-topic -p 0 -C
+```
+
+Since the client is allowed to send messages to this topic, you should see the following:
+```json
+.....OLD MESSAGES.....
+% Reached end of topic my-topic [0] at offset 22
+{"body_args":{},"body_base64":true,"body":""}
+% Reached end of topic my-topic [0] at offset 23
+
+> kcat -b 127.0.0.1:9092 -G mygroup -t topic1 -p 0 -C
+% Reached end of topic topic1 [0] at offset 0
+{"body_args":{},"body_base64":true,"body":""}
+% Reached end of topic topic1 [0] at offset 1
+```
+
+A client can't send the message to a topic that is not allowed in the `allowed_topic` list:
+
+```sh
+curl 'http://localhost:8000/kafka?topic-list=my-topic,topic1,topic3'
+```
+
+Response:
+```json
+{"message":"Bad Gateway","error":"one or more target topics are not in the allowed topics list"}
+```
+
+You will see the following in the proxy error log:
+```sh
+2025/03/12 17:07:28 [error] 56468#0: *6580 [kong] producers.lua:28 [kafka-upstream] sending message to the following topics is not allowed: (topic3), client: 127.0.0.1, server: kong, request: "GET /kafka?topic-list=my-topic,topic1,topic3 HTTP/1.1", host: "localhost:8000",
+request_id: "2535dd1cf1bee70828eaef039c1e1070"
+```

app/_hub/kong-inc/kafka-upstream/overview/_index.md

@@ -99,7 +99,9 @@ This plugin supports the following authentication mechanisms:
 Known limitations:
 
 1. Message compression is not supported.
+{% if_version lte: 3.9.x %}
 2. The message format is not customizable.
+{% endif_version %}
 
 ## Quickstart