def _process_messages(self):
        """Processes messages received from kafka"""
        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")
Processes messages received from kafka