pingles
fetch-consumer-offsets returning error
Started zookeeper on port 2181 Started kafka broker on 9092
Inserted few messages using console producer in test topic
bin/kafka-console-producer.sh --topic test --broker-list localhost:9092
Consumed messages using console consumer in test topic
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test
Running kafka offset checker script returns the following output
bin/kafka-consumer-offset-checker.sh --zookeeper localhost:2181 --topic test --group console-consumer-54827
Group Topic Pid Offset logSize Lag Owner
console-consumer-54827 test 0 17 17 0 console-consumer-54827_xxx
Executing fetch-consumer-offsets function is returning the following error
(fetch-consumer-offsets "localhost:9092"
{"zookeeper.connect" "localhost:2181"}
"test"
"console-consumer-54827")
{"test:0" #object[kafka.common.OffsetMetadataAndError 0x4d7f1000 "OffsetMetadataAndError[-1,,0]"]} I am not able to figure out what wrong I am doing. Can someone please help me.
Could you confirm the versions/releases of Kafka, ZooKeeper and the clj-kafka lib you're using please.
Thanks, Paul
On Fri, Jan 8, 2016 at 1:58 PM, Rakesh [email protected] wrote:
Started zookeeper on port 2181 Started kafka broker on 9092
Inserted few messages using console producer in test topic
bin/kafka-console-producersh --topic test --broker-list localhost:9092
Consumed messages using console consumer in test topic
bin/kafka-console-consumersh --zookeeper localhost:2181 --topic test
Running kafka offset checker script returns the following output
bin/kafka-consumer-offset-checkersh --zookeeper localhost:2181 --topic test --group console-consumer-54827 Group Topic Pid Offset logSize Lag Owner console-consumer-54827 test 0 17 17 0 console-consumer-54827_xxx
Executing fetch-consumer-offsets function is returning the following error
(fetch-consumer-offsets "localhost:9092" {"zookeeperconnect" "localhost:2181"} "test" "console-consumer-54827")
{"test:0" #object[kafkacommonOffsetMetadataAndError 0x4d7f1000 "OffsetMetadataAndError[-1,,0]"]} I am not able to figure out what wrong I am doing Can someone please help me
— Reply to this email directly or view it on GitHub https://github.com/pingles/clj-kafka/issues/75.
I ran into this problem yesterday and after a round of digging in the Clojure source code (specifically, in fn clj-kafka.offset/fetch-consumer-offsets, I found the culprit.
It is the versionId of kafka.api.OffsetFetchRequest that determines where to fetch consumer offsets from. Quoting this guide to the Apache Kafka protocol:
There is no format difference between Offset Fetch Request v0 and v1. Functionality wise, Offset Fetch Request v0 will fetch offset from zookeeper, Offset Fetch Request v1 will fetch offset from Kafka.
So when an instance of kafka.api.OffsetFetchRequest is created in clj-kafka.offset/offset-fetch-request, versionId is hard-coded to 1, that causes consumer offsets to be fetched from Kafka, which simply returns consumer offsets of -1 as response since in version 0.8.x, consumer offsets are stored in ZooKeeper, instead of Kafka. In order to get the correct consumer offsets from ZooKeper, versionId needs to be set to 0. The following is a little function that I wrote based on clj-kafka.offset/offset-fetch-request that fetches consumer offsets from ZooKeeper:
(defn offset-fetch-request [group-id topic max-partition]
(let [offset-request-version 0 ; version 0 fetches from ZooKeeper, version 1 and above fetch from Kafka.
correlation-id 1
topic-partition-java (map (fn [i] (kafka.common.TopicAndPartition. topic i)) (range 0 max-partition))
topic-partition-scala (.toSeq (scala.collection.JavaConversions/asScalaBuffer topic-partition-java))]
(kafka.api.OffsetFetchRequest. group-id topic-partition-scala offset-request-version correlation-id "clj-kafka-id")))
So one can use the function above to write another function that's similar to clj-kafka.offset/fetch-consumer-offsets.
