Overview
Aprende a recuperarte de un token de reanudación inválido en un conector de origen MongoDB Kafka.
Información de seguimiento de pila
El siguiente seguimiento de pila indica que el conector de origen tiene un token de reanudación no válido:
... org.apache.kafka.connect.errors.ConnectException: ResumeToken not found. Cannot create a change stream cursor ... Command failed with error 286 (ChangeStreamHistoryLost): 'PlanExecutor error during aggregation :: caused by :: Resume of change stream was not possible, as the resume point may no longer be in the oplog ...
Causa
Cuando el ID del token de reanudación de tu conector de origen no corresponde a ninguna entrada en tu implementación de MongoDB oplog, tu conector no tiene forma de determinar dónde empezar a procesar tu flujo de cambios de MongoDB. Haz clic en las siguientes pestañas para ver los escenarios en los que puedes experimentar este problema:
En este escenario, pausa su conector de origen y completa el registro de operaciones de su implementación de MongoDB:
Inicia una implementación de Kafka con un conector de origen MongoDB Kafka.
Se produce un evento de cambio en tu espacio de nombres MongoDB fuente y tu conector almacena un token de reanudación correspondiente a este evento.
Pausa tu conector de origen.
Mientras su conector está en pausa, usted llena su registro de operaciones de MongoDB de modo que MongoDB elimine la entrada del registro de operaciones correspondiente a su token de reanudación.
Reinicias tu conector de origen y no puede reanudar el procesamiento ya que tu token de reanudación no existe en tu MongoDB oplog.
En este escenario, tu conector de origen escucha cambios en un espacio de nombres MongoDB que se actualiza poco y la funcionalidad de latido no está activada:
Inicia una implementación de Kafka con un conector de origen MongoDB Kafka.
Se produce un evento de cambio en tu espacio de nombres MongoDB fuente y tu conector almacena un token de reanudación correspondiente a este evento.
Su espacio de nombres origen de MongoDB no se actualiza en el tiempo que tarda su implementación de MongoDB en rotar el evento de cambio correspondiente a su token de reanudación fuera de su oplog.
Produce un evento de cambio en su espacio de nombres MongoDB de origen y su conector de origen no puede reanudar el procesamiento porque su token de reanudación no existe en su registro de operaciones MongoDB.
Para más información sobre el Oplog, consulta el Manual de MongoDB.
Para obtener más información sobre los flujos de cambio, consulte Change Streams guide.
Soluciones
Puede recuperarse de un token de currículum no válido mediante una de las siguientes estrategias:
Tolerar Temporalmente los Errores
Puedes configurar tu conector de origen para tolerar errores mientras produces un evento de flujo de cambios que actualiza el token de reanudación del conector. Esta estrategia de recuperación es la más sencilla, pero existe el riesgo de que el connector ignore brevemente los errores no relacionados con el token de reanudación inválido. Si no te sientes cómodo tolerando brevemente errores en tu implementación, puedes borrar los desplazamientos almacenados en su lugar.
Para configurar tu conector de origen para tolerar temporalmente errores:
Establezca el
errors.toleranceopción para tolerar todos los errores:errors.tolerance=all Inserta, actualiza o elimina un documento en la colección referenciada por tu conector de origen para generar un evento de flujo de cambios que actualice el token de reanudación de tu conector.
Una vez que produces un evento de flujo de cambios, configura la opción
errors.tolerancepara que ya no se toleren errores:errors.tolerance=none
Para obtener más información sobre la errors.tolerance opción, consulte la página Propiedades de manejo de errores y reanudación de interrupciones.
Restablecer compensaciones almacenadas
Puede restablecer sus datos de desplazamiento de Kafka Connect, que contienen su token de reanudación, para permitir que su conector reanude el procesamiento de su flujo de cambios.
Para restablecer tus datos de offset, cambia el valor de la propiedad de configuración offset.partition.name a un nombre de partición que no exista en tu implementación de Kafka. Puede configurar su propiedad offset.partition.name de la siguiente manera:
offset.partition.name=<a string>
Tip
Nombrar sus particiones de desplazamiento
Considera utilizar el siguiente patrón para nombrar tus particiones de offset:
offset.partition.name=<source connector name>.<monotonically increasing number>
Este patrón proporciona los siguientes beneficios:
Registra el número de veces que restableces tu conector.
Documentos al que pertenece una partición de desfase en el conector
Por ejemplo, si nombró su conector de origen "source-values" y está configurando la propiedad offset.partition.name por primera vez, configure su conector de la siguiente manera:
offset.partition.name=source-values.1
La próxima vez que restablezcas los datos de offset de tu conector, configura tu conector de la siguiente manera:
offset.partition.name=source-values.2
Para aprender más sobre la propiedad de configuración offset.partition.name, consulta la página Manejo de Errores y Reanudación de Propiedades tras Interrupciones.
Para aprender a nombrar tu conector, consulta el sitio oficial de Apache Kafka documentación.
Prevención
Para evitar errores de token de reanudación no válidos causados por un espacio de nombres poco actualizado, habilite los latidos. Estos latidos son una función del conector de origen que permite que este actualice su token de reanudación a intervalos regulares, así como cuando cambia el contenido del espacio de nombres de origen de MongoDB.
Especifique la siguiente opción en su configuración de conector de origen para habilitar los latidos:
heartbeat.interval.ms=<a positive integer>
Para aprender más sobre latidos del corazón, consulta la guía Gestión de errores y reanudación tras interrupción-Propiedades.