{"id":56186,"date":"2020-02-06T00:00:00","date_gmt":"2020-02-05T21:00:00","guid":{"rendered":"https:\/\/prohoster.info\/blog\/blog_prohoster\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka"},"modified":"2020-02-18T14:04:24","modified_gmt":"2020-02-18T11:04:24","slug":"povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","status":"publish","type":"post","link":"https:\/\/prohoster.info\/es\/blog\/administrirovanie\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","title":{"rendered":"Reprocesamiento de eventos recibidos de Kafka","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"Reprocesamiento<\/p>\n

<\/p>\n

Hola, Habr.<\/p>\n

<\/p>\n

Recientemente compart\u00ed mi experiencia<\/a><\/noindex> sobre los par\u00e1metros que utilizamos con m\u00e1s frecuencia en nuestro equipo para Kafka Producer y Consumer, con el fin de acercarnos a la entrega garantizada. En este art\u00edculo quiero contar c\u00f3mo organizamos el reprocesamiento de eventos recibidos de Kafka debido a la indisponibilidad temporal de un sistema externo.<\/p>\n

<\/p>\n

Las aplicaciones modernas operan en un entorno muy complejo. La l\u00f3gica de negocio, envuelta en una pila tecnol\u00f3gica moderna, que funciona en una imagen de Docker gestionada por un orquestador como Kubernetes u OpenShift, y que se comunica con otras aplicaciones o soluciones empresariales a trav\u00e9s de una cadena de enrutadores f\u00edsicos y virtuales. En un entorno as\u00ed, siempre puede fallar algo, por lo que el reprocesamiento de eventos en caso de la indisponibilidad de uno de los sistemas externos es una parte importante de nuestros procesos de negocio.<\/p>\n

<\/a><\/noindex><\/p>\n

C\u00f3mo era antes de Kafka<\/h2>\n

<\/p>\n

Anteriormente en el proyecto utiliz\u00e1bamos IBM MQ para la entrega as\u00edncrona de mensajes. Cuando se produc\u00eda alg\u00fan error durante la operaci\u00f3n del servicio, el mensaje recibido pod\u00eda ser colocado en una cola de mensajes muertos (DLQ) para su posterior revisi\u00f3n manual. La DLQ se creaba junto a la cola de entrada, y el traslado del mensaje ocurr\u00eda dentro de IBM MQ. <\/p>\n

<\/p>\n

Si el error era temporal y pod\u00edamos determinarlo (por ejemplo, ResourceAccessException al hacer una llamada HTTP o MongoTimeoutException al hacer una consulta en MongoDb), se activaba la estrategia de reintentos. Independientemente de la ramificaci\u00f3n de la l\u00f3gica de la aplicaci\u00f3n, el mensaje original se trasladaba ya sea a una cola del sistema para env\u00edo diferido o a una aplicaci\u00f3n separada que se hab\u00eda creado hace tiempo para reenv\u00edo de mensajes. En este caso, se registraba en el encabezado del mensaje el n\u00famero de reenv\u00edo, que estaba vinculado al intervalo de retardo o al final de la estrategia a nivel de la aplicaci\u00f3n. Si lleg\u00e1bamos al final de la estrategia, pero el sistema externo segu\u00eda sin estar disponible, el mensaje se colocar\u00eda en la DLQ para revisi\u00f3n manual.<\/p>\n

<\/p>\n

B\u00fasqueda de soluci\u00f3n<\/h2>\n

<\/p>\n

Al buscar en Internet<\/a><\/noindex>, se puede encontrar lo siguiente soluci\u00f3n<\/a><\/noindex>. En resumen, se sugiere establecer un tema para cada intervalo de retardo y realizar en la parte de la aplicaci\u00f3n Consumers que leer\u00e1n los mensajes con el retraso necesario. <\/p>\n

<\/p>\n

\"Reprocesamiento<\/p>\n

<\/p>\n

A pesar de la gran cantidad de comentarios positivos, me parece que no es del todo adecuado. En primer lugar, porque al desarrollador, adem\u00e1s de implementar los requisitos comerciales, le llevar\u00e1 mucho tiempo implementar el mecanismo descrito.<\/p>\n

<\/p>\n

Adem\u00e1s, si se habilita el control de acceso en el cl\u00faster de Kafka, se necesitar\u00e1 algo de tiempo para crear los temas y garantizar los accesos necesarios a ellos. Adem\u00e1s de esto, ser\u00e1 necesario elegir el par\u00e1metro correcto retention.ms para cada uno de los temas de reintento, de modo que los mensajes puedan ser reenviados a tiempo y no se pierdan. La implementaci\u00f3n y la solicitud de accesos deber\u00e1n repetirse para cada servicio existente o nuevo.<\/p>\n

<\/p>\n

Ahora veamos qu\u00e9 mecanismos para el re-procesamiento de mensajes nos ofrece Spring en general y Spring-Kafka en particular. Spring-Kafka tiene una dependencia transitiva en Spring-Retry, que proporciona abstracciones para gestionar diferentes BackOffPolicy. Es una herramienta bastante flexible, pero su desventaja significativa es que almacena los mensajes para reenv\u00edo en la memoria de la aplicaci\u00f3n. Esto significa que el reinicio de la aplicaci\u00f3n debido a una actualizaci\u00f3n o un error durante la explotaci\u00f3n resultar\u00e1 en la p\u00e9rdida de todos los mensajes que esperan ser re-procesados. Dado que este aspecto es cr\u00edtico para nuestro sistema, no lo consideramos m\u00e1s adelante.<\/p>\n

<\/p>\n

Spring-Kafka ofrece varias implementaciones de ContainerAwareErrorHandler, como por ejemplo SeekToCurrentErrorHandler<\/a><\/noindex>, que permite procesar el mensaje m\u00e1s tarde sin mover el offset en caso de un error. A partir de la versi\u00f3n 2.3 de Spring-Kafka, se introdujo la posibilidad de establecer BackOffPolicy.<\/p>\n

<\/p>\n

Este enfoque permite que los mensajes re-procesados sobrevivan al reinicio de la aplicaci\u00f3n, pero el mecanismo DLQ sigue siendo ausente. Precisamente esta opci\u00f3n elegimos a principios de 2019, optimistamente pensando que no necesitar\u00edamos DLQ (tuvimos suerte y realmente no lo necesit\u00e1bamos durante varios meses de explotaci\u00f3n de la aplicaci\u00f3n con este sistema de re-procesamiento). Los errores temporales provocaban la activaci\u00f3n de SeekToCurrentErrorHandler. Los dem\u00e1s errores se imprim\u00edan en el registro, llevaban al desplazamiento del offset y el procesamiento continuaba con el siguiente mensaje.<\/p>\n

<\/p>\n

Soluci\u00f3n final<\/h2>\n

<\/p>\n

La implementaci\u00f3n basada en SeekToCurrentErrorHandler nos llev\u00f3 a desarrollar nuestro propio mecanismo para el reenv\u00edo de mensajes.<\/p>\n

<\/p>\n

Primero que nada, quer\u00edamos aprovechar la experiencia existente y ampliarla seg\u00fan la l\u00f3gica de la aplicaci\u00f3n. Para una aplicaci\u00f3n con una l\u00f3gica lineal, lo \u00f3ptimo ser\u00eda detener la lectura de nuevos mensajes durante un breve intervalo de tiempo, establecido dentro de la estrategia de reintentos. Para otras aplicaciones, nos gustar\u00eda tener un punto \u00fanico que garantizara la ejecuci\u00f3n de la estrategia de reintentos. Adem\u00e1s, este punto \u00fanico deber\u00eda contar con funcionalidad DLQ para ambos enfoques.<\/p>\n

<\/p>\n

La estrategia de reintentos en s\u00ed debe almacenarse en la aplicaci\u00f3n responsable de obtener el siguiente intervalo en caso de un error temporal.<\/p>\n

<\/p>\n

Detener el Consumer para una aplicaci\u00f3n con l\u00f3gica lineal<\/h3>\n

<\/p>\n

Al trabajar con spring-kafka, el c\u00f3digo para detener el Consumer podr\u00eda verse aproximadamente as\u00ed:<\/p>\n

<\/p>\n

public void pauseListenerContainer(MessageListenerContainer listenerContainer, \n                                   Instant retryAt) {\n        if (nonNull(retryAt) && listenerContainer.isRunning()) {\n            listenerContainer.stop();\n            taskScheduler.schedule(() -> listenerContainer.start(), retryAt);\n            return;\n        }\n        \/\/ a DLQ\n    }<\/code><\/pre>\n
<\/p>\n
En el ejemplo, retryAt es el momento en que se debe reiniciar el MessageListenerContainer, si todav\u00eda est\u00e1 funcionando. El reinicio se llevar\u00e1 a cabo en un hilo separado que se ejecuta en TaskScheduler, cuya implementaci\u00f3n tambi\u00e9n proporciona spring. <\/p>\n
<\/p>\n
El valor de retryAt lo encontramos de la siguiente manera:<\/p>\n
<\/p>\n
    \n
  1. Se busca el valor del contador de reintentos.<\/li>\n
  2. De acuerdo con el valor del contador, se busca el intervalo actual de retraso en la estrategia de reintentos. La estrategia se declara en la propia aplicaci\u00f3n, para su almacenamiento elegimos el formato JSON.<\/li>\n
  3. El intervalo encontrado en el array JSON contiene la cantidad de segundos despu\u00e9s de los cuales se deber\u00e1 repetir el procesamiento. Esta cantidad de segundos se suma al tiempo actual, formando el valor para retryAt.<\/li>\n
  4. Si no se encuentra el intervalo, el valor de retryAt es null y el mensaje se enviar\u00e1 a la DLQ para su an\u00e1lisis manual.<\/li>\n<\/ol>\n
    <\/p>\n
    Con este enfoque, solo queda guardar la cantidad de intentos para cada mensaje que actualmente est\u00e1 en proceso, por ejemplo, en la memoria de la aplicaci\u00f3n. Mantener un contador de intentos en la memoria no es cr\u00edtico para este enfoque, ya que una aplicaci\u00f3n con l\u00f3gica lineal no puede procesar en su totalidad. A diferencia de spring-retry, el reinicio de la aplicaci\u00f3n no resultar\u00e1 en la p\u00e9rdida de todos los mensajes para re-procesamiento, sino simplemente en un reinicio de la estrategia. <\/p>\n
    <\/p>\n
    Este enfoque ayuda a aliviar la carga de un sistema externo que puede estar inalcanzable debido a una carga excesiva. En otras palabras, adem\u00e1s de la re-procesamiento, hemos logrado implementar el patr\u00f3n circuit breaker<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    En nuestro caso, el umbral de error es de solo 1, y para minimizar el tiempo de inactividad del sistema debido a una interrupci\u00f3n temporal de la red, utilizamos una estrategia de reintentos muy granular con peque\u00f1os intervalos de retraso. Esto puede no ser adecuado para todas las aplicaciones del grupo empresarial, por lo que la relaci\u00f3n entre el umbral de error y el tama\u00f1o del intervalo debe ajustarse seg\u00fan las caracter\u00edsticas del sistema.<\/p>\n
    <\/p>\n
    Una aplicaci\u00f3n separada para procesar mensajes de aplicaciones con l\u00f3gica no determin\u00edstica<\/h3>\n
    <\/p>\n
    Aqu\u00ed hay un ejemplo de c\u00f3digo que env\u00eda un mensaje a tal aplicaci\u00f3n (Retryer), que volver\u00e1 a enviar al tema DESTINATION al alcanzar el tiempo RETRY_AT:<\/p>\n
    <\/p>\n
    \npublic  void retry(ConsumerRecord record, String retryToTopic, \n                         Instant retryAt, String counter, String groupId, Exception e) {\n        Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());\n        List
     arrayOfHeaders = \n            new ArrayList(Arrays.asList(headers.toArray()));\n        updateHeader(arrayOfHeaders, GROUP_ID, groupId::getBytes);\n        updateHeader(arrayOfHeaders, DESTINATION, retryToTopic::getBytes);\n        updateHeader(arrayOfHeaders, ORIGINAL_PARTITION, \n                     () -> Integer.toString(record.partition()).getBytes());\n        if (nonNull(retryAt)) {\n            updateHeader(arrayOfHeaders, COUNTER, counter::getBytes);\n            updateHeader(arrayOfHeaders, SEND_TO, \"retry\"::getBytes);\n            updateHeader(arrayOfHeaders, RETRY_AT, retryAt.toString()::getBytes);\n        } else {\n            updateHeader(arrayOfHeaders, REASON, \n                         ExceptionUtils.getStackTrace(e)::getBytes);\n            updateHeader(arrayOfHeaders, SEND_TO, \"backout\"::getBytes);\n        }\n        ProducerRecord messageToSend =\n            new ProducerRecord(retryTopic, null, null, record.key(), record.value(), arrayOfHeaders);\n        kafkaTemplate.send(messageToSend);\n    }<\/code><\/pre>\n
    <\/p>\n
    Del ejemplo se puede ver que se transmite mucha informaci\u00f3n en los encabezados. El valor RETRY_AT se encuentra tambi\u00e9n, al igual que para el mecanismo de reintento a trav\u00e9s de la detenci\u00f3n del Consumer.<\/p>\n
    <\/p>\n