Una Lambda que envía registros de DynamoDB Streams a Kafka empezó a fallar en producción.
No habíamos desplegado cambios en la aplicación ni en la infraestructura. Algunas invocaciones seguían completándose correctamente, pero había suficientes fallos como para activar una alerta:
KafkaTimeoutError: Failed to update metadata after 60.0 secs.Antes de enviar un registro, el productor de Kafka necesita información sobre el topic y sus particiones para saber a dónde enviarlo. La llamada estaba fallando mientras esperaba esa información.
El siguiente paso fue entender el alcance del problema.
Varias invocaciones agotaban el tiempo de espera al conectarse al mismo broker, mientras otras seguían escribiendo correctamente. Comparar las invocaciones fallidas con las exitosas ayudó a acotar la investigación. El problema afectaba a una parte del tráfico; Kafka no estaba completamente caído.
Relacionamos los timeouts con una operación de mantenimiento de Kafka. Durante ese proceso, el liderazgo de las particiones afectadas pasó a otro broker. Algunas invocaciones ya escribían correctamente después de la transición, mientras los productores afectados seguían reconectándose y actualizando su metadata.
El tiempo de recuperación también nos dio una pista. Los logs mostraban que los productores afectados se recuperaban apenas unos segundos después de registrar el timeout.
En conjunto, esas observaciones explicaban por qué esperar un poco más era una respuesta razonable. Habíamos encontrado una interrupción temporal durante el mantenimiento y podíamos ver que la recuperación estaba ocurriendo. En este incidente no había una regla de red ni una configuración de rutas que corregir.
La aplicación simplemente dejaba de esperar antes de que el productor terminara de recuperarse.
Dos alternativas
Consideramos dos opciones.
1. Fallar la invocación y dejar que Lambda reintente el lote
Como nuestro handler creaba un productor por invocación, el siguiente intento comenzaría con un productor nuevo.
La idea resultaba atractiva: fallar, empezar de nuevo y obtener un cliente con un estado nuevo.
Pero pensemos en un lote de 50 registros cuando se reintenta el lote completo. Si el registro 20 fallaba, los registros del 1 al 19 podían haberse escrito ya en Kafka. Volver a procesar el lote podía hacer que esos registros se escribieran otra vez.
Manejar entregas duplicadas requeriría cambios más amplios que ajustar la espera por metadata. Esa necesidad ya existía: los event source mappings de DynamoDB Streams procesan los registros al menos una vez. Esperar más no eliminaría la necesidad de manejar duplicados.
2. Mantener la invocación activa y volver a intentar la escritura
La otra propuesta era darle tiempo a Kafka para terminar la transición y después volver a llamar a send():
try:
producer.send(...)
except KafkaTimeoutError:
time.sleep(SLEEP_TIME)
producer.send(...)También parecía razonable, pero planteaba algunas preguntas prácticas.
¿Cuánto tiempo debíamos esperar? Al repetirse en muchas invocaciones, esa pausa sumaría tiempo de ejecución y costo. También dejaría menos tiempo para procesar el resto de cada lote.
Además, necesitábamos distinguir entre un fallo temporal por el que valía la pena esperar y un error que debía propagarse. La investigación ya había mostrado que el productor podía recuperarse. Antes de agregar otro ciclo de reintentos, necesitábamos entender qué estaba haciendo durante esos primeros 60 segundos.
¿Qué significa reintentar en este caso?
La Lambda crea un KafkaProducer de kafka-python por invocación y lo usa para todos los registros del lote. Ese productor administra las conexiones a Kafka en un hilo en segundo plano y mantiene la metadata necesaria para dirigir los registros al broker correcto.
Llamar a send() reutiliza el mismo productor. No crea un cliente nuevo ni fuerza una conexión nueva. La reconexión y la actualización de metadata siguen siendo responsabilidad del productor.
Antes de que send() pueda poner un registro en la cola de envío, el productor necesita metadata de su topic. Si esa información no está disponible, la llamada espera mientras el productor intenta obtenerla. Volver a llamar a send() esperaría a que avanzara ese mismo proceso de recuperación. En la práctica, la aplicación haría esto:
send(record)
↓
esperar hasta 60 segundos por metadata
↓
timeout
esperar X segundos
send(record)
↓
esperar hasta 60 segundos por metadataUna segunda llamada a send() podía funcionar porque el productor había tenido más tiempo para reconectarse u obtener metadata actualizada. Ambas llamadas dependerían del mismo productor para gestionar la recuperación. El beneficio principal era darle más tiempo al mismo proceso de recuperación. En lugar de capturar el timeout, hacer una pausa y volver a llamar a send(), podíamos ampliar la espera que ya existía:
KafkaProducer(
max_block_ms=120000,
# configuración existente
)Aumentamos max_block_ms de 60 a 120 segundos. Eso permitía esperar la metadata hasta 120 segundos, sin obligar a que cada llamada tardara todo ese tiempo. Si la metadata estaba disponible después de 63 segundos, la llamada podía continuar en ese momento.
Este parámetro limita cuánto puede bloquearse send() mientras espera metadata o espacio en el buffer; no es un timeout de confirmación de entrega. El handler todavía necesita comprobar los resultados de los envíos antes de reportar éxito. Además, la espera ampliada debe caber dentro del timeout de la invocación de Lambda y dejar tiempo para terminar el lote y confirmar los envíos pendientes.
Eso encajaba con lo que habíamos observado: los productores afectados se recuperaban poco después del límite original de 60 segundos.
La diferencia puede parecer pequeña, pero representaba mejor lo que la función necesitaba: darle al proceso de recuperación más tiempo para terminar.
Reintentar después de entender la recuperación
Reintentar era una respuesta razonable para este incidente. Kafka estaba en mantenimiento y los logs mostraban que los productores afectados podían recuperarse.
La decisión era dónde y cómo darles más tiempo.
Algunas preguntas concretas nos ayudaron a tomar esa decisión:
- ¿En qué punto se había detenido la operación?
- ¿Qué parte del sistema estaba afectada?
- ¿Qué recuperación ya estaba en marcha?
- ¿Qué cambiaría realmente en el siguiente intento?
La solución en sí era sencilla: esperar más.
Pero incluso una solución tan sencilla deja una pregunta importante sobre el diseño del sistema:
¿Cuál es la forma correcta de esperar?