Есть ли способ получить текущее смещение в Kafka 0.10.x? Я не хочу использовать команды оболочки.Как получить текущее смещение в Kafka 0.10.x без оболочки?
Когда я использую API consumer.endOffsets(...), я могу получить последние смещения (logSize). Однако consumer.position(...) не получает меня текущего смещения!
В общем, я хочу получить текущее смещение, размер журнала и задержку в одном разделе.
Вы можете использовать KafkaConsumer#committed(), чтобы получить последнюю позицию. Таким образом, я отключу автоматическую фиксацию и выполняю ручные коммиты, вы можете вычислить точное отставание каждый раз, когда выполняете фиксацию.
С другой стороны, каждая запись, которую вы обрабатываете, обеспечивает ее смещение через ConsumerRecord#offset(), таким образом вы также можете вычислить задержку после прочтения одной записи (для раздела записи).
KafkaConsumer # commit() в 0.10.x любит OffsetFetchRequest в 0.8.x, я нашел, что OffsetFetchRequest также может получить то, что я хочу. И если я хочу знать время фиксации, есть только способ получить его, используя ConsumerRecord # timestamp()? – wxmimperio
'ConsumerRecord # timestamp()' возвращает встроенную метку времени записи, которая не связана с временем фиксации. Либо производитель может установить это значение временной отметки произвольно; или. если он не установлен, а политика добавления времени брокера включена, брокер устанавливает отметку времени для записи. Таким образом, это отметка * write * и не связана с потребительскими обязательствами. –
Как я могу получить время фиксации смещения? Я хочу написать последнее время фиксации для каждого раздела, какую структуру группы, тему и смещение для брокеров? В 0.8.x я могу получить структуру узлов zookeeper для них. Однако я не могу найти что-либо о фиксации для брокеров кафки. – wxmimperio
Нет доступных API-интерфейсов. Возможный способ - написать свои методы на основе кода ConsumerGroupCommand. – amethystic
Инструмент kafka даст вам текущее смещение. http://www.kafkatool.com/download.html – Paresh