title: Confluent Kafka: Lag in Kafka Consumer when streaming a webcam feed
body: <p>I am working on Streaming my webcam feed to kakfa consumer to do some facial sentiment analysis and store it to DB. </p>
  
  <p>However, I am noticing a considerable lag in Consumer side (3 Sec). In the process to resolve that lag, reduced the message/image quality on the producer side. However, this didn't improve the latency and further it deteriorated the image quality on consumer side.</p>
  
  <p>Below is the code on Kafka Producer:</p>
  
  <pre><code>from confluent_kafka import Producer
  import cv2
  import time, sys
  
  p = Producer({'bootstrap.servers': 'localhost'})
  
  encode_param=[int(cv2.IMWRITE_JPEG_QUALITY),90]
  
  def delivery_report(err, msg):
      """ Called once for each message produced to indicate delivery result.
          Triggered by poll() or flush(). """
      if err is not None:
          print('Message delivery failed: {}'.format(err))
      else:
          print('Message delivered to {} [{}]'.format(msg.topic(), msg.partition()))
  
  camera = cv2.VideoCapture(0)
  
  try:
      while (True):
          success, frame = camera.read()
          frame = cv2.resize(frame, None, fx=0.3, fy=0.3)
          result, buffer = cv2.imencode('.jpg', frame, encode_param)
          #ret, buffer = cv2.imencode('.jpg', frame)
  
  
          p.poll(0)
              p.produce('newtopic', buffer.tobytes(), callback=delivery_report)
  
          print('sent')
          p.flush()
          time.sleep(0.1)
  except:
      print("\nExiting.")
      sys.exit(1)
  </code></pre>
  
  <p>Below is the code on Kafka Consumer:</p>
  
  <pre><code>import cv2
  import sys
  import os
  import imutils
  import logging
  
  from confluent_kafka import Consumer, KafkaError
  from confluent_kafka.admin import AdminClient
  
  
  def example_delete_topics(a, topics):
          fs = a.delete_topics(topics, operation_timeout=30)
  
          # Wait for operation to finish.
          for topic, f in fs.items():
              try:
                  f.result()  # The result itself is None
                  print("Topic {} deleted".format(topic))
              except Exception as e:
                  print("Failed to delete topic {}: {}".format(topic, e))
  a = AdminClient({'bootstrap.servers': 'localhost'})
  example_delete_topics(a, ['newtopic'])
  c = Consumer({
      'bootstrap.servers': 'localhost',
      'group.id': 'mygroup',
      'auto.offset.reset': 'latest'
  })
  camID = 0
  c.subscribe(['newtopic'])
  (H, W) = (None, None)
  
  while True:
      msg = c.poll(1.0)
  
      if msg is None:
          print ("No Message")
          continue
      if msg.error():
          print("Consumer error: {}".format(msg.error()))
          continue
      print ("Received")
      img_bytes = msg.value()
      frame = np.array(Image.open(io.BytesIO(img_bytes)))
      frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
      frame = imutils.resize(frame, width=800)
  
      if W is None or H is None:
          (H, W) = frame.shape[:2]
      #print (H,W)
      try:
          rects, landmarks = face_detect.detect_face(frame,20)
          rects_tracker = []
  
          for (i,rect) in enumerate(rects):
  
              # Intensive calculations on sentiment analysis
  
              draw_border(frame,(rect[0],rect[1]),(rect[0] + rect[2],rect[1]+rect[3]),(127,255,255),1,10,20)
              cv2.putText(frame,final_result,(rect[0],rect[1]),cv2.FONT_HERSHEY_SIMPLEX,0.5,(127,255,255),1,cv2.LINE_AA)
  
      except Exception as e :
          logging.info('Error on line {}'.format(sys.exc_info()[-1].tb_lineno), type(e).__name__, e)
          pass
  
  
      cv2.imshow('Sample', frame)
      key = cv2.waitKey(20)
      if key == 27:  # exit on ESC
          end = time.time()
          seconds = end - start
          fps  = frame_counter / seconds;
          print ("Estimated frames per second : {0}".format(fps))
                      break
  
  c.close()
  </code></pre>
  
  <p>Thoough, the code on consumer side is heavy with lot of steps. On a direct webcam without Kafka each frame takes 150 ms seconds to run. But with Kafka, it takes about 3 seconds. </p>
  
  <p>Could someone help on how can I reduce it to a great extent to make near real-time</p>