# Kafka

O **Kafka** produz registros para os *broker*s Kafka configurados nele.

## Parâmetros

Dê uma olhada nas opções de configuração do componente. Parâmetros suportados por [expressões *Double Braces*](/documentation/connectors-and-triggers/pt-br/double-braces/overview.md) estão marcados com `(DB)`.

<table data-full-width="true"><thead><tr><th>Parâmetro</th><th>Descrição</th><th>Valor padrão</th><th>Tipo de dado</th></tr></thead><tbody><tr><td><strong>Kafka Authentication Account</strong></td><td>Se o servidor Kafka precisar de autenticação, será necessário criar uma conta tipo <em>Basic</em> para esse componente. Suportamos também autenticação via Kerberos.</td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Truststore</strong></td><td>Caso seja necessário informar uma <em>truststore</em> para realizar o SSL Handshake utilizando certificados privados, deve-se criar uma conta do tipo <em>Certificate Chain</em> e informar os certificados concatenados. É opcional inserir a senha a ser cadastrada na criação da <em>truststore</em>, no campo <em>password</em>.  </td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Keystore</strong></td><td>Caso seja necessário informar uma <em>keystore</em> para realizar a autenticação SSL mútua, deve-se criar uma conta do tipo <em>Certificate Chain</em>, informar a cadeia completa com os certificados concatenados e a chave privada a ser utilizada para a autenticação SSL mútua. Caso exista uma chave privada, é necessário informá-la no campo <em>password</em>.</td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Brokers</strong></td><td><em>Brokers</em> do servidor (HOST: PORT) usados para o envio de registros. Para informar múltiplos HOSTS, você pode separá-los por vírgula. Exemplo: HOST1:PORT1,HOST2:PORT2,...,HOSTn:PORTn</td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Security Protocol</strong></td><td>A forma que a conexão é estabelecida. É opcional utilizar um canal de segurança (SSL) e de autenticação (SASL). A utilização de ambos (SASL_SSL) também é possível.</td><td>SSL</td><td><em>String</em></td></tr><tr><td><strong>Topic Name</strong></td><td>Nome do tópico do Kafka.</td><td>{{ DEFAULT(message.topic, "new-topic") }}</td><td><em>String</em></td></tr><tr><td><strong>Schema Registry URL</strong></td><td>Se pelo menos uma das opções <strong>Headers By Avro, Payload As Avro</strong> e <strong>Partition Key As Avro</strong> estiver ativada, o campo será mostrado para configurar o <em>Schema Registry's URL</em>.</td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Schema Registry Account</strong></td><td>Conta para autenticar com o <em>Schema Registry</em>.</td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Schema Registry Truststore</strong></td><td>Se for necessário informar um <em>truststore</em> para fazer o <em>SSL Handshake</em> utilizando certificados privados, deve-se criar uma conta do tipo <em>Certificate Chain</em> e informar os certificados concatenados. É opcional informar a senha a ser cadastrada na criação do <em>truststore</em>, no campo <em>password</em>.</td><td>N/A</td><td>N/A</td></tr><tr><td><strong>Schema Registry Keystore</strong></td><td>Se for necessário informar um <em>keystore</em> para fazer a autenticação mútua SSL, deve-se criar uma conta do tipo <em>Certificate Chain</em>, informar a cadeia completa com os certificados concatenados e a chave privada a ser utilizada para a autenticação mútua SSL. Se houver chave privada, é necessário informá-la no campo <em>password</em>.</td><td>N/A</td><td>N/A</td></tr><tr><td><strong>Headers</strong></td><td>Conjunto de entradas chave-valor (<em>key-value</em>), contendo cabeçalhos a serem enviados na mensagem (campo opcional).</td><td>N/A</td><td><em>Key-value</em></td></tr><tr><td><strong>Binary Headers</strong></td><td>Se a opção estiver ativada, os valores dos cabeçalhos são considerados binários e são interpretados como uma <em>string</em> com a representação <em>base64</em>; do contrário, os valores dos cabeçalhos são interpretados como texto.</td><td><em>False</em></td><td>Booleano</td></tr><tr><td><strong>Headers By Avro Schema</strong></td><td>Se a opção estiver ativada, o componente validará os cabeçalhos com base em um esquema Avro antes de enviar os cabeçalhos.</td><td><em>False</em></td><td>Booleano</td></tr><tr><td><strong>Headers Schema</strong></td><td>Se a opção <strong>Headers By Avro Schema</strong> estiver ativada, será exibido o campo para definir os <em>Headers Schemas</em> a serem validados.</td><td>N/A</td><td>N/A</td></tr><tr><td><strong>Headers Charset</strong></td><td>Nome do código de caracteres para a codificação dos valores dos cabeçalhos (padrão UTF-8).</td><td>UTF-8</td><td><em>String</em></td></tr><tr><td><strong>Payload</strong></td><td><em>Payload</em> que será despachado.</td><td>{{ message.$ }}</td><td><em>String</em></td></tr><tr><td><strong>Payload As Avro</strong></td><td>Se a opção estiver ativada, o componente enviará o <em>payload</em> no formato Avro.</td><td><em>False</em></td><td>Booleano</td></tr><tr><td><strong>Payload Schema</strong></td><td>Este campo fica disponível somente se a opção <strong>Payload As Avro</strong> estiver ativada e informa o <em>Payload Schema</em> a ser validado.</td><td>N/A</td><td>N/A</td></tr><tr><td><strong>Request Timeout</strong></td><td>Configuração que controla o tempo máximo (em milissegundos) que o cliente aguarda pela resposta de uma solicitação. Se a resposta não for recebida antes que o tempo limite se esgote, a solicitação é reenviada automaticamente. Do contrário, haverá uma falha se as tentativas se esgotarem.</td><td>60000</td><td>Inteiro</td></tr><tr><td><strong>Retries</strong></td><td>Se for estabelecido um valor diferente de 0 (zero), qualquer registro cujo envio falhar será reenviado. Esses registros podem ser reenviados com um provável erro transitório.</td><td>N/A</td><td>Inteiro</td></tr><tr><td><strong>Metadata Timeout</strong></td><td>Tempo máximo para o envio do registro ao Kafka.</td><td>5000</td><td>Inteiro</td></tr><tr><td><strong>Key Strategy</strong></td><td>Caso a opção <strong>Partition Key As Avro</strong> esteja ativada, o campo será exibido para configuração da estratégia de <em>subject</em> a ser utilizada para construir o <em>subject name</em> para as <em>keys</em> das mensagens.</td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Value Strategy</strong></td><td>Caso a opção <strong>Payload as Avro</strong> esteja ativada, o campo será exibido para configuração da estratégia de <em>subject</em> a ser utilizada para construir o <em>subject name</em> para os <em>values</em> das mensagens.</td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Fail On Error</strong></td><td>Se a opção estiver ativada, a execução do <em>pipeline</em> com erro será interrompida; do contrário, a execução do <em>pipeline</em> continua, mas o resultado vai mostrar um valor falso para a propriedade "<em>success</em>".</td><td><em>False</em></td><td>Booleano</td></tr><tr><td><strong>Advanced Settings</strong></td><td>Se a opção estiver ativada, você pode acessar as seguintes configurações:</td><td><em>False</em></td><td>Booleano</td></tr><tr><td><strong>Kerberos Service Name</strong></td><td>Valor definido na propriedade <strong>sasl.kerberos.service.name</strong> configurado no lado <em>server</em> do <em>broker</em> Kafka.</td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Partition Number</strong></td><td>Especifica os números da partição para os quais o <a href="/pages/Z9T1DgqB4iuS1UijeSUy"><strong>Kafka Trigger</strong></a> enviará mensagens. Caso essa propriedade não seja configurada, o servidor do Kafka ficará responsável por decidir para qual partição do tópico a mensagem será enviada.</td><td>N/A</td><td>Inteiro</td></tr><tr><td><strong>Partition Key</strong></td><td>Uma chave de partição pode ser especificada para indicar a partição de destino da mensagem. Caso esse campo não seja preenchido, um particionador baseado em <em>hashing</em> é utilizado para determinar a id de partição dada a chave.</td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Partition Key As Avro</strong></td><td>Se a opção estiver ativada, o componente fará o envio da chave de partição no formato Avro.</td><td>N/A</td><td>Booleano</td></tr><tr><td><strong>Partition Key Schema</strong></td><td>Caso a opção <strong>Partition Key As Avro</strong> esteja ativada, o campo será exibido para que seja informado o <em>Schema</em> da chave de partição a ser validada.</td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Producer Client Name</strong></td><td>Identificador de origem dos pedidos (opcional).</td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>ACKS</strong></td><td>Configuração para reconhecer o recebimento da mensagem pelo <em>broker</em> Kafka (valores: 0, 1 ou ALL).</td><td>1</td><td>Inteiro</td></tr><tr><td><strong>Use Dynamic Account</strong></td><td>Quando a opção estiver ativada, o componente irá usar a conta dinamicamente. Defina o nome da conta para permitir um uso flexível. Quando estiver desativada, a conta será usada estaticamente.</td><td><em>False</em></td><td>Booleano</td></tr><tr><td><strong>Account Name</strong></td><td>Nome da conta <em>Basic</em> que é gerada dinamicamente através do <a href="/pages/iulYYuz29bvfGjceZb4e"><strong>componente Store Account</strong></a>.</td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Scoped</strong></td><td>Quando a opção estiver ativada, a conta armazenada é isolada para outro sub-processo. Nesse caso, os sub-processos verão sua própria versão dos dados da conta armazenada.</td><td>N/A</td><td>Booleano</td></tr><tr><td><strong>Truststore Account Name</strong></td><td>Nome da conta <em>truststore</em> que é gerada dinamicamente através do componente <strong>Store Account</strong>. Contas suportadas: <em>Certificate Chain.</em></td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Keystore Account Name</strong></td><td>Nome da conta <em>keystore</em> que é gerada dinamicamente através do componente <strong>Store Account</strong>. Contas suportadas: <em>Certificate Chain.</em></td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Account Name for Registry</strong></td><td>Nome da conta para registro que é gerada dinamicamente através do componente <strong>Store Account</strong>. Contas suportadas: <em>Basic</em> ou <em>Oauth Bearer Token</em>.</td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Truststore Account Name for Registry</strong></td><td>Nome da conta <em>truststore</em> para registro que é gerada dinamicamente através do componente <strong>Store Account</strong>. Contas suportadas: <em>Certificate Chain</em>.</td><td>N/A</td><td><em>String</em></td></tr><tr><td><strong>Keystore Account Name for Registry</strong></td><td>Nome da conta <em>keystore</em> para registro que é gerada dinamicamente através do componente <strong>Store Account</strong>. Contas suportadas: <em>Certificate Chain</em>.</td><td>N/A</td><td><em>String</em></td></tr></tbody></table>

