twalthr commented on a change in pull request #14156:
URL: https://github.com/apache/flink/pull/14156#discussion_r528576749
##########
File path: docs/dev/table/connectors/kafka.md
##########
@@ -189,6 +362,130 @@ Connector Options
Features
----------------
+
+### Key and Value Formats
+
+Both the key and value part of a Kafka record can be serialized to and
deserialized from raw bytes using
+one of the given [formats]({% link dev/table/connectors/formats/index.md %}).
+
+**Value Format**
+
+Since a key is optional in Kafka records, the following statement reads and
writes records with a configured
+value format but without a key format. The `'format'` option is a synonym for
`'value.format'`. All format
+options are prefixed with the format identifier.
+
+<div class="codetabs" markdown="1">
+<div data-lang="SQL" markdown="1">
+{% highlight sql %}
+CREATE TABLE KafkaTable (,
+ `ts` TIMESTAMP(3) METADATA FROM 'timestamp',
+ `user_id` BIGINT,
+ `item_id` BIGINT,
+ `behavior` STRING
+) WITH (
+ 'connector' = 'kafka',
+ ...
+
+ 'format' = 'json',
+ 'json.ignore-parse-errors' = 'true'
+)
+{% endhighlight %}
+</div>
+</div>
+
+The value format will be configured with the following data type:
+
+{% highlight text %}
+ROW<`user_id` BIGINT, `item_id` BIGINT, `behavior` STRING>
+{% endhighlight %}
+
+**Key and Value Format**
+
+The following example shows how to specify and configure key and value
formats. The format options are
+prefixed with either the `'key'` or `'value'` plus format identifier.
+
+<div class="codetabs" markdown="1">
+<div data-lang="SQL" markdown="1">
+{% highlight sql %}
+CREATE TABLE KafkaTable (
+ `ts` TIMESTAMP(3) METADATA FROM 'timestamp',
+ `user_id` BIGINT,
+ `item_id` BIGINT,
+ `behavior` STRING
+) WITH (
+ 'connector' = 'kafka',
+ ...
+
+ 'key.format' = 'json',
+ 'key.json.ignore-parse-errors' = 'true',
+ 'key.fields' = 'user_id;item_id',
+
+ 'value.format' = 'json',
+ 'value.json.fail-on-missing-field' = 'false',
+ 'value.fields-include' = 'ALL'
+)
+{% endhighlight %}
+</div>
+</div>
+
+The key format includes the fields listed in `'key.fields'` (in the same
order). Thus, it will be configured with
Review comment:
The delimiter is a regular Flink string lists delimiter. But I can also
mention it explicitly.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
