В чем причина исключений kafka.common.OffsetOutOfRangeException в Kafka

Question

Я использую Kafka Mirrormaker для имитации рабочего трафика в кластер Elastic MapReduce, на котором работает Spark.Вот мои настройки:

1) Один экземпляр Kafka Mirrormaker использует из производственного экземпляра Kafka и заполняет тестовый экземпляр Kafka

2) Второй экземпляр Kafka Mirrormaker потребляет из тестового экземпляра Kafkaсоздан в 1) и публикуется в новом кластере Kafka

3) Кластер Elastic MapReduce потребляет из кластера Kafka в 3).

Проблема в том, что я вижу kafka.common.OffsetOutOfRangeExceptionошибки в моих лог-файлах.Мне было интересно, как предотвратить их возникновение.

Это процедура, которую я использую при выполнении задания EMR:

1) Убить моего работающего Mirrormaker

2) Отбросить всетемы в кластере

3) Создать все темы

4) Перезапустить Mirrormaker

5) Запустить работу EMR

Раньше я думал, что кафка.common.OffsetOutOfRangeException произошла, потому что я заменял хосты.Однако, даже после удаления / воссоздания тем, я все еще получаю сообщение об ошибке.

Есть ли особые настройки, которые мне нужно сделать на моем Mirrormaker или кластере Kafka, чтобы этого не происходило?

Answer 1

MirrorMaker выполняет репликацию только сообщений из одного кластера в другой, но не смещений.Внутри MirrorMaker использует потребителей и производителей.Это означает, что смещения сообщений в целевом кластере будут другими.Предположим, у вас есть тема в исходном кластере, и при сохранении удалите некоторые данные, поэтому самое старое смещение равно 1000, и вы создаете новую тему в целевом кластере.Сообщение 1000 из исходного кластера будет иметь смещение 0 в целевом кластере.

Вам нужно будет «перевести» смещения вручную.Например, вы можете получить начальные смещения до начала репликации с помощью MirrorMaker и вычесть число смещений в целевом кластере.