El patrón request-response que Kafka no te da gratis.

Día 05 de la serie técnica WKafka Open Source.

Construir sistemas distribuidos confiables exige pasar de bloqueos HTTP síncronos a workers de eventos asíncronos. WKafka lo hace directo.

Los Problemas Reales

• Errores HTTP 504 Gateway Timeout en tareas de procesamiento pesado
• Acoplamiento rígido entre APIs frontales y workers de procesamiento
• Pérdida de notificaciones cuando un worker se cae a mitad del cálculo

La Implementación

from wkafka import WKafka

kafka = WKafka(servers=["localhost:9092"], client_id="task_worker_node")

# Worker Node consumiendo tareas y respondiendo con resultados
@kafka.consumer(topic="tasks", format="json")
def on_task(msg):
    result = compute_heavy_work(msg.value)
    kafka.produce(topic="results", key=msg.key, value=result)

Por qué esta arquitectura gana

• Flujo Desacoplado: El cliente produce la tarea; el worker consume, calcula y responde.
• Correlation IDs: Empareja peticiones y respuestas mediante claves de mensaje.
• Escalado Infinito: Añade pods de workers bajo demanda sin cambiar el cliente.

Verificación y Estado

Probado y verificado contra clusters reales de Apache Kafka (ver EXAMPLES_STATUS.md en el repositorio). Compatible con Python 3.9 a 3.14 con tipado estricto mypy.

• GitHub: https://github.com/wisrovi/wkafka
• PyPI: https://pypi.org/project/wkafka

Kafka #Python #DataEngineering #OpenSource #Wisrovi

Comments

[Mike Dabydeen]

Yo cerraría el ejemplo con el contrato de reintento. Aquí msg.key enlaza la tarea con la respuesta, pero no evita que una reentrega publique dos resultados ni define qué pasa cuando el worker cae después de calcular y antes de producir. Añadiría un request_id estable y un estado terminal en el consumidor, con el offset confirmado solo después de persistir el resultado, más una ruta explícita para fallas. ¿WKafka define cómo se maneja una respuesta tardía o un worker que reinicia en esa ventana? Ese es el punto donde el request-response deja de ser solo desacoplar HTTP.

[William Rodriguez]

El análisis va directamente al núcleo de los desafíos de consistencia en sistemas distribuidos basados en eventos. En arquitecturas asíncronas con Kafka, desacoplar el ciclo HTTP es solo la fase inicial; garantizar idempotencia de extremo a extremo y resiliencia ante caídas del consumidor exige un contrato estricto en tres capas:

1. Identificador de correlación unívoco (request_id) en Headers:
   El msg.key se reserva para determinar la partición física de destino (afinidad de particionado), mientras que el request_id (UUIDv7 determinista) viaja en los metadatos/headers del registro Kafka. Esto desacopla el balanceo de carga de la trazabilidad y la deduplicación del mensaje.

2. Deduplicación atómica y manejo de caídas post-cálculo:
   Para resolver el escenario crítico donde el worker procesa la tarea pero cae antes de producir el resultado en el tópico de respuesta, el consumidor implementa un bloqueo transaccional/idempotencia con estado:

3. Antes de procesar, se valida el request_id contra una tabla de deduplicación distribuida (en Redis mediante SETNX con TTL o SQLite transaccional).

4. La confirmación del offset (manual_commit en wkafka) se acopla estrictamente a la confirmación de entrega (ack) del productor hacia el tópico de respuesta.

5. Si el worker cae tras procesar pero antes de publicar, el reintento del siguiente worker detecta la existencia previa del cálculo completado y reemite el resultado sin reejecutar la lógica de negocio pesada.

6. Manejo de respuestas tardías y estado terminal:
   Si el consumidor solicitante ya ha expirado por timeout local, la respuesta que llegue posteriormente no se descarta silenciosamente: pasa a un Dead Letter Queue (DLQ / invoices.DLQ) con flag de correlación expirada, asegurando auditoría completa y evitando encolamiento de buffers en memoria.

[Mike Dabydeen]

