Syntax

**CONNECTION** _connection_name_ | The name of the Kafka connection to use in the source. For details on creating connections, check the CREATE CONNECTION documentation page. **INCLUDE KEY** | Include a column containing the Kafka message key. If the key is encoded using a format that includes schemas the column will take its name from the schema. For unnamed formats (e.g. TEXT), the column will be named key. The column can be renamed with the optional **AS** *name* statement. **INCLUDE PARTITION** | Include a partition column containing the Kafka message partition. The column can be renamed with the optional **AS** *name* clause. **INCLUDE OFFSET** | Include an offset column containing the Kafka message offset. The column can be renamed with the optional **AS** *name* clause. **INCLUDE TIMESTAMP** | Include a timestamp column containing the Kafka message timestamp. The column can be renamed with the optional **AS** *name* clause.

Note that the timestamp of a Kafka message depends on how the topic and its producers are configured. See the Confluent documentation for details. **INCLUDE HEADERS** | Include a headers column containing the Kafka message headers as a list of records of type (key text, value bytea). The column can be renamed with the optional **AS** *name* clause. **INCLUDE HEADER** _key_ **AS** _name_ [**BYTES**] | Include a _name_ column containing the Kafka message header _key_ parsed as a UTF-8 string. To expose the header value as bytea, use the BYTES option.