Tratamento de erros e retomada das propriedades de interrupção

Visão geral

Use as seguintes definições de configuração para especificar como o conector de origem do MongoDB Kafka se comporta quando encontra erros e para especificar as configurações relacionadas à retomada de leituras interrompidas.

Configurações

Nome	Descrição
mongo.errors.tolerance	Tipo: string Descrição: se deve continuar processando mensagens quando o conector encontrar um erro. Defina isso como `"none"` se quiser que o conector pare de processar mensagens e relate o problema se encontrar um erro. Defina isso `"all"` como se quiser que o conector continue processando mensagens e ignore os erros encontrados. IMPORTANTE: esta propriedade substitui a propriedade errors.tolerance Connect Framework . Padrão: `"none"` Valores aceitos: `"none"` ou `"all"`
mongo.errors.log.enable	Tipo: booleano Descrição: se o conector deve relatar erros no arquivo de log. Defina isso como `true` para registrar todos os erros encontrados pelo conector. Configure isto para `false` para registrar erros que não são tolerados pelo conector. Você pode especificar quais erros o conector deve tolerar usando a configuração `errors.tolerance` `mongo.errors.tolerance` ou. IMPORTANTE: esta propriedade substitui a propriedade errors.log.enable Connect Framework . Padrão: `false` Valores aceitos: `true` ou `false`
mongo.errors.deadletterqueue.topic.name	Tipo: string Descrição: o nome do tópico a ser usado como fila de mensagens não entregues (DLQ). Se você especificar um valor, o conector gravará mensagens inválidas no tópico da fila de mensagens não entregues (DLQ) como strings JSON estendidas. Se você deixar essa configuração em branco, o conector não gravará mensagens inválidas em nenhum tópico. IMPORTANTE: você deve definir `errors.tolerance` `mongo.errors.tolerance` `"all"` a configuração ou como para habilitar essa propriedade. Padrão: `""` Valores aceitos: um nome de tópico Kafka válido
offset.partition.name	Tipo: string Descrição: o nome da partição offset personalizada a ser usada. Você pode usar essa opção para instruir o conector a iniciar um novo fluxo de alterações quando um deslocamento existente contiver um token de currículo inválido. Se você deixar essa configuração em branco, o conector usará o nome da partição padrão com base nos detalhes da conexão. Para exibir uma estratégia para nomear partições de deslocamento, consulte Redefinir deslocamentos armazenados. Padrão: `""` Valores aceitos: uma string. Para saber mais sobre como nomear uma partição, consulte SourceRecord na documentação da API do Apache Kafka.
heartbeat.interval.ms	Tipo: longo Descrição: o número de milissegundos que o conector espera entre o envio de mensagens de pulsação. O conector envia mensagens de pulsação quando os registros de origem não são publicados no intervalo especificado. Este mecanismo melhora a capacidade de retomada do conector para namespaces de baixo volume. As mensagens de pulsação contêm um `postBatchResumeToken` campo de dados. O valor deste campo contém a entrada de oplog do servidor MongoDB que o conector leu pela última vez a partir do fluxo de alteração. Defina isso como `0` para desabilitar mensagens de pulsação. Para saber mais, consulte Prevenção na página Token de retomada inválido. Padrão: `0` Valores aceitos: um número inteiro
heartbeat.topic.name	Tipo: string Descrição: o nome do tópico no qual o conector deve publicar mensagens de pulsação. Você deve fornecer um valor positivo na configuração `heartbeat.interval.ms` para ativar esse recurso. Padrão: `__mongodb_heartbeats` Valores aceitos: um nome de tópico Kafka válido

Heartbeats com transformações de mensagem única

Se você habilitar pulsações e especificar Transformações de Mensagem Única (SMTs) em sua implantação do Kafka Connect, deverá excluir suas mensagens de pulsação dos SMTs. Os SMTs são uma funcionalidade do Kafka Connect que permite a você especificar transformações nas mensagens que passam pelo connector de origem sem ter que implantar uma aplicação de Atlas Stream Processing.

Para excluir mensagens de pulsação de seus SMTs, você deve criar e aplicar um predicado aos seus SMTs. Os predicados são um recurso dos SMTs que permite verificar se uma mensagem corresponde a uma declaração condicional antes de aplicar uma transformação.

A seguinte configuração define o predicado IsHeartbeat que corresponde a mensagens de pulsação enviadas para o tópico de pulsação padrão:

predicates=IsHeartbeat
predicates.IsHeartbeat.type=org.apache.kafka.connect.transforms.predicates.TopicNameMatches
predicates.IsHeartbeat.pattern=__mongodb_heartbeats

A seguinte configuração utiliza o predicado anterior para excluir mensagens de pulsação de uma transformação ExtractField :

transforms=Extract
transforms.Extract.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.Extract.field=<the field to extract from your Apache Kafka key>
transforms.Extract.predicate=IsHeartbeat
transforms.Extract.negate=true
# apply the default key schema as the extract transformation requires a struct object
output.format.key=schema

Se você não excluir suas mensagens de pulsação da transformação anterior, o connector gerará o seguinte erro após processar uma mensagem de pulsação:

ERROR WorkerSourceTask{id=mongo-source-0} Task threw an uncaught and unrecoverable exception. Task is being killed ...
...
Only Struct objects supported for [field extraction], found: java.lang.String

Para saber mais sobre SMTs, consulte Como usar transformações de mensagem única no Kafka Connect da Confluent.

Para saber mais sobre predicados, consulte Filtro (Apache Kafka) do Confluent.

Para saber mais sobre a transformação ExtractField, consulte ExtrairField do Confluent.

Para saber mais sobre o esquema de chave padrão, consulte a página Esquemas Padrão .

Voltar

Inicialização

Todas as propriedades