как настроить кафку подключить auto.offset.reset с остальным api

Question

У меня есть создание приемника kafka connect, который конвертирует данные в другое хранилище;Я хочу установить auto.offset.reset как latest при создании нового соединителя с kafka connect rest api;Я установил consumer.auto.offset.reset: latest в конфигах;

json { "name": "test_v14", "config": { "name": "test_v14", "consumer.auto.offset.reset": "latest", "connector.class": "...", ... } }

Но когда задача началась, потребитель kafka все еще опрашивает записи с самого начала;Как и любые другие способы установить auto.offset.reset как последний;

Answer 1

Начиная с Apache Kafka 2.3, теперь это можно установить как часть конфигурации соединителя.

На рабочий установить:

connector.client.config.override.policy=All

Затем в разъеме вы можете указать

"consumer.override.auto.offset.reset": "latest"

См. Это для более подробной информации: https://rmoff.net/2019/08/09/starting-a-kafka-connect-sink-connector-at-the-end-of-a-topic/

Answer 2

consumer.auto.offset.reset необходимо установить в файле connect-distributed*.properties (рабочий).

Его нельзя применить к какому-либо конкретному соединителю, если только этот класс соединителя явно не создает и не загружает свои собственные объекты-получатели, которые читают это свойство.