def _unpack_message_set(self, tp, messages):
        try:
            for offset, size, msg in messages:
                if self.config['check_crcs'] and not msg.validate_crc():
                    raise Errors.InvalidMessageError(msg)
                elif msg.is_compressed():
                    for record in self._unpack_message_set(tp, msg.decompress()):
                        yield record
                else:
                    key, value = self._deserialize(msg)
                    yield ConsumerRecord(tp.topic, tp.partition, offset, key, value)
        # If unpacking raises StopIteration, it is erroneously
        # caught by the generator. We want all exceptions to be raised
        # back to the user. See Issue 545
        except __HOLE__ as e:
            log.exception('StopIteration raised unpacking messageset: %s', e)
            raise Exception('StopIteration raised unpacking messageset')
dataset/ETHPy150Open dpkp/kafka-python/kafka/consumer/fetcher.py/Fetcher._unpack_message_set