Propiedades de la gestión de errores y la reanudación tras una interrupción

Overview

Utiliza los siguientes ajustes de configuración para especificar cómo debe comportarse el conector de origen MongoDB Kafka cuando encuentre errores y para especificar los ajustes relacionados con la reanudación de lecturas interrumpidas.

Configuraciones

Nombre	Descripción
mongo.errors.tolerance	Tipo: cadena Descripción: Indica si se debe continuar procesando mensajes cuando el conector encuentra un error. Establezca este valor en `"none"` si desea que el conector deje de procesar mensajes e informe del problema si encuentra un error. Establezca este valor en `"all"` si desea que el conector continúe procesando mensajes e ignore cualquier error que encuentre. IMPORTANTE: Esta propiedad anula la propiedad errors.tolerance del marco de trabajo Connect. `"none"` Valor predeterminado: Valores aceptados: `"none"` o `"all"`
mongo.errors.log.enable	Tipo: booleano Descripción: Indica si el conector debe informar de los errores en el archivo de registro. Establezca este valor en para `true` registrar todos los errores que encuentre el conector. Establezca este valor en `false` para registrar los errores que el conector no tolera. Puede especificar qué errores debe tolerar el conector mediante la configuración `errors.tolerance` `mongo.errors.tolerance` o. IMPORTANTE: Esta propiedad anula la propiedad errors.log.enable del marco de trabajo Connect. `false` Valor predeterminado: Valores aceptados: `true` o `false`
mongo.errors.deadletterqueue.tema.name	Tipo: cadena Descripción: Nombre del tema que se utilizará como cola de mensajes no entregados. Si especifica un valor, el conector escribirá los mensajes no válidos en el tema de la cola de mensajes no entregados como cadenas JSON extendidas. Si deja esta configuración en blanco, el conector no escribirá mensajes no válidos en ningún tema. IMPORTANTE: `errors.tolerance` `mongo.errors.tolerance` Debe establecer `"all"` la configuración o en para habilitar esta propiedad. Valor predeterminado: `""` Valores aceptados: Un nombre de tema de Kafka válido
offset.partition.name	Tipo: cadena Descripción: El nombre de partición de desplazamiento personalizado que se utilizará. Puede usar esta opción para indicarle al conector que inicie un nuevo flujo de cambios cuando un desplazamiento existente contenga un token de reanudación no válido. Si deja esta configuración en blanco, el conector utiliza el nombre de partición predeterminado según los detalles de la conexión. Para ver una estrategia para nombrar particiones de desplazamiento, consulte Restablecer desplazamientos almacenados. Predeterminado: `""` Valores aceptados: Una cadena. Para obtener más información sobre cómo nombrar una partición, consulte SourceRecord en la documentación de la API de Apache Kafka.
heartbeat.interval.ms	Tipo: largo Descripción: El número de milisegundos que el conector espera entre el envío de mensajes de latido. El conector envía mensajes de latido cuando los registros de origen no se publican en el intervalo especificado. Este mecanismo mejora la capacidad de reanudación del conector para espacios de nombres de bajo volumen. Los mensajes de latido contienen un `postBatchResumeToken` campo de datos. El valor de este campo contiene la entrada del oplog del servidor MongoDB que el conector leyó por última vez del flujo de cambios. Establezca este valor en `0` para deshabilitar los mensajes de latido. Para obtener más información, consulte Prevención en la página Token dereanudación no válido.`0` Valor predeterminado: Valores aceptados: Un número entero
heartbeat.tema.name	Tipo: cadena Descripción: El nombre del tema en el que el conector debe publicar mensajes de latido. Debe proporcionar un valor positivo en la `heartbeat.interval.ms` configuración para habilitar esta función. Predeterminado: `__mongodb_heartbeats` Valores aceptados: Un nombre de tema de Kafka válido

Latidos cardíacos con transformaciones de mensaje único

Si habilita los latidos y especifica las Transformaciones de mensaje único (SMT) en su implementación de Kafka Connect, debe excluir sus mensajes de latido de sus SMT. Las SMT son una funcionalidad de Kafka Connect que le permite especificar transformaciones en los mensajes que pasan a través de su conector de origen sin necesidad de implementar una aplicación de Stream Processing.

Para excluir los mensajes de latidos de tus SMT, debes crear y aplicar un predicado a tus SMT. Los predicados son una funcionalidad de los SMT que te permite comprobar si un mensaje coincide con una instrucción condicional antes de aplicar una transformación.

La siguiente configuración define el predicado IsHeartbeat que coincide con los mensajes de latido enviados al tema de latido por defecto:

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

La siguiente configuración utiliza el predicado anterior para excluir los mensajes de latido de una ExtractField transformación:

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

Si no excluyes tus mensajes de latido de la transformación anterior, tu conector generará el siguiente error una vez que procese un mensaje de latido:

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 aprender más sobre los SMT, consulta Cómo usar las transformaciones de un solo mensaje (SMT) en Kafka Connect de Confluent.

Para obtener más información sobre predicados, consulta Filtro (Apache Kafka) de Confluent.

Para obtener más información sobre la transformación ExtractField, consulta ExtractField de Confluent.

Para obtener más información sobre el esquema de claves por defecto, consulte la página Esquemas por defecto.

Volver

Startup

Todas las propiedades