title: Kafka-python How to consume json message
body: <p>I am a fairly new in Python and starting with Kafka. I have a requirement where I need to send and consume json messages. For this I am using <a href="https://kafka-python.readthedocs.io/en/master/usage.html" rel="noreferrer">kafka-python</a> to communicate with Kafka. </p>
  
  <pre><code>#Producer.py
  from kafka import KafkaProducer
  import json
  producer = KafkaProducer(bootstrap_servers='localhost:9092',value_serializer=lambda v: json.dumps(v).encode('utf-8'))
  producer.send('offering_new', {"dataObjectID": "test1"})
  
  #Consumer.py
  import json
  from kafka import KafkaConsumer
  consumer = KafkaConsumer(bootstrap_servers='localhost:9092',auto_offset_reset='earliest', value_deserializer=lambda m: json.loads(m.decode('utf-8')))
  consumer.subscribe(['offering_new'])
  for message in consumer :
      print(message)
  </code></pre>
  
  <p>However I'm getting the following exception on consumer :</p>
  
  <pre><code>Traceback (most recent call last):
    File "&lt;stdin&gt;", line 1, in &lt;module&gt;
    File "/home/paras/.local/lib/python3.6/site-packages/kafka/consumer/group.py", line 1111, in __next__
      return next(self._iterator)
    File "/home/paras/.local/lib/python3.6/site-packages/kafka/consumer/group.py", line 1082, in _message_generator
      for msg in self._fetcher:
    File "/home/paras/.local/lib/python3.6/site-packages/kafka/consumer/fetcher.py", line 482, in __next__
      return next(self._iterator)
    File "/home/paras/.local/lib/python3.6/site-packages/kafka/consumer/fetcher.py", line 388, in _message_generator
      self._next_partition_records = self._parse_fetched_data(completion)
    File "/home/paras/.local/lib/python3.6/site-packages/kafka/consumer/fetcher.py", line 799, in _parse_fetched_data
      unpacked = list(self._unpack_message_set(tp, records))
    File "/home/paras/.local/lib/python3.6/site-packages/kafka/consumer/fetcher.py", line 458, in _unpack_message_set
      tp.topic, record.value)
    File "/home/paras/.local/lib/python3.6/site-packages/kafka/consumer/fetcher.py", line 492, in _deserialize
      return f(bytes_)
    File "&lt;stdin&gt;", line 1, in &lt;lambda&gt;
    File "/usr/lib/python3.6/json/__init__.py", line 354, in loads
      return _default_decoder.decode(s)
    File "/usr/lib/python3.6/json/decoder.py", line 339, in decode
      obj, end = self.raw_decode(s, idx=_w(s, 0).end())
    File "/usr/lib/python3.6/json/decoder.py", line 357, in raw_decode
      raise JSONDecodeError("Expecting value", s, err.value) from None
  json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
  </code></pre>
  
  <p>I am running the above code in Python Shell. Can someone tell me where I am going wrong?</p>