title: How to send and consume json messages using confluent-kafka in Python
body: <p>I am fairly new to Python and getting started with Kafka. So I have setup a Kafka broker and I am trying to communicate with it using <a href="https://github.com/confluentinc/confluent-kafka-python" rel="noreferrer">confluent-kafka</a>. 
  I have been able to produce and consume simple messages using it, however, I have some django objects which I need to serialize and send it ti kafka.</p>
  
  <p>Previously I was using <a href="https://github.com/dpkp/kafka-python" rel="noreferrer">kafka-python</a>, on which I was able to send and consume json messages, however I was having some weird issues it it.</p>
  
  <p><strong>#Producer.py</strong></p>
  
  <pre><code>def send_message(topic,message) :
  try :
      try :
          p.produce(topic,message,callback=delivery_callback)
      except BufferError as b :
          sys.stderr.write('%% Local producer queue is full (%d messages awaiting delivery): try again\n' %len(p))
      # Serve delivery callback queue.
      # NOTE: Since produce() is an asynchronous API this poll() call
      #       will most likely not serve the delivery callback for the
      #       last produce()d message.
      p.poll(0)
      # Wait until all messages have been delivered
      sys.stderr.write('%% Waiting for %d deliveries\n' % len(p))
      p.flush()
  except Exception as e :
      import traceback
      print(traceback.format_exc())
  </code></pre>
  
  <p><strong>#Consumer.py</strong></p>
  
  <pre><code>conf = {'bootstrap.servers': "localhost:9092", 'group.id': 'test', 'session.timeout.ms': 6000,
          'auto.offset.reset': 'earliest'}
  c = Consumer(conf)
  c.subscribe(["mykafka"])
  try:
      while True:
          msg = c.poll(timeout=1.0)
          if msg is None:
              continue
          if msg.error():
              raise KafkaException(msg.error())
          else:
              sys.stderr.write('%% %s [%d] at offset %d with key %s:\n' %
                                  (msg.topic(), msg.partition(), msg.offset(),
                                  str(msg.key())))
              print(msg.value())
  except Exception as e:
      import traceback
      print(traceback.format_exc())
  finally:
      c.close()
  </code></pre>
  
  <p>I serialize my django model objects like this :</p>
  
  <pre><code>from django.core import serializers
  # assuming obj is a model instance
  serialized_obj = serializers.serialize('json', [ obj, ])
  </code></pre>
  
  <p>So what change do I need to make in my producer and consumer to produce and cosume json messages?</p>