Как сделать перезапуск в состоянии продюсером? - PullRequest
0 голосов
/ 06 февраля 2019

Последняя версия kafka поддерживает семантику "точно один раз" (EoS).Чтобы поддержать это понятие, дополнительные детали добавляются к каждому сообщению.Это означает, что у вашего потребителя;если вы печатаете смещения сообщений, они не обязательно будут последовательными.Это усложняет опрос темы, чтобы прочитать последнее зафиксированное сообщение.

В моем случае потребитель напечатал что-то вроде этого

Offset-0 0
Offset-2 1
Offset-4 2

Проблема: чтобы написать перезагружаемую гордость;Я опрашиваю тему и читаю содержание последнего сообщения.В этом случае;последнее сообщение будет смещено # 5, которое не является действительной записью потребителя.Следовательно, я вижу ошибки в своем коде.

Я могу использовать решение, предоставленное по адресу: Получение последнего сообщения, отправленного в тему кафки .Единственная проблема заключается в том, что вместо использования consumer.seek (partition, last_offset = 1);Я хотел бы использовать consumer.seek (partition, last_offset-2).Это может немедленно решить мою проблему, но это не идеальное решение.

Что было бы наиболее надежным и лучшим решением для получения последнего подтвержденного сообщения для потребителя, написанного на Java? ИЛИ

Можно ли использовать локальное хранилище состояний для раздела? ИЛИ

Какой самый рекомендуемый способ сохранить последнее сообщение длявыдерживать отказ производителя? ИЛИ

Перезапускаются ли разъемы kafka?Есть ли какой-то конкретный API, который я могу использовать для перезапуска производителей?

К вашему сведению - я не ищу быстрых исправлений

1 Ответ

0 голосов
/ 21 марта 2019

В моем случае несколько производителей помещают данные в одну большую тему.Поэтому чтение всей темы было бы кошмаром.

Решение, которое я нашел, состоит в том, чтобы сохранить другую тему , например, "P1_Track", где производитель может хранить метаданные.В рамках транзакции производитель отправит данные в одну большую тему и P1_Track.

Когда я перезапускаю производителя, он прочитает P1_Track и выяснит, с чего начать.

Думает о сохранении последнего подтвержденного сообщения в базе данных и его использовании при перезапуске процесса производителя.

...