Доступ к хранилищу смещений из коннектора
У меня есть исходный коннектор, написанный на Java, работающий в распределенном режиме с
offset.storage.topic = connect-offsets. Тема заполняется при фиксации задач. Из моего класса задач (расширяет org.apache.kafka.connect.source.SourceTask) я могу прочитать эту тему, используя context.offsetStorageReader().
Есть ли способ прочитать смещение из моего класса соединителя?
При запуске или перезапуске соединителя я хочу иметь возможность использовать эту информацию, чтобы решить, как назначать разделы задачам. Обычно у меня есть небольшое количество «новых» разделов и большое количество «старых» разделов, я необходимость назначаю все новые разделы одной и той же задаче, но «старые» разделы могут быть назначены любой задаче. Знание того, существует ли запись для раздела, позволило бы мне выяснить, какие задачи являются «новыми» (в теме смещения нет записи), а какие «старыми» (в теме смещения есть запись).
Я не мог понять, как это сделать легко. В конце концов это стало спорным вопросом, потому что я изменил свой разъем, чтобы смещение не было частью решения о назначении раздела для задачи.
Вы можете сделать это, используя контекст (класс SourceTaskContext). Что-то вроде context.offsetStorageReader().offset(partition), это вернет вам карту, а затем вы можете передать ей свой ключ «смещения», чтобы получить смещение.