-
Notifications
You must be signed in to change notification settings - Fork 78
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
I cannot get messages using clj-kafka.consumer.zk #74
Comments
I think the docs are just sucky here. (core/with-resource [c (zk-consumer/consumer consumer-config)]
zk-consumer/shutdown
(take 2 (zk-consumer/messages c "test"))) What happens is that this code will open a consumer, create a lazy seq of it's messages, and then immediately close the consumer and return the seq. However since the seq is lazy, once you start reading, the consumer has already been closed and thus has no new messages for it. To return the actual messages you have to force the seq's evaluation within the with-resource. (core/with-resource [c (zk-consumer/consumer consumer-config)]
zk-consumer/shutdown
(doall (take 2 (zk-consumer/messages c "test")))) |
@j-pb Oh great ! Thank you ! (def x
(core/with-resource [c (zk/consumer config)]
zk/shutdown
(doall (take 1 (zk/messages c "test"))))
)
;; #'clj-kafka-example.core/x
(realized? x)
;; true
x
;;(#clj_kafka.core.KafkaMessage{:topic "test", :offset 0, :partition 0, :key nil, :value #object["[B" 0x6bbcc481 "[B@6bbcc481"]})
(def x
(core/with-resource [c (zk/consumer config)]
zk/shutdown
(take 1 (zk/messages c "test"))
)
)
;; #'clj-kafka-example.core/x
(realized? x)
;; false
x
;;() I have another question. When I called a consumer 3 times, it returned the same message. (core/with-resource [c (zk/consumer config)]
zk/shutdown
(doall (take 1 (zk/messages c "test"))))
;; (#clj_kafka.core.KafkaMessage{:topic "test", :offset 0, :partition 0, :key nil, :value #object["[B" 0x6d1531d "[B@6d1531d"]})
(core/with-resource [c (zk/consumer config)]
zk/shutdown
(doall (take 1 (zk/messages c "test"))))
;; (#clj_kafka.core.KafkaMessage{:topic "test", :offset 0, :partition 0, :key nil, :value #object["[B" 0x2cdb87c2 "[B@2cdb87c2"]})
(core/with-resource [c (zk/consumer config)]
zk/shutdown
(doall (take 1 (zk/messages c "test"))))
;; (#clj_kafka.core.KafkaMessage{:topic "test", :offset 0, :partition 0, :key nil, :value #object["[B" 0x1b90d67c "[B@1b90d67c"]}) Another problem is it worked well to take messages less than messages in a topic, however it hanged Assume there are 10 messages and it worked to take 3 messages from the topic but it hanged to try to (core/with-resource [c (zk/consumer config)]
zk/shutdown
(doall (take 3 (zk/messages c "test"))))
;; (#clj_kafka.core.KafkaMessage{:topic "test", :offset 0, :partition 0, :key nil, :value #object["[B" 0x45df4b8d "[B@45df4b8d"]}
;; #clj_kafka.core.KafkaMessage{:topic "test", :offset 1, :partition 0, :key nil, :value #object["[B" 0x645c0a87 "[B@645c0a87"]}
;; #clj_kafka.core.KafkaMessage{:topic "test", :offset 2, :partition 0, :key nil, :value #object["[B" 0x4b45519a "[B@4b45519a"]}) (core/with-resource [c (zk/consumer config)]
zk/shutdown
(doall (take 11 (zk/messages c "test")))) The above case, it seemed waiting for a next message but it didn't exit after produce a new message on the topic(looked hanged). Thanks, |
I tried to get kafka messages using clj-kafka.consumer.zk but I cannot do it yet.
The producer code is as follows:
The below is the consumer code:
The last consumer expression returns (). I am expecting the result is ("this is my message" "this is my message").
Using kafka bundled kafka-console-consumer.sh can retrieve messages correctly.
$ bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning this is my message this is my message
I am using kafka_2.10-0.8.2.1. Is there something wrong in my environment ? Or is there any way to solve this problem ?
The text was updated successfully, but these errors were encountered: