Восстановление состояния согласованности в Флинка при использовании Кафки, как EventStore

Question

Проблемы

Я реализует microservice как событие-сорсинг агрегат, который, в свою очередь, реализуются как FLiNK FlatMapFunction. В базовой настройке агрегат считывает события и команды из двух тем кафки. Затем он записывает новые события в эту первую тему и обрабатывает результаты в третьей теме. Поэтому Кафка выступает в качестве магазина событий. Надеюсь, что этот рисунок помогает:

RPC Request        RPC Result 
    |             | 
    ~~~~> Commands-|    |---> Results ~~~~~~| 
       |-->Aggregate--| 
    ~> Input evs. -|    |---> output evs. ~~~ 
    |             | 
    ~~~~~<~~~~~~~~~~~<~~~feedbak loop~~~~~<~~~~~~~~<~~~ 

В связи с тем, что Кафка не checkpoined, команда потенциально может быть воспроизведена в два раза и, кажется, что выходные данные события также могут быть записаны в два раз темы.

Как можно восстановить состояние в тех случаях с повторными сообщениями? Возможно ли, чтобы агрегат знал, когда его исходные потоки обновлены, чтобы начать обработку команд?

Мои мысли

Я думал несколько решений:

1. Если Flink реализует Откат неподтвержденные события, мойкой можно было бы осуществить, который получит ток смещения от источника событий. При перезапуске этот приемник удаляет события более позднего, чем смещения, в теме kafka. Это его способ, KafkaSource и KafkaSink будут генерироваться из одного и того же строителя, а затем подвергаться топологии. Это решение имеет сильную проблему, учитывая, что другие службы могут читать новые события в этой теме и вызывать несогласованность.

2. Если удаление событий из Flink невозможно в 2, источник statefull может потенциально считывать события из смещения и пытаться сопоставить повторяющиеся события в совокупности и отбрасывать их. Эти параметры кажутся ненадежными, так как могут быть ситуации, когда исправления не являются детерминированными и подвержены ошибкам, поскольку их следует переосмыслить для каждого агрегата и топологии, и это не гарантировало бы восстановление (например, в случае последовательных перезапусков). Поэтому это плохое решение.

3. Другой подход - это один. Это создать специальный KafkaSource с двумя специальными водяными знаками: первый, KafkaSourceStartedWatermark, всегда будет отправлен при запуске источника для уведомления зависимых операторов. Когда этот водяной знак отправляется, источник внутренне записывает текущее смещение Кафки. Второй, KafkaSourceUpToDateWatermark, отправляется источником при достижении смещения. Эти водяные знаки будут прозрачно перемещаться по топологии. Оператор должен иметь возможность обрабатывать эти водяные знаки, реализуя специальный интерфейс WatermarkNotifiable. Затем агрегат сможет буферизовать или отбрасывать команды RPC, пока он не будет обновлен в каждом источнике входного сигнала.

   interface WatermarkNotifiable { 
       void started(String watermarkId);//KafkaSourceStartedWatermark watermark 
       void upToDate(String watermarkId);//KafkaSOurceUpToDateWatermark watermark 
   } 
   

4. Если внедрение инфраструктуры в 3 не представляется возможным, то KafkaSource может реализовать конструктор, задающий специальное мероприятие, водяной знак, который может поехать к операторам, но это потребовало бы, что все операторы зависят от этих водяных знаков УЭ - значит, тогда.

5. Другой подход заключается в том, чтобы не обрабатывать команды старше, чем критерии. Например, команды имеют временную метку записи. Если используется время, синхронизация времени имеет решающее значение.

Связанные StackOverflow расспрашивает

1. Using Kafka as a (CQRS) Eventstore. Good idea?
2. Kafka - Know if Consumer is up to date
3. Kafka & Flink duplicate messages on restart

Answer 1

Создание нового типа оператора Conmuter. Это похоже на источник. Он состоит из нескольких источников, представляющих темы событий и команд. Он начинается в состоянии «восстановления». В этом состоянии он читает из тем событий до последних. Между тем, для команд он сохраняет или опускает их. После обновления, он считает, что восстановлено и «открывает» способ команд. Он может быть реализован отдельно как источник плюс оператор.

FlinkKafkaProducerXX недостаточно для этого, но это было бы базой для его реализации.