fetch-consumer-offsets returning error about clj-kafka

[pingles]

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
#75.

[rakeshp]

Kafka: kafka_2.11-0.8.2.2
Zookeeper: zookeeper-3.4.6
clj-kafka: 0.3.4

[finalfantasia]

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.