Can't Consume JSON Messages From Kafka Using Kafka-Python's Deserializer -


i trying send simple json object through kafka , read out other side using python , kafka-python. however, keep seeing following error:

2017-04-07 10:28:52,030.30.9998989105:kafka.future:8228:error:10620:error processing callback traceback (most recent call last):   file "c:\anaconda2\lib\site-packages\kafka\future.py", line 79, in _call_backs     f(value)   file "c:\anaconda2\lib\site-packages\kafka\consumer\fetcher.py", line 760, in _handle_fetch_response     unpacked = list(self._unpack_message_set(tp, messages))   file "c:\anaconda2\lib\site-packages\kafka\consumer\fetcher.py", line 539, in _unpack_message_set     tp.topic, msg.value)   file "c:\anaconda2\lib\site-packages\kafka\consumer\fetcher.py", line 570, in _deserialize     return f(bytes_)   file "c:\users\myuser\workspace\pythonkafkatest\src\example.py", line 55, in <lambda>     value_deserializer=lambda m: json.loads(m).decode('utf-8'))   file "c:\anaconda2\lib\json\__init__.py", line 339, in loads     return _default_decoder.decode(s)   file "c:\anaconda2\lib\json\decoder.py", line 364, in decode     obj, end = self.raw_decode(s, idx=_w(s, 0).end())   file "c:\anaconda2\lib\json\decoder.py", line 382, in raw_decode     raise valueerror("no json object decoded") valueerror: no json object decoded

i’ve done research , common cause of error json wrong. have tried printing out json before send adding following code , json prints no errors. 

  while true:         json_obj1 = json.dumps({"dataobjectid": "test1"})         print json_obj1         producer.send('my-topic', {"dataobjectid": "test1"})         producer.send('my-topic', {"dataobjectid": "test2"})         time.sleep(1)

this leads me suspect can produce json, not consume it.

here code:

import threading import logging import time import json  kafka import kafkaconsumer, kafkaproducer   class producer(threading.thread):     daemon = true      def run(self):         producer = kafkaproducer(bootstrap_servers='localhost:9092',                                  value_serializer=lambda v: json.dumps(v).encode('utf-8'))          while true:             producer.send('my-topic', {"dataobjectid": "test1"})             producer.send('my-topic', {"dataobjectid": "test2"})             time.sleep(1)   class consumer(threading.thread):     daemon = true      def run(self):         consumer = kafkaconsumer(bootstrap_servers='localhost:9092',                                  auto_offset_reset='earliest',                                  value_deserializer=lambda m: json.loads(m).decode('utf-8'))         consumer.subscribe(['my-topic'])          message in consumer:             print (message)   def main():     threads = [         producer(),         consumer()     ]      t in threads:         t.start()      time.sleep(10)  if __name__ == "__main__":     logging.basicconfig(         format='%(asctime)s.%(msecs)s:%(name)s:%(thread)d:' +                '%(levelname)s:%(process)d:%(message)s',         level=logging.info     )     main()

i can send , receive strings if remove value_serializer , value_deserializer. when run code can see json sending in. here short snipit:

consumerrecord(topic=u'my-topic', partition=0, offset=5742, timestamp=none, timestamp_type=none, key=none, value='{"dataobjectid": "test1"}', checksum=-1301891455, serialized_key_size=-1, serialized_value_size=25) consumerrecord(topic=u'my-topic', partition=0, offset=5743, timestamp=none, timestamp_type=none, key=none, value='{"dataobjectid": "test2"}', checksum=-1340077864, serialized_key_size=-1, serialized_value_size=25) consumerrecord(topic=u'my-topic', partition=0, offset=5744, timestamp=none, timestamp_type=none, key=none, value='test', checksum=1495943047, serialized_key_size=-1, serialized_value_size=4) consumerrecord(topic=u'my-topic', partition=0, offset=5745, timestamp=none, timestamp_type=none, key=none, value='\xc2hello, stranger!', checksum=-1090450220, serialized_key_size=-1, serialized_value_size=17) consumerrecord(topic=u'my-topic', partition=0, offset=5746, timestamp=none, timestamp_type=none, key=none, value='test', checksum=1495943047, serialized_key_size=-1, serialized_value_size=4) consumerrecord(topic=u'my-topic', partition=0, offset=5747, timestamp=none, timestamp_type=none, key=none, value='\xc2hello, stranger!', checksum=-1090450220, serialized_key_size=-1, serialized_value_size=17)

so tried removing value_deserializer consumer, , code executes without deserializer message comes out string, isn't need. so, why doesn't value_deserializer work? there different way json kafka message should using?

my problem solved after decoding message first utf-8, , json.load/dump it:

value_deserializer=lambda m: json.loads(m.decode('utf-8'))

instead of:

value_deserializer=lambda m: json.loads(m).decode('utf-8')

hope work producer's side


Comments

Post a Comment

Popular posts from this blog

How to understand 2 main() functions after using uftrace to profile the C++ program? -

i trying uftrace profile following simple c++ program: #include <iostream> class { public: a() {std::cout << "a created" << std::endl;} ~a() {std::cout << "a destroyed" << std::endl;} }; int main() { a; return 0; } the profile result this: # uftrace a.out created destroyed # duration tid function 2.026 [ 4828] | __cxa_atexit(); [ 4828] | main() { [ 4828] | __static_initialization_and_destruction_0() { 89.397 [ 4828] | std::ios_base::init::init(); 0.768 [ 4828] | __cxa_atexit(); 93.029 [ 4828] | } /* __static_initialization_and_destruction_0 */ 94.425 [ 4828] | } /* main */ [ 4828] | main() { [ 4828] | a::a() { 11.104 [ 4828] | std::operator <<(); 10.825 [ 4828] | std::basic_ostream::operator <<(); 24.514 [ 4828] | } /* a::a */ [ 4828] | a::~a() { 0.978 [ 4828] | ...
Read more

c# - Update a combobox from a presenter (MVP) -

i using mvp in project, new in mvp. have 2 comboboxes. when select option in combobox, combobox should filled new data. this action in presenter. view 'view1' in presenter, , introduced combobox1 , combobox2 properties in 'view1', because need 'datasource', 'displaymember', 'valuemember' , 'refresh()' in method below. but, when using pattern, enough send property like public string combobox2 { { return combobox1.selectedvalue.tosstring(); } } into presenter not whole combobox. how can solve problem? public void onselectedindexchangedcombobox1() { if (view1.combobox1.selectedindex == -1) { return; } datatable dt = tools.getdatatable("a path"); var query = (from o in dt.asenumerable() o.field<string>("afield") == farmerview.combobox1.selectedvalue.tostring() orderby o.field<string>("anotherfield") select...
Read more

How to put a lock and transaction on table using spring 4 or above using jdbcTemplate and annotations like @Transactional? -

i trying put lock on table while writing on table , if happened in between roll-back . trying convert below code lock table test_g1 read; lock table test_g write; -- begin; start transaction; insert test_g1 values(143); insert test_g values(145); select * test_g1; select * test_g; rollback; select * test_g; unlock tables; how convert above code @transactional spring jdbctemplate code? @transactional(rollbackfor={dataaccessexception.class}) public void test(){ jdbctemplate.execute("insert test1 (id, nam) values (4, 'a')"); throw new dataaccessexception("error") { }; } here trying throw error, insert statement should rollback it's not happening . thanks edit-1 attaching code , doing in jdbcdaoimpl.java , mentioned problem comment above test() . app.java package com.cgiri.javabrains.spring4; import org.springframework.beans.factory.annotation.autowired; import org.springframework.context.applic...
Read more