docs: 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