def start(self, start_offset):
        """
        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.
        """
        # 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