[feature][connector-kafka] GAL-681 kafka connector 支持配置解析kafka record headers

2 files changed, 21 insertions, 13 deletions

diff --git a/docs/connector/sink/kafka.md b/docs/connector/sink/kafka.md
index 716a179..78b7f34 100644
--- a/docs/connector/sink/kafka.md
+++ b/docs/connector/sink/kafka.md
@@ -20,19 +20,20 @@ In order to use the Kafka connector, the following dependencies are required. Th
 
 Kafka sink custom properties. if properties belongs to Kafka Producer Config, you can use `kafka.` prefix to set.
 
-|                Name                |  Type   | Required | Default |                                                                                                                      Description                                                                                                                       |
-|------------------------------------|---------|----------|---------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| topic                              | String  | Yes      | (none)  | Topic name is required. It used to write data to kafka.                                                                                                                                                                                                |
-| kafka.bootstrap.servers            | String  | Yes      | (none)  | A list of host/port pairs to use for establishing the initial connection to the Kafka cluster. This list should be in the form `host1:port1,host2:port2,...`.                                                                                          |
-| log.failures.only                  | Boolean | No       | true    | Defines whether the producer should fail on errors, or only log them. If this is set to true, then exceptions will be only logged, if set to false, exceptions will be eventually thrown and cause the streaming program to fail (and enter recovery). |
-| format                             | String  | No       | json    | Data format. The default value is `json`. The Optional values are `json`, `protobuf`.                                                                                                                                                                  |
-| [format].config                    | /       | No       | (none)  | Data format properties. Please refer to [Format Options](../formats) for details.                                                                                                                                                                      |
-| rate.limiting.strategy             | String  | No       | (none)  | The rate limiting strategy to use. The Optional values are `none`, `sliding_window`.                                                                                                                                                                   |
-| rate.limiting.window.size          | Integer | No       | 5       | The window size of the rate limiting. For example, limit rate less than 10Mbps in 5 seconds time interval.                                                                                                                                             |
-| rate.limiting.limit.rate           | String  | No       | 10Mbps  | A maximum rate of traffic that can be transmitted over a network or between networks. The units of the bytes rate are Mbps, Kbps,and bps. For example, 10Mbps, 100Kbps, 1000bps.                                                                       |
-| rate.limiting.block.duration       | String  | No       | 5min    | If the rate limit is exceeded, the data will be blocked for the specified duration.  The units of the duration are seconds, minutes, and hours. For example, 10s, 1m, 1h.                                                                              |
-| rate.limiting.block.reset.duration | String  | No       | 30s     | The time interval for resetting the rate limit. The units of the duration are seconds, minutes, and hours. For example, 10s, 1m, 1h.                                                                                                                   |
-| kafka.config                       | /       | No       | (none)  | Kafka producer properties. Please refer to [Kafka Producer Config](https://kafka.apache.org/documentation/#producerconfigs) for details.                                                                                                               |
+| Name                                |  Type   | Required | Default | Description                                                                                                                                                                                                                                            |
+|-------------------------------------|---------|----------|---------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| topic                               | String  | Yes      | (none)  | Topic name is required. It used to write data to kafka.                                                                                                                                                                                                |
+| kafka.bootstrap.servers             | String  | Yes      | (none)  | A list of host/port pairs to use for establishing the initial connection to the Kafka cluster. This list should be in the form `host1:port1,host2:port2,...`.                                                                                          |
+| log.failures.only                   | Boolean | No       | true    | Defines whether the producer should fail on errors, or only log them. If this is set to true, then exceptions will be only logged, if set to false, exceptions will be eventually thrown and cause the streaming program to fail (and enter recovery). |
+| format                              | String  | No       | json    | Data format. The default value is `json`. The Optional values are `json`, `protobuf`.                                                                                                                                                                  |
+| [format].config                     | Map     | No       | (none)  | Data format properties. Please refer to [Format Options](../formats) for details.                                                                                                                                                                      |
+| headers.config                      | Map     | No       | (none)  | Kafka record headers info. exp: 'headers.key: value' will put key and value into record headers.                                                                                                                                                       |
+| rate.limiting.strategy              | String  | No       | (none)  | The rate limiting strategy to use. The Optional values are `none`, `sliding_window`.                                                                                                                                                                   |
+| rate.limiting.window.size           | Integer | No       | 5       | The window size of the rate limiting. For example, limit rate less than 10Mbps in 5 seconds time interval.                                                                                                                                             |
+| rate.limiting.limit.rate            | String  | No       | 10Mbps  | A maximum rate of traffic that can be transmitted over a network or between networks. The units of the bytes rate are Mbps, Kbps,and bps. For example, 10Mbps, 100Kbps, 1000bps.                                                                       |
+| rate.limiting.block.duration        | String  | No       | 5min    | If the rate limit is exceeded, the data will be blocked for the specified duration.  The units of the duration are seconds, minutes, and hours. For example, 10s, 1m, 1h.                                                                              |
+| rate.limiting.block.reset.duration  | String  | No       | 30s     | The time interval for resetting the rate limit. The units of the duration are seconds, minutes, and hours. For example, 10s, 1m, 1h.                                                                                                                   |
+| kafka.config                        | Map     | No       | (none)  | Kafka producer properties. Please refer to [Kafka Producer Config](https://kafka.apache.org/documentation/#producerconfigs) for details.                                                                                                               |
 
 ## Example
 
diff --git a/docs/connector/source/kafka.md b/docs/connector/source/kafka.md
index 07dff22..680d1c1 100644
--- a/docs/connector/source/kafka.md
+++ b/docs/connector/source/kafka.md
@@ -24,6 +24,13 @@ Kafka source custom properties. if properties belongs to Kafka Consumer Config,
 | [format].config         | Map    | No       | (none)  | Data format properties. Please refer to [Format Options](../formats) for details.                                                                             |
 | kafka.config            | Map    | No       | (none)  | Kafka consumer properties. Please refer to [Kafka Consumer Config](https://kafka.apache.org/documentation/#consumerconfigs) for details.                      |
 
+## Internal Fields
+
+| Name        | Type                | Description                         |
+|-------------|---------------------|-------------------------------------|
+| __timestamp | Long                | The timestamp of this kafka record. |
+| __headers   | Map[String, String] | The headers of this kafka record.   |
+
 ## Example
 
 This example read data of kafka topic `SESSION-RECORD` and print to console.