Overview
Aprenda cómo recuperarse de un token de reanudación no válido en un conector de origen MongoDB Kafka.
Seguimiento de la 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 su conector de origen no corresponde a ninguna entrada en su implementación de MongoDB oplog, su conector no puede determinar dónde empezar a procesar su flujo de cambios de MongoDB. Haga clic en las siguientes pestañas para ver los escenarios en los que puede 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.
Produce un evento de cambio en su espacio de nombres MongoDB de origen y su 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.
Reinicia su conector de origen y no puede reanudar el procesamiento porque su token de reanudación no existe en su registro de operaciones de MongoDB.
En este escenario, su conector de origen escucha cambios en un espacio de nombres MongoDB que se actualiza con poca frecuencia y la función de latido no está habilitada:
Inicia una implementación de Kafka con un conector de origen MongoDB Kafka.
Produce un evento de cambio en su espacio de nombres MongoDB de origen y su conector almacena un token de reanudación correspondiente a este evento.
Su espacio de nombres de origen MongoDB no se actualiza en el tiempo que le toma a su implementación MongoDB rotar el evento de cambio correspondiente a su token de reanudación fuera de su registro de operaciones.
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 obtener más información sobre el registro de operaciones, consulte el Manual de MongoDB.
Para obtener más información sobre los flujos de cambio, consulte Guíade flujos de cambio.
Soluciones
Puede recuperarse de un token de currículum no válido mediante una de las siguientes estrategias:
Tolerar errores temporalmente
Puede configurar su conector de origen para que tolere errores mientras genera un evento de flujo de cambios que actualice el token de reanudación del conector. Esta estrategia de recuperación es la más sencilla, pero existe el riesgo de que su conector ignore brevemente errores no relacionados con el token de reanudación no válido. Si no se siente cómodo tolerando errores brevemente en su implementación, puede eliminar las compensaciones almacenadas.
Para configurar su conector de origen para que tolere errores temporalmente:
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 produzca un evento de flujo de cambios, configure la opción
errors.tolerancepara que ya no tolere 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 los datos de desplazamiento, cambie el valor de la propiedad de configuración offset.partition.name a un nombre de partición que no exista en su implementación de Kafka. Puede configurar la propiedad offset.partition.name de la siguiente manera:
offset.partition.name=<a string>
Tip
Nombrar sus particiones de desplazamiento
Considere utilizar el siguiente patrón para nombrar sus particiones de desplazamiento:
offset.partition.name=<source connector name>.<monotonically increasing number>
Este patrón proporciona los siguientes beneficios:
Registra la cantidad de veces que reinicia su conector
Documentos a qué conector pertenece una partición desplazada
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 restablezca los datos de desplazamiento de su conector, configúrelo de la siguiente manera:
offset.partition.name=source-values.2
Para obtener más información sobre la offset.partition.name propiedad de configuración, consulte la página Propiedades de manejo de errores y reanudación de interrupciones.
Para obtener información sobre cómo nombrar su conector, consulte el sitio web 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 la configuración de su conector de origen para habilitar los latidos:
heartbeat.interval.ms=<a positive integer>
Para obtener más información sobre los latidos, consulte la guía Manejo de errores y reanudación desde propiedades de interrupción.