def receive(self):
        '''
        Return the message received and the address.
        '''
        try:
            msg = next(self.consumer)
        except ValueError as error:
            log.error('Received kafka error: %s', error, exc_info=True)
            raise ListenerException(error)
        log_source = msg.key
        try:
            decoded = json.loads(msg.value.decode('utf-8'))
        except ValueError:
            log.error('Not in json format: %s', msg.value.decode('utf-8'))
            return '', ''
        log_message = decoded.get('message')
        log.debug('[%s] Received %s from %s', log_message, log_source, time.time())
        return log_message, log_source
Return the message received and the address.