{"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\/pt\/blog\/administrirovanie\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","title":{"rendered":"Reprocessamento de eventos recebidos do Kafka","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"Reprocessamento<\/p>\n

<\/p>\n

Ol\u00e1, Habr.<\/p>\n

<\/p>\n

Recentemente, eu compartilhei minha experi\u00eancia<\/a><\/noindex> sobre quais par\u00e2metros usamos com mais frequ\u00eancia em nossa equipe para Kafka Producer e Consumer, a fim de nos aproximarmos da entrega garantida. Neste artigo, quero explicar como organizamos a reprocessamento de eventos recebidos do Kafka em resultado da indisponibilidade tempor\u00e1ria de um sistema externo.<\/p>\n

<\/p>\n

Aplicativos modernos operam em um ambiente muito complexo. A l\u00f3gica de neg\u00f3cios, envolta em uma pilha tecnol\u00f3gica moderna, funcionando em uma imagem Docker, que \u00e9 gerenciada por um orquestrador como Kubernetes ou OpenShift, e se comunicando com outros aplicativos ou solu\u00e7\u00f5es empresariais atrav\u00e9s de uma cadeia de roteadores f\u00edsicos e virtuais. Em tal ambiente, sempre pode haver algo que quebre, portanto o reprocessamento de eventos em caso de indisponibilidade de um dos sistemas externos \u00e9 uma parte importante dos nossos processos de neg\u00f3cios.<\/p>\n

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

Como era antes do Kafka<\/h2>\n

<\/p>\n

Anteriormente, no projeto, utiliz\u00e1vamos IBM MQ para entrega ass\u00edncrona de mensagens. Quando ocorria algum erro no processo de funcionamento do servi\u00e7o, a mensagem recebida poderia ser colocada em uma dead-letter-queue (DLQ) para posterior an\u00e1lise manual. A DLQ era criada ao lado da fila de entrada, e a realoca\u00e7\u00e3o da mensagem ocorria dentro do IBM MQ. <\/p>\n

<\/p>\n

Se o erro era tempor\u00e1rio e consegu\u00edamos identific\u00e1-lo (por exemplo, ResourceAccessException durante uma chamada HTTP ou MongoTimeoutException durante uma consulta no MongoDb), ent\u00e3o entrava em vigor a estrat\u00e9gia de tentativas. Independentemente da ramifica\u00e7\u00e3o da l\u00f3gica do aplicativo, a mensagem original era realocada para uma fila sist\u00eamica para envio adiado ou para um aplicativo separado que havia sido criado h\u00e1 muito tempo para reenviar mensagens. Nesse caso, o n\u00famero da tentativa de reenvio era gravado no cabe\u00e7alho da mensagem, vinculado ao intervalo de atraso ou ao fim da estrat\u00e9gia a n\u00edvel de aplicativo. Se cheg\u00e1ssemos ao fim da estrat\u00e9gia, mas o sistema externo ainda estivesse indispon\u00edvel, a mensagem seria colocada na DLQ para an\u00e1lise manual.<\/p>\n

<\/p>\n

Busca pela solu\u00e7\u00e3o<\/h2>\n

<\/p>\n

Pesquisando na internet<\/a><\/noindex>, pode-se encontrar o seguinte uma solu\u00e7\u00e3o<\/a><\/noindex>. Em resumo, prop\u00f5e-se criar um t\u00f3pico para cada intervalo de atraso e implementar, do lado do aplicativo, Consumers que ler\u00e3o mensagens com o atraso necess\u00e1rio. <\/p>\n

<\/p>\n

\"Reprocessamento<\/p>\n

<\/p>\n

Apesar de haver muitos coment\u00e1rios positivos, n\u00e3o me parece totalmente satisfat\u00f3rio. Principalmente porque o desenvolvedor, al\u00e9m de implementar os requisitos de neg\u00f3cios, ter\u00e1 que investir muito tempo na implementa\u00e7\u00e3o do mecanismo descrito.<\/p>\n

<\/p>\n

Al\u00e9m disso, se o gerenciamento de acesso estiver habilitado no cluster Kafka, ser\u00e1 necess\u00e1rio dedicar algum tempo para criar t\u00f3picos e garantir os acessos necess\u00e1rios a eles. Al\u00e9m disso, ser\u00e1 preciso escolher o par\u00e2metro correto retention.ms para cada um dos t\u00f3picos de retry, para que as mensagens consigam ser reenviadas a tempo e n\u00e3o desapare\u00e7am. A implementa\u00e7\u00e3o e a solicita\u00e7\u00e3o de acessos ter\u00e3o que ser repetidas para cada servi\u00e7o existente ou novo.<\/p>\n

<\/p>\n

Vamos agora ver quais mecanismos de reprocessamento de mensagens o Spring nos oferece em geral e o spring-kafka em particular. O spring-kafka tem uma depend\u00eancia transitiva do spring-retry, que fornece abstra\u00e7\u00f5es para gerenciar diferentes BackOffPolicy. \u00c9 uma ferramenta bastante flex\u00edvel, mas sua desvantagem significativa \u00e9 o armazenamento de mensagens para reenvio na mem\u00f3ria da aplica\u00e7\u00e3o. Isso significa que a reinicializa\u00e7\u00e3o da aplica\u00e7\u00e3o devido a uma atualiza\u00e7\u00e3o ou a um erro durante a execu\u00e7\u00e3o resultar\u00e1 na perda de todas as mensagens que est\u00e3o aguardando reprocessamento. Como este ponto \u00e9 cr\u00edtico para o nosso sistema, n\u00e3o continuamos a explor\u00e1-lo.<\/p>\n

<\/p>\n

O pr\u00f3prio spring-kafka oferece v\u00e1rias implementa\u00e7\u00f5es de ContainerAwareErrorHandler, por exemplo, SeekToCurrentErrorHandler<\/a><\/noindex>, que permite que, sem deslocar o offset em caso de erro, a mensagem seja processada mais tarde. A partir da vers\u00e3o spring-kafka 2.3, surgiu a possibilidade de definir BackOffPolicy.<\/p>\n

<\/p>\n

Esta abordagem permite que as mensagens reprocessadas sobrevivam \u00e0 reinicializa\u00e7\u00e3o da aplica\u00e7\u00e3o, mas o mecanismo DLQ ainda est\u00e1 ausente. Foi essa op\u00e7\u00e3o que escolhemos no in\u00edcio de 2019, otimistamente acreditando que o DLQ n\u00e3o seria necess\u00e1rio (tivemos sorte e ele realmente n\u00e3o foi necess\u00e1rio durante v\u00e1rios meses de opera\u00e7\u00e3o da aplica\u00e7\u00e3o com esse sistema de reprocessamento). Erros tempor\u00e1rios resultaram na ativa\u00e7\u00e3o do SeekToCurrentErrorHandler. Os demais erros eram registrados no log, resultavam no deslocamento do offset, e o processamento continuava com a pr\u00f3xima mensagem.<\/p>\n

<\/p>\n

Solu\u00e7\u00e3o final<\/h2>\n

<\/p>\n

A implementa\u00e7\u00e3o baseada em SeekToCurrentErrorHandler nos levou a desenvolver nosso pr\u00f3prio mecanismo para reenvio de mensagens.<\/p>\n

<\/p>\n

Acima de tudo, quer\u00edamos aproveitar a experi\u00eancia existente e expandi-la de acordo com a l\u00f3gica do aplicativo. Para um aplicativo com l\u00f3gica linear, seria ideal interromper a leitura de novas mensagens por um curto per\u00edodo determinado na estrat\u00e9gia de tentativas. Para outros aplicativos, gostar\u00edamos de ter um \u00fanico ponto que garantisse a execu\u00e7\u00e3o da estrat\u00e9gia de tentativas. Al\u00e9m disso, esse \u00fanico ponto deve ter a funcionalidade de DLQ para ambas as abordagens.<\/p>\n

<\/p>\n

A pr\u00f3pria estrat\u00e9gia de tentativas deve ser armazenada no aplicativo que \u00e9 respons\u00e1vel por obter o pr\u00f3ximo intervalo em caso de um erro tempor\u00e1rio.<\/p>\n

<\/p>\n

Parar o Consumer para aplicativos com l\u00f3gica linear<\/h3>\n

<\/p>\n

Ao trabalhar com spring-kafka, o c\u00f3digo para parar o Consumer pode parecer assim:<\/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        \/\/ para DLQ\n    }<\/code><\/pre>\n
<\/p>\n
No exemplo, retryAt \u00e9 o tempo em que o MessageListenerContainer deve ser reiniciado, se ainda estiver em execu\u00e7\u00e3o. O rein\u00edcio ocorrer\u00e1 em uma thread separada, iniciada no TaskScheduler, cuja implementa\u00e7\u00e3o tamb\u00e9m \u00e9 fornecida pelo spring. <\/p>\n
<\/p>\n
O valor de retryAt \u00e9 encontrado da seguinte forma:<\/p>\n
<\/p>\n
    \n
  1. Buscamos o valor do contador de tentativas.<\/li>\n
  2. De acordo com o valor do contador, buscamos o intervalo de atraso atual na estrat\u00e9gia de tentativas. A estrat\u00e9gia \u00e9 declarada no pr\u00f3prio aplicativo, para seu armazenamento escolhemos o formato JSON.<\/li>\n
  3. O intervalo encontrado no array JSON cont\u00e9m o n\u00famero de segundos ap\u00f3s os quais a reprocessamento deve ser repetido. Esse n\u00famero de segundos \u00e9 adicionado ao tempo atual, formando o valor para retryAt.<\/li>\n
  4. Se o intervalo n\u00e3o for encontrado, o valor de retryAt ser\u00e1 null e a mensagem ser\u00e1 enviada para a DLQ para an\u00e1lise manual.<\/li>\n<\/ol>\n
    <\/p>\n
    Com essa abordagem, resta apenas manter a contagem de chamadas repetidas para cada mensagem que est\u00e1 atualmente em processamento, por exemplo, na mem\u00f3ria do aplicativo. Manter o contador de tentativas na mem\u00f3ria n\u00e3o \u00e9 cr\u00edtico para essa abordagem, uma vez que um aplicativo com l\u00f3gica linear n\u00e3o pode realizar o processamento como um todo. Ao contr\u00e1rio do spring-retry, a reinicializa\u00e7\u00e3o do aplicativo n\u00e3o resultar\u00e1 na perda de todas as mensagens para reprocessamento, mas simplesmente relan\u00e7ar\u00e1 a estrat\u00e9gia. <\/p>\n
    <\/p>\n
    Essa abordagem ajuda a aliviar a carga sobre o sistema externo, que pode estar indispon\u00edvel devido a uma carga muito grande. Em outras palavras, al\u00e9m do reprocessamento, conseguimos implementar o padr\u00e3o. circuit breaker<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    No nosso caso, o limite de erro \u00e9 de apenas 1, e para minimizar o tempo de inatividade do sistema devido a uma interrup\u00e7\u00e3o tempor\u00e1ria na rede, usamos uma estrat\u00e9gia de tentativas de chamadas muito granular com pequenos intervalos de atraso. Isso pode n\u00e3o ser adequado para todas as aplica\u00e7\u00f5es do grupo, portanto, a rela\u00e7\u00e3o entre o limite de erro e o tamanho do intervalo precisa ser ajustada com base nas caracter\u00edsticas do sistema.<\/p>\n
    <\/p>\n
    Um aplicativo separado para processar mensagens de aplicativos com l\u00f3gica n\u00e3o determin\u00edstica.<\/h3>\n
    <\/p>\n
    Aqui est\u00e1 um exemplo de c\u00f3digo que envia uma mensagem para tal aplicativo (Retryer), que reenvia para o t\u00f3pico DESTINATION ao atingir o tempo 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
    Do exemplo, pode-se ver que muitas informa\u00e7\u00f5es s\u00e3o transmitidas nos cabe\u00e7alhos. O valor RETRY_AT est\u00e1 presente da mesma forma que para o mecanismo de repeti\u00e7\u00e3o atrav\u00e9s da parada do Consumer. Al\u00e9m do DESTINATION e RETRY_AT, tamb\u00e9m transmitimos:<\/p>\n
    <\/p>\n