`). A configuração global é aplicada primeiro e, em seguida, a configuração no nível do tópico é aplicada (se existir).
```xml theme={null}
cgrp
3000
logs
4000
smallest
logs
100000
stats
50000
logs
250
stats
400
```
Para ver uma lista das opções de configuração disponíveis, consulte a [referência de configuração do librdkafka](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md). Use o sublinhado (`_`) em vez do ponto na configuração do ClickHouse. Por exemplo, `check.crcs=true` será `true`.
### Suporte a Kerberos
Para lidar com o Kafka com suporte a Kerberos, adicione o elemento filho `security_protocol` com o valor `sasl_plaintext`. Isso é suficiente se o ticket de concessão de tickets do Kerberos for obtido e armazenado em cache pelo sistema operacional.
O ClickHouse consegue manter credenciais do Kerberos usando um arquivo keytab. Considere os elementos filhos `sasl_kerberos_service_name`, `sasl_kerberos_keytab` e `sasl_kerberos_principal`.
Exemplo:
```xml theme={null}
SASL_PLAINTEXT
/home/kafkauser/kafkauser.keytab
kafkauser/kafkahost@EXAMPLE.COM
```
## Colunas virtuais
* `_topic` — Tópico do Kafka. Tipo de dado: `LowCardinality(String)`.
* `_key` — Chave da mensagem. Tipo de dado: `String`.
* `_offset` — Offset da mensagem. Tipo de dado: `UInt64`.
* `_timestamp` — Timestamp da mensagem. Tipo de dado: `Nullable(DateTime)`.
* `_timestamp_ms` — Timestamp da mensagem em milissegundos. Tipo de dado: `Nullable(DateTime64(3))`.
* `_partition` — Partição do tópico do Kafka. Tipo de dado: `UInt64`.
* `_headers.name` — Array das chaves dos cabeçalhos da mensagem. Tipo de dado: `Array(String)`.
* `_headers.value` — Array dos valores dos cabeçalhos da mensagem. Tipo de dado: `Array(String)`.
Colunas virtuais adicionais quando `kafka_handle_error_mode='stream'`:
* `_raw_message` - Mensagem bruta que não pôde ser analisada com sucesso. Tipo de dado: `String`.
* `_error` - Mensagem de exceção ocorrida durante uma falha na análise. Tipo de dado: `String`.
Observação: as colunas virtuais `_raw_message` e `_error` são preenchidas apenas em caso de exceção durante a análise; elas ficam sempre vazias quando a mensagem é analisada com sucesso.
## Mapeamento de colunas para metadados de mensagens do Kafka
Ao produzir mensagens com `INSERT INTO`, o motor Kafka sempre usa uma coluna chamada `_key` (do tipo `String`) como a chave da mensagem do Kafka e uma coluna chamada `_timestamp` (do tipo `DateTime`) como o timestamp da mensagem do Kafka — se essas colunas existirem na tabela. Por padrão, essas colunas também aparecem no payload da mensagem junto com as outras colunas.
Com `kafka_map_virtual_columns_on_write = 1`, o comportamento muda:
* `_key` (tipo `String`) — mapeada para a chave da mensagem do Kafka.
* `_timestamp` (tipo `DateTime`) — mapeada para o timestamp da mensagem do Kafka.
* `_headers.name` (tipo `Array(String)`) e `_headers.value` (tipo `Array(String)`) — mapeados para os cabeçalhos da mensagem do Kafka. Cada par `(_headers.name[i], _headers.value[i])` se torna um cabeçalho do Kafka. Como `_headers.name` e `_headers.value` compartilham o prefixo Nested `_headers`, o ClickHouse exige que ambos os arrays tenham o mesmo tamanho em cada linha.
As colunas com esses nomes são **excluídas do payload da mensagem** somente se seus tipos corresponderem aos listados acima; caso contrário, elas permanecem no payload, de modo que schemas que por acaso reutilizem esses nomes para dados não relacionados continuem funcionando.
Exemplo:
```sql theme={null}
CREATE TABLE kafka_out
(
event_json String,
`_key` String,
`_timestamp` DateTime,
`_headers.name` Array(String),
`_headers.value` Array(String)
)
ENGINE = Kafka
SETTINGS
kafka_broker_list = 'broker:9092',
kafka_topic_list = 'events',
kafka_group_name = 'events-producer',
kafka_format = 'JSONEachRow',
kafka_map_virtual_columns_on_write = 1;
INSERT INTO kafka_out VALUES
('{"a":1}', 'session-42', now(), ['source', 'trace_id'], ['api', 'abc-123']);
```
A mensagem produzida no Kafka tem payload `{"event_json":"{\"a\":1}"}`, chave `session-42`, o timestamp atual e dois cabeçalhos `source=api` e `trace_id=abc-123`.
## Suporte a formatos de dados
O motor Kafka oferece suporte a todos os [formatos](/pt-BR/reference/formats) compatíveis com o ClickHouse.
O número de linhas em uma mensagem do Kafka depende de o formato ser baseado em linhas ou em blocos:
* Para formatos baseados em linhas, o número de linhas em uma mensagem do Kafka pode ser controlado pela configuração `kafka_max_rows_per_message`.
* Para formatos baseados em blocos, não é possível dividir um bloco em partes menores, mas o número de linhas em um bloco pode ser controlado pela configuração geral [max\_block\_size](/pt-BR/reference/settings/session-settings#max_block_size).
## Engine para armazenar offsets confirmados no ClickHouse Keeper
Se `allow_experimental_kafka_offsets_storage_in_keeper` estiver habilitada, mais duas configurações poderão ser especificadas para o motor de tabela Kafka:
* `kafka_keeper_path` especifica o caminho da tabela no ClickHouse Keeper
* `kafka_replica_name` especifica o nome da réplica no ClickHouse Keeper
As duas configurações devem ser especificadas juntas, ou nenhuma delas. Quando ambas são especificadas, um novo engine Kafka experimental passa a ser usado. Esse novo engine não depende do armazenamento dos offsets confirmados no Kafka, mas os armazena no ClickHouse Keeper. Ele ainda tenta fazer commit dos offsets no Kafka, mas só depende desses offsets quando a tabela é criada. Em qualquer outra situação (se a tabela for reiniciada ou recuperada após algum erro), os offsets armazenados no ClickHouse Keeper serão usados para continuar consumindo mensagens. Além do offset confirmado, ele também armazena quantas mensagens foram consumidas no último lote, portanto, se o insert falhar, a mesma quantidade de mensagens será consumida, o que permite a desduplicação, se necessário.
Exemplo:
```sql theme={null}
CREATE TABLE experimental_kafka (key UInt64, value UInt64)
ENGINE = Kafka('localhost:19092', 'my-topic', 'my-consumer', 'JSONEachRow')
SETTINGS
kafka_keeper_path = '/clickhouse/{database}/{uuid}',
kafka_replica_name = '{replica}'
SETTINGS allow_experimental_kafka_offsets_storage_in_keeper=1;
```
### Limitações conhecidas
Como o novo motor é experimental, ele ainda não está pronto para produção. Existem algumas limitações conhecidas na implementação:
* Excluir e recriar a tabela rapidamente, ou especificar o mesmo caminho do ClickHouse Keeper para motores diferentes, pode causar problemas. Como prática recomendada, você pode usar `{uuid}` em `kafka_keeper_path` para evitar conflitos de caminho.
* Para garantir leituras repetíveis, as mensagens não podem ser consumidas de várias partições em uma única thread. Por outro lado, os consumidores do Kafka precisam ser consultados regularmente para continuarem ativos. Como resultado dessas duas exigências, decidimos permitir a criação de vários consumidores apenas se `kafka_thread_per_consumer` estiver ativado; caso contrário, é muito complicado evitar problemas relacionados à consulta regular dos consumidores.
**Veja também**
* [Colunas virtuais](/pt-BR/reference/engines/table-engines#table_engines-virtual_columns)
* [background\_message\_broker\_schedule\_pool\_size](/pt-BR/reference/settings/server-settings/settings#background_message_broker_schedule_pool_size)
* [system.kafka\_consumers](/pt-BR/reference/system-tables/kafka_consumers)