code: def _process_messages(self):
          
          try:
              for message in self.consumer:
                  try:
                      if message is None:
                          self.logger.debug("no message")
                          break
                      loaded_dict = json.loads(message.value)
                      self.logger.debug("got valid kafka message")
  
                      with self.uuids_lock:
                          if 'uuid' in loaded_dict:
                              if loaded_dict['uuid'] in self.uuids and \
                                      self.uuids[loaded_dict['uuid']] != 'poll':
                                  self.logger.debug("Found Kafka message from request")
                                  self.uuids[loaded_dict['uuid']] = loaded_dict
                              else:
                                  self.logger.debug("Got poll result")
                                  self._send_result_to_redis(loaded_dict)
                          else:
                              self.logger.debug("Got message not intended for this process")
                  except ValueError:
                      extras = {}
                      if message is not None:
                              extras["data"] = message.value
                      self.logger.warning('Unparseable JSON Received from kafka',
                                                  extra=extras)
  
              self._check_kafka_disconnect()
  
          except OffsetOutOfRangeError:
              # consumer has no idea where they are
              self.consumer.seek_to_end()
              self.logger.error("Kafka offset out of range error")
docs: Processes messages received from kafka