code: def start(self, start_offset):
          
          # Have we been started already, and not stopped?
          if self._start_d is not None:
              raise RestartError("Start called on already-started consumer")
  
          # Keep track of state for debugging
          self._state = '[started]'
  
          # Create and return a deferred for alerting on errors/stoppage
          start_d = self._start_d = Deferred()
  
          # Start a new fetch request, possibly just for the starting offset
          self._fetch_offset = start_offset
          self._do_fetch()
  
          # Set up the auto-commit timer, if needed
          if self.consumer_group and self.auto_commit_every_s:
              self._commit_looper = LoopingCall(self._auto_commit)
              self._commit_looper.clock = self.client.reactor
              self._commit_looper_d = self._commit_looper.start(
                  self.auto_commit_every_s, now=False)
              self._commit_looper_d.addCallbacks(self._commit_timer_stopped,
                                                 self._commit_timer_failed)
          return start_d
docs: Starts fetching messages from Kafka and delivering them to the
          :attr:`.processor` function.
  
          :param int start_offset:
              The offset within the partition from which to start fetching.
              Special values include: :const:`OFFSET_EARLIEST`,
              :const:`OFFSET_LATEST`, and :const:`OFFSET_COMMITTED`. If the
              supplied offset is :const:`OFFSET_EARLIEST` or
              :const:`OFFSET_LATEST` the :class:`Consumer` will use the
              OffsetRequest Kafka API to retrieve the actual offset used for
              fetching. In the case :const:`OFFSET_COMMITTED` is used,
              `commit_policy` MUST be set on the Consumer, and the Consumer
              will use the OffsetFetchRequest Kafka API to retrieve the actual
              offset used for fetching.
  
          :returns:
              A :class:`~twisted.internet.defer.Deferred` which will resolve
              successfully when the consumer is cleanly stopped, or with
              a failure if the :class:`Consumer` encounters an error from which
              it is unable to recover.
  
          :raises: :exc:`RestartError` if already running.