docs: def _process_message(self, msg):
          """De-serialize the message and execute the job.
  
          :param msg: Kafka message.
          :type msg: :doc:`kq.Message <message>`
          """
          self._logger.info(
              'Processing Message(topic={}, partition={}, offset={}) ...'
              .format(msg.topic, msg.partition, msg.offset))
          try:
              job = self._deserializer(msg.value)
              job_repr = get_call_repr(job.func, *job.args, **job.kwargs)
  
          except Exception as err:
              self._logger.exception('Job was invalid: {}'.format(err))
              self._execute_callback('invalid', msg, None, None, None, None)
          else:
              self._logger.info('Executing job {}: {}'.format(job.id, job_repr))
  
              if job.timeout:
                  timer = threading.Timer(job.timeout, _thread.interrupt_main)
                  timer.start()
              else:
                  timer = None
              try:
                  res = job.func(*job.args, **job.kwargs)
              except KeyboardInterrupt:
                  self._logger.error(
                      'Job {} timed out or was interrupted'.format(job.id))
                  self._execute_callback('timeout', msg, job, None, None, None)
              except Exception as err:
                  self._logger.exception(
                      'Job {} raised an exception:'.format(job.id))
                  tb = traceback.format_exc()
                  self._execute_callback('failure', msg, job, None, err, tb)
              else:
                  self._logger.info('Job {} returned: {}'.format(job.id, res))
                  self._execute_callback('success', msg, job, res, None, None)
              finally:
                  if timer is not None:
                      timer.cancel()