{% hint style="info" %}
**Informações importantes:**&#x20;

* Devido a necessidade de uma grande alocação de memória, não suportamos os seguintes tipos de *Security Protocol*: PLAINTEXT e SASL\_PLAINTEXT. Para mais informações, [visite a documentação externa do Apache Kafka](https://cwiki.apache.org/confluence/display/KAFKA/KIP-498%3A+Add+client-side+configuration+for+maximum+response+size+to+protect+against+OOM).
* As mensagens trafegadas no formato Avro deverão ser do tamanho máximo suportado por Pipelines SMALL, MEDIUM ou LARGE. O componente não suporta cenários extremos de leitura de dezenas de mega/giga/tera/peta bytes. O suporte para o formato Avro está na fase Beta.
* Atualmente, os parâmetros **Use Dynamic Account,** **Account Name**, **Scoped**, **Truststore Account Name**, **Keystore Account Name**,**Account Name for Registry**, **Truststore Account Name for Registry** e **Keystore Account Name for Registry** podem ser usados apenas no Pipeline Engine v2.
  {% endhint %}

## Exemplo de resposta de requisição ao Kafka <a href="#exemplo-de-resposta-de-requisio-ao-kafka" id="exemplo-de-resposta-de-requisio-ao-kafka"></a>

```
{
  "message": "{}",
  "offset": 201,
  "timestamp": 1585168528773,
  "serializedKeySize": -1,
  "serializedValueSize": 2,
  "topic": "Welcome-Kafka",
  "partition": 1,
  "success": verdadeiro
}
```

* **message:** mensagem enviada.
* **offset:** *offset* do registro no tópico/partição.
* **timestamp:** horário/data do registro no tópico/partição.
* **serializedKeySize:** tamanho da chave serializada, com valor descomprimido em bytes. Se o valor é nulo, o tamanho devolvido é -1.
* **serializedValueSize:** tamanho do valor serializado, com valor descomprimido em bytes. Se o valor é nulo, o tamanho devolvido é -1.
* **topic:** nome do tópico.
* **partition:** partição para onde o registro foi enviado.
* **success:** se "verdadeiro", o envio foi realizado com sucesso.

## Fluxo de Mensagens <a href="#fluxo-de-mensagens" id="fluxo-de-mensagens"></a>

### Entrada <a href="#entrada" id="entrada"></a>

O componente aceita qualquer mensagem de entrada e pode fazer uso dela através de *Double Braces*.      &#x20;

### Saída <a href="#sada" id="sada"></a>

O componente não altera nenhuma informação da mensagem de entrada. Portanto, ela é retornada para o componente seguinte ou é utilizada como resposta final se este componente for o último passo do *pipeline*.           &#x20;

## Kafka em Ação <a href="#kafka-em-ao" id="kafka-em-ao"></a>

### Autenticação utilizando SSL ou SASL <a href="#autenticao-utilizando-ssl-ou-sasl" id="autenticao-utilizando-ssl-ou-sasl"></a>

Isso permite a autenticação dos seus produtores e clientes ao *cluster* do Kafka (verificação de identidade). Essa também é uma forma segura de permitir que os seus clientes confirmem a identidade.

### Autenticação usando Kerberos <a href="#autenticao-usando-kerberos" id="autenticao-usando-kerberos"></a>

Para utilizar a autenticação via Kerberos no **Kafka** é necessário ter cadastrado o arquivo de configuração “krb5.conf” no parâmetro de *Realm*. Caso não tenha feito isso, acione o nosso suporte via *chat*. Após concluir esse passo, basta configurar corretamente uma conta do tipo Kerberos e utilizá-la no componente.


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://docs.digibee.com/documentation/connectors-and-triggers/pt-br/connectors/queues-and-messaging/kafka.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.