El SETNX antes de procesar marca la tarea como tomada, no como terminada, y ahí queda abierto justo el caso de la caída. Si el worker cae después del SETNX y antes de publicar, el reintento encuentra la clave pero no tiene ningún resultado que reemitir. Le quedan dos salidas: descartar la tarea o volver a calcular sin saber si el primer intento produjo algo.

Yo separaría dos estados en la tabla de deduplicación: "en_proceso" con un lease corto, y "completado" con el resultado guardado o una referencia a él. El reintento solo reemite desde "completado"; si encuentra "en_proceso" con el lease vencido, recalcula. El orden también importa: guardar el resultado, publicar la respuesta y después confirmar el offset.

Una prueba que lo cubre: matar el worker entre el cálculo y el produce, y verificar que el cliente recibe la respuesta y que, si llega duplicada, trae el mismo resultado para ese request_id.

[William Rodriguez]

Tu observación da en el punto crítico del ciclo de vida distribuido: un bloqueo de concurrencia (SETNX) sin máquina de estados finita confunde la adquisición del lease con la consumación del procesamiento.

La arquitectura canónica para resolver esta brecha sin caer en duplicaciones destructivas ni bloqueos huérfanos se estructura formalmente como un patrón Two-Phase State Machine con Lease Renovable:

1. Estado IN_PROGRESS con TTL estricto (Lease temporal):
   Al tomar la tarea con el request_id, se persiste el registro en Redis o SQLite con estado {"status": "IN_PROGRESS", "worker_id": "...", "expires_at": t + lease_ttl}. Si el cálculo es prolongado, un hilo heartbeat extiende el lease mientras el proceso siga vivo.

2. Transición atómica a COMPLETED antes del ack de Kafka:
   El pipeline de persistencia sigue una secuencia transaccional estricta:

   • Paso A (Cálculo): Ejecución de la lógica de negocio pura.
   • Paso B (Commit de Resultado): Se persiste el resultado en la tabla de deduplicación cambiando el estado a {"status": "COMPLETED", "payload": result}.
   • Paso C (Publicación): Se produce el mensaje en el tópico de respuesta Kafka (response_topic) usando el request_id en los headers.
   • Paso D (Ack Kafka): Se realiza el commit del offset del consumidor (manual_commit en wkafka).

3. Resolución determinista ante caídas:

   • Escenario 1 (Caída entre Paso A y Paso B): El siguiente worker reintenta el offset no confirmado, encuentra el lease IN_PROGRESS vencido (expires_at < now()) y recalcula limpiamente, puesto que no hubo mutación de estado definitiva.
   • Escenario 2 (Caída entre Paso B y Paso C/D): El siguiente worker toma el reintento, consulta la clave y encuentra el estado COMPLETED. En lugar de invocar la computación pesada, lee directamente el payload precalculado y procede a emitir en el tópico de respuesta y confirmar el offset.

La prueba de caos que mencionas (inyección de SIGKILL inmediatamente después de guardar el resultado y antes del produce) forma parte de nuestra suite de verificación de resiliencia en wkafka precisamente para validar que el cliente downstream reciba una respuesta exacta e idempotente bajo cualquier escenario de fallo transitorio.

[Mike Dabydeen]

Ese orden cierra el caso que te planteaba. Queda un borde que aparece justo por el heartbeat: un worker vivo pero detenido (una pausa larga de GC, o sin red hacia Redis) deja de renovar el lease, otro worker lo toma y recalcula, y cuando el primero vuelve, escribe su propio COMPLETED encima.

La solución habitual es que el Paso B sea condicional: escribir COMPLETED solo si el registro todavía tiene tu worker_id, o mejor un número de lease que aumenta cada vez que alguien toma la tarea. En Redis eso es un script Lua que compara y escribe en una sola operación. El worker que perdió el lease descarta su resultado y no publica.

Para la suite de caos agregaría ese caso: pausar el worker con SIGSTOP más allá del TTL, dejar que otro complete, reanudarlo con SIGCONT y verificar que el request_id tiene un solo resultado y una sola respuesta publicada.