java - 从kafka高级消费者获取偏移量

Question

我想从java程序中的高级消费者获取kafka消息偏移量。由于我使用自定义 commitoffset 属性，我想测试我的自定义 commitoffset 是否正常工作。谁能帮我如何抵消？？？ 我遇到过几个 kafka 工具(如 getoffsetshell)，但它对我的测试没有帮助。

Answer 1

当从 ConsumerIterator 获取消息时，您还可以通过执行以下操作来获取偏移量:

    ConsumerConnector consumerConnector = Consumer.createJavaConsumerConnector(getConsumerConfig());
    KafkaStream<byte[], byte[]> stream = getKafkaStream(consumerConnector);
    ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
    while(iterator.hasNext()) {
        MessageAndMetadata<byte[], byte[]> messageAndMetadata = iterator.next();
        String message = new String(messageAndMetadata.message());
        long offset = messageAndMetadata.offset();
    }