从 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();
    }