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: string Descripción: Indica si se debe continuar con el procesamiento de los mensajes cuando el conector encuentra un error. Establezca esto en `"none"` si desea que el conector deje de procesamiento mensajes e informe del problema si encuentra un error. Establezca esto en `"all"` si desea que el conector siga en procesamiento mensajes e ignore los errores que encuentre. IMPORTANTE: Esta propiedad anula la errors.tolerance propiedad de Connect Framework. Por defecto: `"none"` Valores aceptados: `"none"` o `"all"`
mongo.errors.log.enable	Tipo: booleano descripción: Indica si el conector debe informar errores en la entrada de registro. Establezca esto en `true` para registrar todos los errores que el conector encuentre. Establezca esto en `false` para registro los errores que el conector no tolera. Puede especificar qué errores debe tolerar el conector utilizando la configuración `errors.tolerance` o `mongo.errors.tolerance`. IMPORTANTE: Esta propiedad anula la errors.log.enable propiedad de Connect Framework. Por defecto: `false` Valores aceptados: `true` o `false`
mongo.errors.deadletterqueue.tema.name	Tipo: string descripción: El nombre del tema que se utilizará como fila de letra muerta. Si especifica un valor, el conector guarda mensajes no válidos en el tema de la fila de letra muerta como cadenas JSON extendidas. Si deja esta configuración en blanco, el conector no guarda mensajes inválidos en ningún tema. IMPORTANTE: Debe configurar la configuración `errors.tolerance` o `mongo.errors.tolerance` en `"all"` para habilitar esta propiedad. Por defecto: `""` Valores aceptados: Un nombre de tema de Kafka válido
offset.partition.name	Tipo: string descripción: El nombre de la partición de desplazamiento personalizado que se va a utilizar. Puede utilizar esta opción para indicar 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 por defecto basado en los detalles de la conexión. Para ver una estrategia para nombrar particiones de desplazamiento, consulte Restablecer desplazamientos almacenados. Por defecto: `""` Valores aceptados: Una string. Para aprender más sobre cómo nombrar una partición, consulte SourceRecord en la documentación de la API de Apache Kafka.
heartbeat.interval.ms	Tipo: long 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 reanudabilidad del conector para namespace de bajo volumen. Los mensajes de latido contienen un campo de datos `postBatchResumeToken`. El valor de este campo contiene la entrada oplog del servidor MongoDB que el conector leyó por última vez del flujo de cambios. Establezca esto en `0` para deshabilitar los mensajes de latido. Para aprender más, consulte Prevención en la página Token de reanudación no válido. por defecto: `0` Valores aceptados: Un número entero
heartbeat.tema.name	Tipo: string descripción: El nombre del tema en el que el conector debe publicar mensajes de latido. Debe proporcionar un valor positivo en la configuración `heartbeat.interval.ms` para habilitar esta funcionalidad. Por defecto: `__mongodb_heartbeats` Valores admitidos: 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